zjsddlmZddlmZmZddlZddlmZddlm Z m Z ddl m Z erddlm Z ddlmZdd lmZ dddZdddZ d d!dZdS)") annotations) TYPE_CHECKINGAnyN)stream)_get_global_group_warn_cur_rank_not_in_groupconvert_object_to_tensor)Tensor)task)GroupTtensorr dstintgroup Group | Nonesync_opboolreturn task | Nonec4tj||||dS)a Send a tensor to the receiver. Args: tensor (Tensor): The Tensor to send. Its data type should be float16, float32, float64, int32, int64, int8, uint8, bool or bfloat16. dst (int): The destination rank id. group (Group, optional): The group instance return by new_group or None for global default group. Default: None. sync_op (bool, optional): Whether this op is a sync op. The default value is True. Returns: Return a task object. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env: DISTRIBUTED) >>> import paddle >>> import paddle.distributed as dist >>> dist.init_parallel_env() >>> if dist.get_rank() == 0: ... data = paddle.to_tensor([7, 8, 9]) ... dist.send(data, dst=1) >>> else: ... data = paddle.to_tensor([1, 2, 3]) ... dist.recv(data, src=0) >>> print(data) >>> # [7, 8, 9] (2 GPUs) F)rrruse_calc_stream)rsend)rrrrs u/lsinfo/ai/hellotax_ai/data_center/backend/venv/lib/python3.11/site-packages/paddle/distributed/communication/send.pyrr#s)H ;Cugu   c(t|||dS)aB Send tensor asynchronously Args: tensor (Tensor): The Tensor to send. Its data type should be float16, float32, float64, int32, int64, int8, uint8, bool or bfloat16. dst (int): The destination rank. group (Group, optional): The group instance return by new_group or None for global default group. Default: None. Returns: Return a task object. Warning: This API only supports the dygraph mode. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env: DISTRIBUTED) >>> import paddle >>> import paddle.distributed as dist >>> dist.init_parallel_env() >>> if dist.get_rank() == 0: ... data = paddle.to_tensor([7, 8, 9]) ... task = dist.isend(data, dst=1) >>> else: ... data = paddle.to_tensor([1, 2, 3]) ... task = dist.irecv(data, src=0) >>> task.wait() # type: ignore[union-attr] >>> print(data) >>> # [7, 8, 9] (2 GPUs) F)r)r)rrrs risendrLsF UE 2 2 22r object_list list[Any] int | None dst_in_groupc |t|dkrtd|tn|}t|rdS|'|td||}n|dn|}t d|D\}}d|D}t j|d}t||| t|d kr |d}nt j |}t||| dS) a[ Send a list of Python objects to the receiver. Args: object_list (list): The list of Python objects to send. dst (int, optional): The destination rank id. Default: 0. group (Group, optional): The group instance return by new_group or None for global default group. Default: None. dst_in_group (int, optional): The destination rank within the group. Cannot be specified together with dst. Default: None. Returns: This function does not return any value. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env: DISTRIBUTED) >>> import paddle >>> import paddle.distributed as dist >>> dist.init_parallel_env() >>> if dist.get_rank() == 0: ... data = ["hello", {"key": 100}, [1, 2, 3]] ... dist.send_object_list(data, dst=1) >>> else: ... data = [None] * 3 # type: ignore ... dist.recv_object_list(data, src=0) >>> print(data) >>> # ["hello", {"key": 100}, [1, 2, 3]] (2 GPUs) Nrz#object_list cannot be None or emptyz7Cannot specify both 'dst' and 'dst_in_group' arguments.c,g|]}t|Sr ).0objs r z$send_object_list..s! ? ? ?C "3 ' ' ? ? ?rc6g|]}|Sr$)item)r%sizes rr'z$send_object_list..s ::: :::rint64)dtype)rr) len ValueErrorrrget_global_rankzippaddle to_tensorrconcat) rrrr! tensor_list size_listsize_list_valuesobject_sizes_tensor object_tensors rsend_object_listr:rsKFc+..!33>???#(=    eE"5)) ?I ##L11;aaC! ? ?; ? ? ?K;: :::!*+;7KKK #U3333 ;1#A  k22 Cu------r)rNT) rr rrrrrrrr)N)rr rrrrrr)NNN)rrrr rrr!r ) __future__rtypingrrr2 paddle.distributed.communicationr&paddle.distributed.communication.grouprr4paddle.distributed.communication.serialization_utilsr r paddle.base.corer r rrr:r$rrrAsF#"""""%%%%%%%% 333333=%%%%%%<<<<<<  &&&&&R#3#3#3#3#3P# B.B.B.B.B.B.B.r