o Me8!@stddlZddlZddlZddlZddlmZmZddlmZddl m Z gZ da ddZ ddZd d Zd d ZdS) N)ProcessManager)core)wait_server_readycCsfddlm}|t||d}|d}|dds|s-t||dds|r|dS)Nr)KVServer)sizerunningF) Z*paddle.distributed.fleet.utils.http_serverrintstartgetZ should_stoptimesleepstop)portZ http_server_drrZ http_serverZ wait_secondsrUD:\Projects\ConvertPro\env\Lib\site-packages\paddle/distributed/parallel_with_gloo.py_start_kv_servers   rcCs|dkdus Jdt}|}d|d<|dkr8d|i}ttt|dd||fd }d |_d |d<|t|gt }||_ ||_ |dd|_ t|dd|_d |_d |_t |at|dkrvd|d<|d Sd S)u/ Initialize parallel environment with gloo for cpu only. Args: - rank_id（int, required) - the index of current rank; - rank_num (int, required) - the number of ranks in this parallel env; - server_endpoint (str, required) - endpoint of server to init gloo context in ip:port format; Returns: None Examples: .. code-block:: python import paddle import multiprocessing from contextlib import closing import socket port_set = set() def find_free_port(): def _free_port(): with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s: s.bind(('', 0)) return s.getsockname()[1] while True: port = _free_port() if port not in port_set: port_set.add(port) return port def test_gloo_init(id, rank_num, server_endpoint): paddle.distributed.gloo_init_parallel_env( id, rank_num, server_endpoint) def test_gloo_init_with_multiprocess(num_of_ranks): jobs = [] server_endpoint = "127.0.0.1:%s" % (find_free_port()) for id in range(num_of_ranks): p = multiprocessing.Process( target=test_gloo_init, args=(id, num_of_ranks, server_endpoint)) jobs.append(p) p.start() for proc in jobs: proc.join() if __name__ == '__main__': # Arg: number of ranks (processes) test_gloo_init_with_multiprocess(2) FzRrank_num should greater than or equal to 2 for parallel environment initialzation.r rZ_worker:)targetargsTiiN)rdictrrr splitdaemonr rrZGlooParallelStrategyZrankrank_num ip_addressZip_portZ init_secondsZ run_secondsZGlooParallelContext_global_gloo_ctxinitjoin)Zrank_idrZserver_endpointmanagerZhttp_server_statusrZhttp_server_procZ gloo_strategyrrrgloo_init_parallel_env(s<7   r"cCstdusJdtdS)a Call barrier function with initialized gloo context. Args: None Returns: None Examples: .. code-block:: python import paddle import multiprocessing from contextlib import closing import socket port_set = set() def find_free_port(): def _free_port(): with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s: s.bind(('', 0)) return s.getsockname()[1] while True: port = _free_port() if port not in port_set: port_set.add(port) return port def test_gloo_barrier(id, rank_num, server_endpoint): paddle.distributed.gloo_init_parallel_env( id, rank_num, server_endpoint) paddle.distributed.gloo_barrier() def test_gloo_barrier_with_multiprocess(num_of_ranks): jobs = [] server_endpoint = "127.0.0.1:%s" % (find_free_port()) for id in range(num_of_ranks): p = multiprocessing.Process( target=test_gloo_barrier, args=(id, num_of_ranks, server_endpoint)) jobs.append(p) p.start() for proc in jobs: proc.join() if __name__ == '__main__': # Arg: number of ranks (processes) test_gloo_barrier_with_multiprocess(2) Nzgloo context is not initialzed.)rZbarrierrrrr gloo_barriers6 r#cCstdur tdSdS)a Release the parallel environment initialized by gloo Args: None Returns: None Examples: .. code-block:: python import paddle import multiprocessing from contextlib import closing import socket port_set = set() def find_free_port(): def _free_port(): with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s: s.bind(('', 0)) return s.getsockname()[1] while True: port = _free_port() if port not in port_set: port_set.add(port) return port def test_gloo_release(id, rank_num, server_endpoint): paddle.distributed.gloo_init_parallel_env( id, rank_num, server_endpoint) paddle.distributed.gloo_barrier() paddle.distributed.gloo_release() def test_gloo_release_with_multiprocess(num_of_ranks): jobs = [] server_endpoint = "127.0.0.1:%s" % (find_free_port()) for id in range(num_of_ranks): p = multiprocessing.Process( target=test_gloo_release, args=(id, num_of_ranks, server_endpoint)) jobs.append(p) p.start() for proc in jobs: proc.join() if __name__ == '__main__': # Arg: number of ranks (processes) test_gloo_release_with_multiprocess(2) N)rreleaserrrr gloo_releases7 r%)ossysr warningsmultiprocessingrrZ paddle.fluidrZ5paddle.distributed.fleet.base.private_helper_functionr__all__rrr"r#r%rrrrs   _ :