class_docstr: DistributedContext provides useful methods for effective distributed training.
  
  A DistributedContext has the following required args:
   - rank: the index of this worker in the entire job
   - size: the number of workers in the entire job
   - local_rank: the index of this worker on this machine
   - local_size: the number of workers on this machine
   - cross_rank: the index of this machine in the entire job
   - cross_size: the number of machines in the entire job
  
  Additionally, any time that cross_size > 1, you must also provide:
   - chief_ip: the ip address to reach the chief worker (where rank==0)
  
  .. note::
  
     DistributedContext has ``.allgather()``, ``.gather()``, and ``.broadcast()`` methods, which
     are easy to use and which can be useful for coordinating work across workers, but it is not a
     replacement for the allgather/gather/broadcast operations in your particular distributed
     training framework.
class_docstr_tok: ["DistributedContext","provides","useful","methods","for","effective","distributed","training",".","A","DistributedContext","has","the","following","required","args",":","-","rank",":","the","index","of","this","worker","in","the","entire","job","-","size",":","the","number","of","workers","in","the","entire","job","-","local_rank",":","the","index","of","this","worker","on","this","machine","-","local_size",":","the","number","of","workers","on","this","machine","-","cross_rank",":","the","index","of","this","machine","in","the","entire","job","-","cross_size",":","the","number","of","machines","in","the","entire","job","Additionally",",","any","time","that","cross_size",">","1",",","you","must","also","provide",":","-","chief_ip",":","the","ip","address","to","reach","the","chief","worker","(","where","rank==0",")","..","note",":",":","DistributedContext","has","``",".allgather","(",")","``",",","``",".gather","(",")","``",",","and","``",".broadcast","(",")","``","methods",",","which","are","easy","to","use","and","which","can","be","useful","for","coordinating","work","across","workers",",","but","it","is","not","a","replacement","for","the","allgather\/gather\/broadcast","operations","in","your","particular","distributed","training","framework","."]
code: def _init_ipc(self, force_tcp: bool) -> None:
      if self.size < 2:
          # No broadcasting necessary.
          return
  
      # Global broadcast server.
      if self._is_chief:
          logger.debug(
              f"Chief setting up server with ports {self._pub_port}/{self._pull_port}."
          )
          self._chief_zmq = ipc.ZMQBroadcastServer(
              num_connections=self.size - 1,
              pub_url=f"tcp://*:{self._pub_port}",
              pull_url=f"tcp://*:{self._pull_port}",
          )
          self._chief_zmq.safe_start()
  
      else:
          logger.debug(
              f"Non-Chief {self.rank} setting up comm to "
              f"{self._chief_ip} w/ ports "
              f"{self._pub_port}/{self._pull_port}."
          )
          self._worker_zmq = ipc.ZMQBroadcastClient(
              srv_pub_url=f"tcp://{self._chief_ip}:{self._pub_port}",
              srv_pull_url=f"tcp://{self._chief_ip}:{self._pull_port}",
          )
          self._worker_zmq.safe_start()
  
      # Local broadcast server.
      self.tempdir = None
      if self.local_size < 2:
          # If local size is less than 2, we don't need a local chief but still need to
          # participate in the global all gather, otherwise the other participants block forever.
          _ = self.allgather(None)
      elif self._is_local_chief:
          pub_url = None
          pull_url = None
          if hasattr(socket, "AF_UNIX") and not force_tcp:
              # On systems with unix sockets, we get a slight performance bump by using them.
              self.tempdir = tempfile.mkdtemp(prefix="ipc")
              pub_url = f"ipc://{self.tempdir}/pub.sock"
              pull_url = f"ipc://{self.tempdir}/pull.sock"
  
          logger.debug(
              f"Local Chief setting up server with urls {pub_url}/{pull_url}."
          )
          self._local_chief_zmq = ipc.ZMQBroadcastServer(
              num_connections=self.local_size - 1,
              pub_url=pub_url,
              pull_url=pull_url,
          )
  
          if pub_url is None:
              pub_url = f"tcp://localhost:{self._local_chief_zmq.get_pub_port()}"
  
          if pull_url is None:
              pull_url = f"tcp://localhost:{self._local_chief_zmq.get_pull_port()}"
  
          # Do a global allgather to initialize local clients on every node.
          local_chief = (self.cross_rank, pub_url, pull_url)
          _ = self.allgather(local_chief)
          self._local_chief_zmq.safe_start()
  
      else:
          # Start with the global allgather.
          all_local_chiefs = self.allgather(None)
          my_local_chief = [
              x
              for x in all_local_chiefs
              if x is not None and x[0] == self.cross_rank
          ]
          assert len(my_local_chief) == 1, (
              f"did not find exactly 1 local_chief for machine {self.cross_rank} "
              f"in {all_local_chiefs}"
          )
          _, pub_url, pull_url = my_local_chief[0]
  
          assert isinstance(pub_url, str), f"invalid pub_url: {pub_url}"
          assert isinstance(
              pull_url, str
          ), f"invalid pub_url: {pull_url}"
  
          logger.debug(
              f"Local Worker setting up server with urls {pub_url}/{pull_url}."
          )
          self._local_worker_zmq = ipc.ZMQBroadcastClient(
              pub_url, pull_url
          )
          self._local_worker_zmq.safe_start()
  
imports: import logging
  import os
  import socket
  import tempfile
  from typing import Any, List, Optional
  from determined import constants, ipc, util
label_desc: Use image node_id 2 for calling the DistributedContext obj's underlying member method code with example usage: obj._init_ipc(force_tcp) without return types