class_docstr: A rendezvous implementation that uses `etcd <https://etcd.io/>`__ as the backend store.
class_docstr_tok: ["A","rendezvous","implementation","that","uses","`","etcd","<","https",":","\/\/etcd.io\/",">","`","__","as","the","backend","store","."]
docstr: We observed a rendezvous state in 'joinable' state, and attempt to join this
  particular version, and then wait for all other peers to join.
docstr_tok: ["We","observed","a","rendezvous","state","in","'joinable","'","state",",","and","attempt","to","join","this","particular","version",",","and","then","wait","for","all","other","peers","to","join","."]
code: def join_phase(self, expected_version):
      """
      We observed a rendezvous state in 'joinable' state, and attempt to join this
      particular version, and then wait for all other peers to join.
      """
      # Failure to join will propagate an exception, causing a re-entry.
      active_version, this_rank = self.join_rendezvous(expected_version)
      state = json.loads(active_version.value)
      log.info(
          "Joined rendezvous version %s as rank %s. Full state: %s",
          state["version"],
          this_rank,
          state,
      )
  
      # If this worker was first to reach num_min_workers requirement,
      # and rendezvous is still joinable (therefore it is elastic),
      # then this worker will be responsible for waiting out the "last call"
      # timeout and closing (i.e. transitioning to 'frozen') the rendezvous
      # afterwards.
      # As a safety against a potential failure of this worker (during the
      # last call timeout), the rendezvous state is made ephemeral
      # when min_num_workers is reached.
  
      if (
          this_rank == self._num_min_workers - 1
          and state["status"] == "joinable"
      ):
          log.info(
              "Rank %s is responsible for join last call.", this_rank
          )
          last_call_deadline = time.time() + self._last_call_timeout
          self.handle_join_last_call(
              expected_version, last_call_deadline
          )
          log.info("Rank %s finished join last call.", this_rank)
  
      # Wait for rendezvous state to be frozen, which means a fixed set of peers
      log.info("Waiting for remaining peers.")
      active_version = self.wait_for_peers(expected_version)
      state = json.loads(active_version.value)
  
      assert (
          state["version"] == expected_version
      ), "Logic error: failed to observe version mismatch"
  
      return self.confirm_phase(expected_version, this_rank)
  
imports: import json
  import logging
  import sys
  import threading
  import time
  from typing import Optional
  import etcd
  from torch.distributed.elastic.rendezvous import RendezvousClosedError, RendezvousError, RendezvousHandler, RendezvousParameters, RendezvousTimeoutError
  from .utils import parse_rendezvous_endpoint
  from .etcd_store import EtcdStore, cas_delay
label_desc: Use image node_id 5 for calling the EtcdRendezvous obj's underlying member method code with example usage: obj.join_phase(expected_version) and returns: self