U yh @sddlZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z m Z m Z z ddl Z Wnek r|YnXeeZddZd e edddZGdd d ZdS) N)OptionalTextIOUnionc Cstjddtjtjd}|D]x}|\}}}}}z,t|||}|d|d|WStk r}z|td|W5d}~XYqXqt ddS)a Find 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)addrsaddrrr proto_serb/var/www/html/venv/lib/python3.8/site-packages/torch/distributed/elastic/rendezvous/etcd_server.pyfind_free_ports"   "rdata_dircCsL|r*|dkr*td|||rHtd|tj|dddS)Nzstopping etcd serverzdeleting etcd data dir: %sT ignore_errors)pollloggerinfo terminatewaitshutilrmtree) subprocessrrrr stop_etcdBs  r)c@seZdZdZdeedddZejdddZ e dd d Z edd d Z edd dZ de e ee edfddddZd ee ee edfddddZddZd!e ddddZddddZdS)" 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 sJd|_ |rR|n t j dd|_ d|_d|_dS)Nrzbin/etcdZTORCHELASTIC_ETCD_BINARY_PATHetcdZtorchelastic_etcd_data)prefix)_port_hostospathdirname__file__joinenvironget_etcd_binary_pathisfiletempfilemkdtemp_base_data_dirZ _etcd_cmd _etcd_proc)selfrrootZdefault_etcd_binrrr__init__ms zEtcdServer.__init__)returncCs|jstdn|jSdS)Nz>No etcd server process started. Call etcd_server.start() first)r<rr=rrr_get_etcd_server_processs z#EtcdServer._get_etcd_server_processcCs|jS)z)Return the port the server is running on.)r.rArrrget_portszEtcdServer.get_portcCs|jS)z)Return the host the server is running on.)r/rArrrget_hostszEtcdServer.get_hostcCs|jd|jS)z,Return the etcd server endpoint (host:port).:)r/r.rArrr get_endpointszEtcdServer.get_endpoint<)timeout num_retriesstderrr@c Csd}z2tj|jt|}tj|dd||||WStk r}z@|d7}t|j t dt|||krt j |jddW5d}~XYqXqtt|j |jdS)a Start 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. stderr: the standard error file handle. Valid values are `subprocess.PIPE`, `subprocess.DEVNULL`, an existing file descriptor (a positive integer), an existing file object, and `None`. Raises: TimeoutError: if the server is not ready within the specified timeout rT)exist_okz4Failed to start etcd server, got error: %s, retryingrN)r0r1r4r;strmakedirs_start Exceptionr)r<r"warningr&r'atexitregister)r=rIrJrKZ curr_retriesrrrrrstarts  zEtcdServer.start)rrIrKr@cCst}t}|d|_|d}td|jdd|dd|jd|jdd|jd|jd d|jd|g }t d || | t j |d |d |_ ||dS) NrM z --enable-v2z --data-dirz--listen-client-urlszhttp://rEz--advertise-client-urlsz--listen-peer-urlszStarting etcd server: [%s]T) close_fdsrK)r getsocknamer.shlexsplitr4r7r/r"r#rr(Popenr<_wait_for_ready)r=rrIrKsockZ sock_peerZ peer_portZetcd_cmdrrrrPs0  zEtcdServer._startcCstj|j|jdddS)zNReturn an etcd client object that can be used to make requests to this server./v2 rrversion_prefix read_timeout)r,Clientr/r.rArrr get_clients zEtcdServer.get_client)rIr@cCstj|j|jddd}t|}t|kr|dk rX|j}td|zt d|j WdSt k rt dYq$Xq$tddS)Nr^r`z*Etcd server process exited with the code: zetcd server ready. version: %srMz.Timed out waiting for etcd server to be ready!)r,rcr/r.timerBr! returncoderr"r#versionrQsleep TimeoutError)r=rIclientZmax_timeexitcoderrrr\s&   zEtcdServer._wait_for_readycCstdt|j|jdS)zGStop the server and cleans up auto generated resources (e.g. data dir).zEtcdServer stop method calledN)r"r#r)r<r;rArrrstops zEtcdServer.stop)N)rGrHN)rGN)rG)__name__ __module__ __qualname____doc__rrNr?r(r[rBintrCrDrFrrrUrPrdr\rmrrrrr*Ms4  (  r*)N)rSloggingr0rYr&r r(r9rftypingrrrr,ModuleNotFoundError getLoggerrnr"rrNr)r*rrrr s"  %