o wOi@sddlZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl Z ddl m Z e Z ddZd de efddZGd d d ZdS) N)Optional) get_loggerc Cstjddtjtjd}|D];}|\}}}}}zt|||}|d|d|WStyH}z|td|WYd}~q d}~wwt d)a Finds a free port and binds a temporary socket to it so that the port can be "reserved" until used. .. note:: the returned socket must be closed before using the port, otherwise a ``address already in use`` error will happen. The socket should be held and closed as close to the consumer of the port as possible since otherwise, there is a greater chance of race-condition where a different process may see the port as being free and take it. Returns: a socket binded to the reserved free port Usage:: sock = find_free_port() port = sock.getsockname()[1] sock.close() use_port(port) localhostN)hostportfamilytype)rrrz Socket creation attempt failed: zFailed to create a socket) socket getaddrinfo AF_UNSPEC SOCK_STREAMbindlistenOSErrorcloseprint RuntimeError)addrsaddrrrproto_serW/home/ubuntu/.local/lib/python3.10/site-packages/torchelastic/rendezvous/etcd_server.pyfind_free_ports     rdata_dircCsT|r|durtd|||r(td|tj|dddSdS)Nzstopping etcd serverzdeleting etcd data dir: T ignore_errors)pollloginfo terminatewaitshutilrmtree) subprocessrrrr stop_etcd?s r'c@seZdZdZddeefddZdefddZdefd d Z defd d Z ddededdfddZ ddededdfddZ de jfddZddeddfddZdddZdS) EtcdServera .. note:: tested on etcd server v3.4.3 Starts and stops a local standalone etcd server on a random free port. Useful for single node, multi-worker launches or testing, where a sidecar etcd server is more convenient than having to separately setup an etcd server. This class registers a termination handler to shutdown the etcd subprocess on exit. This termination handler is NOT a substitute for calling the ``stop()`` method. The following fallback mechanism is used to find the etcd binary: 1. Uses env var TORCHELASTIC_ETCD_BINARY_PATH 2. Uses ``/bin/etcd`` if one exists 3. Uses ``etcd`` from ``PATH`` Usage :: server = EtcdServer("/usr/bin/etcd", 2379, "/tmp/default.etcd") server.start() client = server.get_client() # use client server.stop() Args: etcd_binary_path: path of etcd server binary (see above for fallback path) NrcCspd|_d|_tjt}tj|d}tjd||_ tj |j s%d|_ |r)|nt j dd|_ d|_d|_dS)Nrzbin/etcdTORCHELASTIC_ETCD_BINARY_PATHetcdtorchelastic_etcd_data)prefix)_port_hostospathdirname__file__joinenvironget_etcd_binary_pathisfiletempfilemkdtemp_base_data_dir _etcd_cmd _etcd_proc)selfrrootdefault_etcd_binrrr__init__js  zEtcdServer.__init__returncC|jS)zI Returns: the port the server is running on. )r.r>rrrget_port|zEtcdServer.get_portcCrC)zI Returns: the host the server is running on. )r/rDrrrget_hostrFzEtcdServer.get_hostcCs|jd|jS)zK Returns: the etcd server endpoint (host:port) :)r/r.rDrrr get_endpointszEtcdServer.get_endpoint<timeout num_retriesc Csd} ztj|jt|}tj|dd|||WStyN}z'|d7}t|j t dt|d||krDt j |jddWYd}~nd}~wwq) a Starts the server, and waits for it to be ready. When this function returns the sever is ready to take requests. Args: timeout: time (in seconds) to wait for the server to be ready before giving up. num_retries: number of retries to start the server. Each retry will wait for max ``timeout`` before considering it as failed. Raises: TimeoutError: if the server is not ready within the specified timeout rT)exist_okz(Failed to start etcd server, got error: z , retryingrN)r0r1r4r;strmakedirs_start Exceptionr'r=r warningr$r%atexitregister)r>rLrM curr_retriesrrrrrstarts&  zEtcdServer.startcCst}t}|d|_|d}td|jdd|dd|jd|jdd|jd|jd d|jd|g }t d |d | | t j |d d |_ ||dS)NrO z --enable-v2z --data-dirz--listen-client-urlszhttp://rHz--advertise-client-urlsz--listen-peer-urlszStarting etcd server: []T) close_fds)r getsocknamer.shlexsplitr4r7r/r r!rr&Popenr=_wait_for_ready)r>rrLsock sock_peer peer_portetcd_cmdrrrrRs0 zEtcdServer._startcCstj|j|jdddS)zx Returns: An etcd client object that can be used to make requests to this server. /v2 rrversion_prefix read_timeout)r+Clientr/r.rDrrr get_clients zEtcdServer.get_clientcCstj|j|jddd}t|}t|krI|jdur(td|jjz t d|j WdSt yBt dYnwt|kstd)Nrergz*Etcd server process exited with the code: zetcd server ready. version: rOz.Timed out waiting for etcd server to be ready!)r+rjr/r.timer=rr returncoder r!versionrSsleep TimeoutError)r>rLclientmax_timerrrr`s"      zEtcdServer._wait_for_readycCstdt|j|jdS)zY Stops the server and cleans up auto generated resources (e.g. data dir) zEtcdServer stop method calledN)r r!r'r=r;rDrrrstops zEtcdServer.stopN)rJrK)rJ)rBN)__name__ __module__ __qualname____doc__rrPrAintrErGrIrXrRr+rjrkr`rtrrrrr(Js r(ru)rUr0r]r$r r&r9rmtypingrr+torchelastic.utils.loggingrr rrPr'r(rrrrs  &