class_docstr: Class for global accumulation state. add_stats gets reduced.
code: def reduce(self, stats):
      if self.reduce_future is not None and self.reduce_future.done():
          if self.reduce_future.exception() is not None:
              logging.info(
                  "global stats accumulation error: %s",
                  self.reduce_future.exception(),
              )
              self.enqueue_global_stats(self.sent_global_stats)
          else:
              self.add_stats(
                  self.global_stats, self.reduce_future.result()
              )
          self.reduce_future = None
  
      stats_diff = stats
      if self.prev_stats is not None:
          stats_diff = {
              k: v - self.prev_stats[k] for k, v in stats.items()
          }
  
      self.enqueue_global_stats(stats_diff)
      self.prev_stats = copy.deepcopy(stats)
  
      if self.reduce_future is None:
          # Only reduce when not currently reducing.
          # Otherwise, we keep queued_global_stats for next time.
          self.sent_global_stats = self.queued_global_stats
          self.queued_global_stats = None
          # Additional copy to deal with potential partial reductions.
          self.reduce_future = self.rpc_group.all_reduce(
              "global stats",
              copy.deepcopy(self.sent_global_stats),
              self.add_stats,
          )
  
imports: import copy
  import dataclasses
  import logging
  import torch
  import moolib
label_desc: Use image node_id 4 for calling the GlobalStatsAccumulator obj's underlying member method code with example usage: obj.reduce(stats) without return types