x-jddlmZddlZddlmZmZddlmZddl m Z ddl m Z m Z er'ddlmZmZddl mZddlmZdd lmZeeeegefZGd d Zej dddZddZddZdS)) annotationsN) TYPE_CHECKINGCallable) framework)_get_global_group_warn_cur_rank_not_in_group) GeneratorSequence)Tensor)task)GroupcHeZdZUdZded<ded<ded<ded < ddd Zd S)P2POpa A class that makes point-to-point operations for "batch_isend_irecv". This class creates the type of P2P operation, communication buffer, peer rank, Group. Instances of this class will be passed to ``paddle.distributed.batch_isend_irecv`` for point-to-point communication. Args: op (callable): A function to send data to or receive data from a peer process. The type of ``op`` is either ``paddle.distributed.isend`` or ``paddle.distributed.irecv``. tensor (Tensor): Tensor to send or receive. peer (int): The destination or source rank. group (Group, optional): The group instance return by new_group or None for global default group. Default: None. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env: DISTRIBUTED) >>> import paddle >>> import paddle.distributed as dist >>> dist.init_parallel_env() >>> rank = dist.get_rank() >>> world_size = dist.get_world_size() >>> send_t = paddle.arange(2) + rank >>> # paddle.tensor([0, 1]) # Rank-0 >>> # paddle.tensor([1, 2]) # Rank-1 >>> recv_t = paddle.empty(shape=[2], dtype=send_t.dtype) >>> send_op = dist.P2POp(dist.isend, send_t, (rank + 1) % world_size) >>> recv_op = dist.P2POp(dist.irecv, recv_t, (rank - 1 + world_size) % world_size) _P2POpTypeopr tensorintpeer Group | NonegroupNreturnNonec|tjtjfvrtd||_||_||_|tn||_dS)NztInvalid ``op`` function. Expected ``op`` to be of type ``paddle.distributed.isend`` or ``paddle.distributed.irecv``.) distisendirecv RuntimeErrorrrrrr)selfrrrrs r/var/www/html/banglarbhumi/venv/lib/python3.11/site-packages/paddle/distributed/communication/batch_isend_irecv.py__init__zP2POp.__init__Psa dj$*- - -0    ,1M&(((u N) rrrr rrrrrr)__name__ __module__ __qualname____doc____annotations__r r!rrr$so$$LNNNNNN III# EEEEEEEr!rrr tasks task | NonerGenerator[None, None, None]c#vK|tn|}|j}| dV|t|dkr|dS||dS#|t|dkr|w||wxYw)Nr)r process_group_start_coalescinglen_end_coalescing)rr)pgs r_coalescing_managerr2ds$)=    eE  B&  =CJJ!OO      u % % % % % =CJJ!OO      u % % % %s A77AB8 p2p_op_listSequence[P2POp]rct|trtd|Dstd|djjtfd|DstddS)zu Helper to check that the ``p2p_op_list`` is a list of P2POp instances and all ops use the same backend. c3@K|]}t|tVdSr") isinstancer).0p2p_ops r z%_check_p2p_op_list..ys=44&, 65!!444444r!z[Invalid ``p2p_op_list``. Each op is expected to to be of type ``paddle.distributed.P2POp``.rc38K|]}|jjkVdSr")rbackend)r8r9r<s rr:z%_check_p2p_op_list..s-II6w&,..IIIIIIr!z(All groups need to use the same backend.N)r7listallrrr<)r3r<s @r_check_p2p_op_listr?ts k4 ( ( 440;44411  :   !n"*G IIII[III I IGEFFFGGr! list[P2POp] list[task]ct||dj}t|rdStjr|t n|}|j}g}t||5|D]B}|j}|j }|j }|j}||||} | | | C dddn #1swxYwY|Std)a Send or Receive a batch of tensors asynchronously and return a list of requests. Process each of the point-to-point operations in ``p2p_op_list`` and return the corresponding tasks. NCCL are currently supported. Args: p2p_op_list (List[P2POp]): A list of point-to-point operations(type of each operator is ``paddle.distributed.P2POp``). The order of the isend/irecv in the list matters and it needs to match with corresponding isend/irecv on the remote end. Returns: A list of distributed tasks returned by calling the corresponding op in the op_list. 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() >>> rank = dist.get_rank() >>> world_size = dist.get_world_size() >>> send_t = paddle.arange(2) + rank >>> # paddle.tensor([0, 1]) # Rank-0 >>> # paddle.tensor([1, 2]) # Rank-1 >>> recv_t = paddle.empty(shape=[2], dtype=send_t.dtype) >>> send_op = dist.P2POp(dist.isend, send_t, (rank + 1) % world_size) >>> recv_op = dist.P2POp(dist.irecv, recv_t, (rank - 1 + world_size) % world_size) >>> tasks = dist.batch_isend_irecv([send_op, recv_op]) >>> for task in tasks: ... task.wait() >>> print(recv_t) >>> # paddle.tensor([1, 2]) # Rank-0 >>> # paddle.tensor([0, 1]) # Rank-1 rNz*Don't support static graph mode currently.) r?rrrin_dynamic_moderr<r2rrrappendr) r3rr<r)r9rrr comm_groupr s rbatch_isend_irecvrFs>d{### N E"5)) ""I',}!###%-  . . ' '% ' 'Y{#\ r&$ 33#LL&&& ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' GHHHs,AB??CCr")rr r)r*rr+)r3r4rr)r3r@rrA) __future__r contextlibtypingrrpaddle.distributed distributedrpaddler&paddle.distributed.communication.grouprrcollections.abcr r r paddle.base.corer r rrrcontextmanagerr2r?rFr(r!rrQs#"""""********!!!!!! 633333333%%%%%%((((((63.45J=E=E=E=E=E=E=E=E@ '+ & & & & &GGGG$FIFIFIFIFIFIr!