o Ti@sddlZddlZddlZddlmZddlmZddlm Z da da ga de fddZdd Zd d Zdd dZdddZddZdejdefddZdedejfddZddZdS)N)commrequired_torch_version)get_acceleratorreturncCs tddS)Ng?) min_versionrrrN/home/ubuntu/.local/lib/python3.10/site-packages/deepspeed/runtime/pipe/p2p.py can_send_recvs r cCs4|atjdks JdtsddtjDadSdS)Nz There is no pipeline parallelismcSsg|]}tj|dqS))ranks)dist new_group).0grouprrr "sz'init_process_groups..)_gridpipe_parallel_sizer p2p_groups_groups)gridrrr init_process_groupss rcCsRd}tjd}t||dks#||kr||ks%||kr||ks'JddSdSdS)Nrr zOFunctionality currently limited to send and receive between adjacent ranks only)rrabs) src_stage dest_stage first_stage last_stagerrr _is_valid_send_recv%s rFcCs|dksJdt}t||tj|d}|r&t||}t|dStr/t ||St ||}tj|d}tj ||||dSNFzDoesn't support async_op truestage_id)rasync_op) r get_stage_idrstage_to_globalr isend_asyncappendr send_get_send_recv_group broadcast)tensorrr!r dest_rankoprsrc_rankrrr r'.s      r'cCsz|dksJdt}t||tj|d}|r&t||}t|dStr/t ||St ||}tj ||||dSr) rr"rr#r irecvr%r&r recvr(r))r*rr!rr-r,rrrr r/Cs     r/cCs$tD]}|qgatdS)N)r%waitr synchronize)r,rrr r0Ws r0msgdestcCsjt|}ttj|t}tj t |gtj dt}t j ||dt j ||ddS)a+Send an arbitrary python object to ``dest``. Note: ``msg`` must be serializable by msgpack. WARN: This incurs a CPU -> GPU transfer and should be used sparingly for performance reasons. Args: msg (typing.Any): The object to send. dest (int): Destination rank. dtype)dstN)msgpackpackbtorch ByteTensor ByteStorage from_buffertor device_namer*lenlongr r')r2r3 length_tensorrrr send_obj`s "rBsendercstjdgtjdt}tj||dtj| tj dt}tj||dt | }fdd|}|S)zReceive an arbitrary python object from ``sender``. WARN: This incur a CPU <-> GPU transfers and should be used sparingly for performance reasons. Args: sender (int): The rank sending the message. rr4)srccst|r |tSt|ttfr(fdd|D}t|tr&t|}|St|trCt}| D] \}}|||<q4|S|S)z'Recursively move to the current device.csg|]}|qSrr)rx__torr rsz)recv_obj.._to..) r9 is_tensorr=rr> isinstancetuplelistdictitems)xretkeyvalrFrr rGs   zrecv_obj.._to)r9r*r@r=rr>r r/emptyitemuint8r7unpackbcpunumpytobytes)rClengthr2rrFr recv_objws   rZcCs`d}d}tjd}||kr||ks||kr||kr|}n ||kr#|}n|} tj|d}t|S)z@the group id is always the smaller rank unless its a wrap aroundNrr r)rrr#r)rrr rrgroup_idrrr r(s  r()F)r7typingr9 deepspeedrr deepspeed.utils.torchrdeepspeed.acceleratorrrrr%boolr rrr'r/r0AnyintrBrZr(rrrr s$      *