ΑiI"|SSKJr SSKrSSKJrJr SSKJr SSKJ r /r Sq Sr S Sjr S SjrS S jrg) ) annotationsN)ManagerProcess)core)wait_server_readycXSSKJn U"[U5US9nUR5 SnUR SS5(dUR 5(dF[ R"U5 UR SS5(aM/UR 5(dMFUR5 g)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.13/site-packages/paddle/distributed/parallel_with_gloo.py_start_kv_serverrsC3t940KL   Iu - -[5L5L5N5N <    Iu - -[5L5L5N5NczUS:SLdS5e[5nUR5nSUS'US:XaKSU0n[[[ UR S5S5XE4S 9nS UlS US'UR5 [U/5 [R"5nXl Xl UR S5SUl [ UR S5S5UlS UlS Ul[R""U5q[$R'5 US:XaSUS'WR)5 g g )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)rdictrrrsplitdaemonrrrGlooParallelStrategyrankrank_num ip_addressip_port init_seconds run_secondsGlooParallelContext_global_gloo_ctxinitjoin)rank_idr(server_endpointmanagerhttp_server_statusr http_server_proc gloo_strategys rgloo_init_parallel_envr7*sNr qLU "] " iG $)y!!|8$"#o++C0346HO #'(,9% '(--/M %.44S9!M!%M 'M// >!|(-9%rcH[cS5e[R5 g)ap 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.)r.barrierrr gloo_barrierr;s#l  'K)KK 'rc<[b[R5 gg)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() # 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)r.releaser:rr gloo_releaser>sn#  "$r)r1rr(rr2strreturnNone)r@rA) __future__rrmultiprocessingrr paddle.baser5paddle.distributed.fleet.base.private_helper_functionr__all__r.rr7r;r>r:rrrGsb# , ` ` ` 25`  ` F7t8#r