U ʹj0@sddlZddlZddlmZmZmZmZmZmZm Z m Z m Z ddl m ZddlmZmZmZmZddlZddlmZddlmZddlmZddlmZmZd d gZed Zed Z eeeefee!efd ddZ"GdddZ#eeedddZ$ej%eej%ddd Z&eee ddd Z'e e!e e!ee!dddZ(e e!e e!ee!dddZ)Gdddej*Z+ej*e!e!ej*dd d!Z,e-e!e-d"d#d$Z.dS)%N) ListCallableOptionalUnionTypeVarDictAnycastSequence)CheckpointException_wrap_exception_is_wrapped_exceptionWRAPPED_EXCEPTION) ShardedTensor)Shard)DTensor)STATE_DICT_TYPE MetadataIndexfind_tensor_shardfind_state_dict_objectTR)resultsreturncCs ttttfddt|DS)NcSsi|]\}}t|r||qS)r).0ierrrra/var/www/html/bwave/python/venv/lib/python3.8/site-packages/torch/distributed/checkpoint/utils.py .sz%_get_failure_dict..)r rintr enumerate)rrrr_get_failure_dict)s r#c@s,eZdZdZeejeedddZ edddZ eddd Z ee e d d d Z e eee d d dZe ee d ddZeee e dddZeege feee geefedddZeege feee gefedddZeege fee dddZeege fe dddZdS) _DistWrapperaH This is a wrapper around PG that provides a series of features around object collectives. It works without distributed initialized, where most collectives turns into nops. All variants that take functions are exception robust, meaning that if one or more ranks raise errors, all ranks will observe those. )groupuse_distcoordinator_rankcCsB||_||_||_|jr2t||_|j|k|_n d|_d|_dS)NrT)r%r&r'distget_rankrankis_coordinator)selfr%r&r'rrr__init__<s z_DistWrapper.__init__rcCs|jSN)r*r,rrrr)Lsz_DistWrapper.get_rankcCs|jrt|jSdS)Nr )r&r(get_world_sizer%r0rrrr1Os z_DistWrapper.get_world_size)objectrcCs.|g}|jr tj||j|jdtt|dS)z\ Same as c10d::broadcast_object_list but works without distributed enabled. ) object_listr%srcr)r&r(Zbroadcast_object_listr%r'r r)r,r2r3rrrbroadcast_objectTsz_DistWrapper.broadcast_objectcCs\|jrR|jr(tttdgt|jnd}tj||jr<|nd|j |jd|}n|g}|S)zT Same as c10d::gather_object but works without distributed enabled. N)objZobject_gather_listdstr%) r&r+r rrr(r1r% gather_objectr')r,r2 gather_objsresultrrrr8as z_DistWrapper.gather_objectcCs@|jr6tttdgt|j}tj|||jdn|g}|S)zX Same as c10d::all_gather_object but works without distributed enabled. N)r3r6r%)r&r rrr(r1r%all_gather_object)r,r2r9rrrr;wsz_DistWrapper.all_gather_object)r3rcCsX|jr@tttdg}tj||jr&|nd|j|jd|d}n|dk sLt |d}|S)zU Same as c10d::scatter_object but works without distributed enabled. N)Zscatter_object_output_listZscatter_object_input_listr4r%r) r&r rrr(Zscatter_object_listr+r'r%AssertionError)r,r3Z gather_resultZ local_replyrrrscatter_objects  z_DistWrapper.scatter_object)stepmap_fun reduce_funrc Csz |}Wn*tk r4}z t|}W5d}~XYnX||}d}|jr|dk sVtt|}t|dkrz(ttt t t f|ttt |}Wn0tk r}zt|||j <W5d}~XYnXt|dkrt ||g|}||} t| t r| | S)a^ Compute a value on each rank, then do centralized reduce on a single rank, followed by a scatter. This method operates in the following way: Run ``map_fun`` on all ranks Gather results on rank 0 Call ``reduce_fun`` on all those values Scatter to each rank part of the result. Nr) BaseExceptionr r8r+r<r#lenr rrrr rr*r1r= isinstance) r,r>r?r@ local_dataeall_data all_results node_failuresr:rrrreduce_scatters4        z_DistWrapper.reduce_scatterc Csz |}Wn*tk r4}z t|}W5d}~XYnX||}d}|jr|dk sVtt|}t|dkrz|ttt |}Wn0tk r}zt|||j <W5d}~XYnXt|dkrt ||}| |} t | t r| tt| S)aa Compute a value on each rank, then do centralized reduce on a single rank, followed by a broadcast. This method operates in the following way: Run ``map_fun`` on all ranks Gather results on rank 0 Call ``reduce_fun`` on all those values Broadcast the reduced value to all ranks. Nr)rAr r8r+r<r#rBr rrr*r r5rCr) r,r>r?r@rDrErFr:rH final_resultrrr all_reduces(         z_DistWrapper.all_reduce)r>r?rc Cslz |}Wn*tk r4}z t|}W5d}~XYnX||}t|}t|dkr^t||ttt|S)z Compute a value on each rank, then all_gather them. This method operates in the following way: Run ``map_cp`` on all ranks all_gather the values to all ranks Nr) rAr r;r#rBr r rr)r,r>r?r:rErGrHrrr all_gathers     z_DistWrapper.all_gatherc Csnd}|jrLz |}Wn6tk rJ}zt||jt|i}W5d}~XYnX||}t|trd|tt|S)z Compute a value on rank 0 and broadcast it. This method operates in the following way: Run ``map_cp`` on rank 0 broadcast the value N) r+rAr r*r r5rCr r)r,r>r?r:rErJrrr broadcast s     z_DistWrapper.broadcastN)__name__ __module__ __qualname____doc__rr(Z ProcessGroupboolr!r-r)r1rr5rr8r;r=strrrrIrKrLrMrrrrr$2s:     1  *   r$)tensorindexrcCs|jdkrtd|jd|}|jdk r`t||jkr`t||jjj |jkr`||jS|D] }t|jj |jkrd|Sqdtd|jd|jddS)NzCannot lookup z5 since its a ShardedTensor and no offset was providedzCould not find shard at 'z ' for FQN: '') offset ValueErrorfqnZ local_shardsrUrBtorchSizemetadataZ shard_offsets)rTrUZshardsZshardrrr _find_shard&s&      r]cCsrt|tr|St|tr(t||jS|jdk rn|jtdgt | krT|St d|j d|jd|S)NrFQN: '1' is not a ShardedTensor, can't find by offset: 'rV) rCrZto_localrr]rTrWrZr[rBsizerXrY)rTrUrrrr>s    ) state_dictrUrcCsd|j|krtd|jd||j}t|tjr`sz%_element_wise_add..ziprcrdrrr_element_wise_add_srjcCsddt||DS)NcSsg|]\}}||qSrrrerrrrfdsz%_element_wise_sub..rgrirrr_element_wise_subcsrkcs|eZdZejeedfdd ZejfeeedddZ eddd Z e dd d Z e dd d Z ddZdddZZS) _ReaderView) base_streamrWrBcs*t||_||_||_|ddS)Nr)superr-rWrBrmseek)r,rmrWrB __class__rrr-hs  z_ReaderView.__init__)_ReaderView__offset_ReaderView__whencercCsD|tjkr|j|}n |tjkr6tj}|j|j|}|j||Sr/)osSEEK_SETrWSEEK_ENDrBrmro)r,rrrsrrrroos    z_ReaderView.seekr.cCs|j|jSr/)rmtellrWr0rrrrwwsz_ReaderView.tellcCs |jSr/)rmreadabler0rrrrxzsz_ReaderView.readablecCs |jSr/)rmseekabler0rrrry}sz_ReaderView.seekablecCs |j|Sr/)rmreadinto)r,rdrrrrzsz_ReaderView.readintocCs |j|Sr/)rmread)r,r`rrrr|sz_ReaderView.read)r{)rNrOrPioIOBaser!r-rtrurorwrRrxryrzr| __classcell__rrrprrlgsrl)filerWlengthrcCs t|||Sr/)rl)rrWrrrr_create_file_viewsr) device_type device_idrcCs|dkr dS|d|S)z$ Device info normalization. cpu:r)rrrrr_normalize_device_infosr)/rtr}typingrrrrrrrr r Ztorch.distributed distributedr(apir r rrrZZ'torch.distributed._shard.sharded_tensorrZ-torch.distributed._shard.sharded_tensor.shardrZtorch.distributed._tensorrr\rr__all__rrr!r#r$r]rbrrrjrkr~rlrrSrrrrrs>,     u