class_docstr: An optimizer that wraps a _DistributedOptimizer, intercepting push-pull operations.
  This class enables overlapping gradient push-pull with both backward and forward propagation while maintaining
  correct dependencies. It can achieve even higher training performance than the default BytePS with proper system
  parameters. To understand the principles behind barrier crossing, check the paper
  https://dl.acm.org/citation.cfm?id=3359642
class_docstr_tok: ["An","optimizer","that","wraps","a","_DistributedOptimizer",",","intercepting","push-pull","operations",".","This","class","enables","overlapping","gradient","push-pull","with","both","backward","and","forward","propagation","while","maintaining","correct","dependencies",".","It","can","achieve","even","higher","training","performance","than","the","default","BytePS","with","proper","system","parameters",".","To","understand","the","principles","behind","barrier","crossing",",","check","the","paper","https",":","\/\/dl.acm.org\/citation.cfm","?","id=3359642"]
docstr: Poll the completion of the tensor's backward or push-pull from a FIFO event_queue
docstr_tok: ["Poll","the","completion","of","the","tensor","'s","backward","or","push-pull","from","a","FIFO","event_queue"]
code: def _poll(self):
      """Poll the completion of the tensor's backward or push-pull from a FIFO event_queue"""
      while True:
          p, handle, ctx = self._event_queue.get()
          if p is None:
              self._logger.debug("poller exits.")
              break
          # Check whether the push-pull is finished. If so, start updating parameters.
          if handle is not None and poll(handle):
              output = synchronize(handle)
              p.grad.set_(self._compression.decompress(output, ctx))
              self._logger.debug(
                  "{} {} finished push-pull".format(
                      self._desc, self._get_parameter_name(p)
                  )
              )
              self._push_pull_delay[p] = self.backward_passes_per_step
              # So only support SGD, Adam and RMSprop optimizers in torch
              if isinstance(self._opt, torch.optim.SGD):
                  self._sgd(p)
              elif isinstance(self._opt, torch.optim.Adam):
                  self._adam(p)
              elif isinstance(self._opt, torch.optim.RMSprop):
                  self._rmsprop(p)
              else:
                  raise ValueError(
                      "Invalid optimizer! Only support SGD, Adam and RMSprop."
                  )
              self._zero_one_grad(p)
              # notify update completion and parameter is ready for forward propagation
              if p in self._locks:
                  self._locks[p].release()
          else:
              self._event_queue.put((p, handle, ctx))
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  from byteps.torch.compression import Compression
  from byteps.torch.ops import push_pull_async_inplace
  from byteps.torch.ops import poll, synchronize
  from byteps.torch.ops import init, shutdown
  from byteps.torch.ops import size, local_size, rank, local_rank
  import threading
  import logging
  import time
  import math
  import torch
  import byteps.torch
label_desc: Use image node_id 9 for calling the _CrossBarrier obj's underlying member method code with example usage: obj._poll() without return types