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: When there's an existing valid rendezvous in state 'final', we have to wait until the next opportunity to join.
  
  Such opportunity may come from:
  
  1. rendezvous state changed by someone else, in which case we unblock and retry.
  2. rendezvous becomes invalid because at least one member failed to renew their
     leased keep_alive node. We detect this, and destroy the rendezvous.
docstr_tok: ["When","there","'s","an","existing","valid","rendezvous","in","state","'final","'",",","we","have","to","wait","until","the","next","opportunity","to","join",".","Such","opportunity","may","come","from",":","1.","rendezvous","state","changed","by","someone","else",",","in","which","case","we","unblock","and","retry",".","2.","rendezvous","becomes","invalid","because","at","least","one","member","failed","to","renew","their","leased","keep_alive","node",".","We","detect","this",",","and","destroy","the","rendezvous","."]
code: def wait_for_rendezvous_to_free(self, expected_version):
      """
      When there's an existing valid rendezvous in state 'final', we have to wait until the next opportunity to join.
  
      Such opportunity may come from:
  
      1. rendezvous state changed by someone else, in which case we unblock and retry.
      2. rendezvous becomes invalid because at least one member failed to renew their
         leased keep_alive node. We detect this, and destroy the rendezvous.
      """
      active_version, state = self.get_rdzv_state()
      while True:
          if (
              state["status"] != "final"
              or state["version"] != expected_version
          ):
              return
  
          # Check if current rendezvous state is valid, in the sense that all
          # its members are alive (renewing their lease).
          # If not, try destroy this rendezvous, so a new one can be created.
          alive_members = self.client.get(
              self.get_path(f"/rdzv/v_{expected_version}")
          )
          keep_alive_keys = [ch.key for ch in alive_members.children]
  
          for key in state["keep_alives"]:
              if key not in keep_alive_keys:
                  # This participant didn't renew their lease. We'll declare this
                  # rendezvous version as dead (but only if it hadn't changed)
                  log.info("Keep-alive key %s is not renewed.", key)
                  log.info(
                      "Rendezvous version %s is incomplete. ",
                      expected_version,
                  )
                  log.info("Attempting to destroy it.")
  
                  # Compare-and-delete operation. Throws if compare failed,
                  # which means rendezvous was already destroyed/re-created/closed,
                  # and we can try to re-enter the barrier.
                  self.client.delete(
                      key=self.get_path("/rdzv/active_version"),
                      prevValue=active_version.value,
                  )
  
                  log.info(
                      "Destroyed rendezvous version %s successfully.",
                      expected_version,
                  )
  
                  # We can return (and retry) immediately
                  return
  
          # Existing rendezvous seems valid, no reason to destroy it.
          # We just have to wait until something changes and re-check.
          try:
              overall_timeout = (
                  max(self._rendezvous_deadline - time.time(), 0.0)
                  + 1.0
              )
              self.client.watch(
                  key=self.get_path("/rdzv"),
                  index=active_version.etcd_index + 1,
                  recursive=True,
                  timeout=overall_timeout,
              )
          except (etcd.EtcdEventIndexCleared, etcd.EtcdWatchTimedOut):
              pass
  
          if time.time() > self._rendezvous_deadline:
              raise RendezvousTimeoutError()
          active_version, state = self.get_rdzv_state()
  
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 14 for calling the EtcdRendezvous obj's underlying member method code with example usage: obj.wait_for_rendezvous_to_free(expected_version) without return types