o ۾i#@sdZddlZddlmZmZddlmZddlZdZdZ ee fdee e ej fded e d ee e ej fgej fd e d e d dfddZee fdee e e ee ejffded e deee e ej fgdfd e d e d dfddZdS)z6Packed tensor utilities for efficient weight transfer.N)CallableIterator)Anyi@iteratorgroupsrcpost_iter_funcbuffer_size_bytes num_buffersreturnc Cs|}ddt|D}d}ddt|D} ddt|D} ddt|D} ||tj||zMg| |<d| |< |t|tjd} | | | | || 7<| ||krinqBtj | |dd | |<|j | ||d |d |}Wn,t yt| |dkrtj | |dd | |<|j | ||d YWd d SwWd n1swYq)) aBroadcast tensors in a packed manner from trainer to workers. Args: iterator: Iterator of model parameters. Returns a tuple of (name, tensor) group: Process group (PyNcclCommunicator) src: Source rank (0 in current implementation) post_iter_func: Function to apply to each (name, tensor) pair before packing, should return a tensor buffer_size_bytes: Size in bytes for each packed tensor buffer. Both producer and consumer must use the same value. num_buffers: Number of buffers for double/triple buffering. Both producer and consumer must use the same value. cSg|]}tjqStorchcudaStream.0_rrb/home/ubuntu/.local/lib/python3.10/site-packages/vllm/distributed/weight_transfer/packed_tensor.py )z-packed_broadcast_producer..rcSg|]}gqSrrrrrrr,cSg|]}dqSrrrrrrr-rcSg|] }tjdtjddqSrrdtypedeviceremptyuint8rrrrr.T)dimrN)range synchronizerrstreamnext contiguousviewr$appendnumelcat broadcast StopIterationlen) rrrr r r target_packed_tensor_sizestreams buffer_idxpacking_tensor_listpacking_tensor_sizespacked_tensorstensorrrrpacked_broadcast_producersT       r=post_unpack_funccCsXdtjdttdtttdttjdttdttttjff dd}|}d d t|D}d } d d t|D} d d t|D} dd t|D} || tj || g| | <d | | <zh t |\} \}}t ||j}| | | |||f| | |7<| | |krnqftj| | tjdd| | <|j| | |dt| | \}}}}||| | t|t|t|t|| d|} WnNtyt| | d krtj| | tjdd| | <|j| | |dt| | \}}}}||| | t|t|t|t|YWddSwWdn 1s&wYqM)aConsume packed tensors and unpack them into a list of tensors. Args: iterator: Iterator of parameter metadata. Returns (name, (shape, dtype)) group: Process group (PyNcclCommunicator) src: Source rank (0 in current implementation) post_unpack_func: Function to apply to each list of (name, tensor) after unpacking buffer_size_bytes: Size in bytes for each packed tensor buffer. Both producer and consumer must use the same value. num_buffers: Number of buffers for double/triple buffering. Both producer and consumer must use the same value. packed_tensornamesshapesdtypes tensor_sizesr cSs&||}ddt||||D}|S)ayUnpack a single tensor into a list of tensors. Args: packed_tensor: The packed torch.uint8 tensor to unpack names: List of tensor names shapes: List of tensor shapes dtypes: List of tensor dtypes tensor_sizes: List of tensor sizes in bytes Returns: unpacked List[(name, tensor)] cSs,g|]\}}}}|||j|fqSr)r.r/)rnameshaper r<rrrrs zDpacked_broadcast_consumer..unpack_tensor..)splitzip)r?r@rArBrCunpacked_tensors unpacked_listrrr unpack_tensorps z0packed_broadcast_consumer..unpack_tensorcSr rrrrrrrrz-packed_broadcast_consumer..rcSrrrrrrrrscSrrrrrrrrrcSrrr"rrrrrr%Trrr(r)N)rTensorliststrintr tupler*r+rr,r-mathproditemsizer0r#r$r3rGr4r5)rrrr>r r rJr6r7r8packing_tensor_meta_datar:r;rDrEr tensor_sizer@rArBrCrrrpacked_broadcast_consumerYs            #rU)__doc__rPcollections.abcrrtypingrr DEFAULT_PACKED_BUFFER_SIZE_BYTESDEFAULT_PACKED_NUM_BUFFERSrOrMrKrNr=rLr rUrrrrsR  M