x-jI"hddlmZddlZddlmZmZddlmZddlm Z gZ da dZ ddZ ddZddZdS)) annotationsN)ManagerProcess)core)wait_server_readycjddlm}|t||}|d}|dds|s>t j||dd*|>|dS)Nr)KVServer)sizerunningF) *paddle.distributed.fleet.utils.http_serverr intstartget should_stoptimesleepstop)port http_server_dr r http_server wait_secondss e/var/www/html/banglarbhumi/venv/lib/python3.11/site-packages/paddle/distributed/parallel_with_gloo.py_start_kv_serverrsCCCCCC(3t994000KL   Iu - -![5L5L5N5N! <      Iu - -![5L5L5N5N!rank_idrrank_numserver_endpointstrreturnNonec|dkdus Jdt}|}d|d<|dkrcd|i}ttt |dd||f }d |_d |d<|t|gtj }||_ ||_ |dd|_ t |dd|_d |_d |_tj|at$|dkrd|d<|d Sd S)a 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() # type: ignore >>> 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) FzSrank_num should greater than or equal to 2 for parallel environment initialization.r r_worker:)targetargsTiiN)rdictrrrsplitdaemonrrrGlooParallelStrategyrankr ip_addressip_port init_seconds run_secondsGlooParallelContext_global_gloo_ctxinitjoin)rrrmanagerhttp_server_statusr http_server_proc gloo_strategys rgloo_init_parallel_envr:*sr qLU " " "] # " " iiG $)y!!||8$"#o++C003446H$O   #'(,9%   '(((-//M M%M.44S99!>M!%M 'M/ >>!||(-9%|rcZt JdtdS)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() # type: ignore >>> 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) Nz gloo context is not initialized.)r3barrierrr gloo_barrierr>s2l  ' ')K ' ' 'rcJttdSdS)aN 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() # type: ignore >>> 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)r3releaser=rr gloo_releaserAs+n#  """""$#r)rrrrrrr r!)r r!) __future__rrmultiprocessingrr paddle.baser5paddle.distributed.fleet.base.private_helper_functionr__all__r3rr:r>rAr=rrrGs#""""" ,,,,,,,, ` ` ` ` F7777t8#8#8#8#8#8#r