o "jC:@sddlZddlZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Z ddgZed dZd d Zd d ZddZdddZddZddZddZdS)N) dygraph_only)fleet)logger)save_for_auto_inferencesaverc KsD|dd}tdks|durt|}tj||fi|S|dd}t|ts-Jd|dvs5Jdt |d|durOt d |d |d |d t }|dkr_|dksmJd |d||}|}|dkr|jdkrt||sJd|jdkrt||sJdt|}tj||fi|S|jdkrt|}tj||fi|St||rtj||fi|St|tttfsJt|tr|g}|dd}z9t dt|t||||d} t dt|t|vrt|}tj| |fi|WdSWdStd|d)a: Save a state dict to the specified path in both distributed and single-card environment. Note: Now supports saving ``state_dict`` of Layer/Optimizer, Tensor and nested structure containing Tensor, Program. Note: Different from ``paddle.jit.save``, since the save result of ``paddle.save`` is a single file, there is no need to distinguish multiple saved files by adding a suffix. The argument ``path`` of ``paddle.save`` will be directly used as the saved file name instead of a prefix. In order to unify the saved file name format, we recommend using the paddle standard suffix: 1. for ``Layer.state_dict`` , recommend to use ``.pdparams`` ; 2. for ``Optimizer.state_dict`` , recommend to use ``.pdopt`` . For specific examples, please refer to API code examples. Args: obj(Object) : The object to be saved. path(str|BytesIO) : The path/buffer of the object to be saved. If saved in the current directory, the input path string will be used as the file name. protocol(int, optional): The protocol version of pickle module must be greater than 1 and less than 5. Default: 4. **configs(dict, optional): optional keyword arguments. The following options are currently supported: 1. use_binary_format(bool): To be used in paddle.save. When the saved object is static graph variable, you can specify ``use_binary_for_var``. If True, save the file in the c++ binary format when saving a single static graph variable; otherwise, save it in pickle format. Default: False. 2. gather_to(int|list|tuple|None): To specify which global rank to save in.Defalut is None. None value means distributed saving with no gathering to a single card. 3. state_type(str): Value can be 'params' or 'opt', specifying to save parametres or optimizer state. 4. max_grouped_size(str|int): To limit the max size(how many bits) a object group to be transfered a time. If str, the format must be as num+'G/M/K', for example, 3G, 2K, 10M, etc. Default is 3G. Returns: None Examples: .. code-block:: python >>> # doctest: +SKIP('TODO: the error will be fix in the feature') >>> import paddle >>> paddle.distributed.init_process_group(backend='nccl') >>> paddle.distributed.fleet.init(is_collective=True) >>> model = build_model() >>> optimizer = build_optimizer(model) >>> dist_optimizer = paddle.distributed_optimizer(optimizer) >>> dist_model = paddle.distributed_optimizer(model) >>> # gather params to rank 0 and then save >>> paddle.incubate.distributed.utils.io.save(model.state_dict(), path="path/to/save.pdparams", gather_to=[0], state_type="params") >>> # save whoe params on all ranks >>> paddle.incubate.distributed.utils.io.save(model.state_dict(), path="path/to/save.pdparams", gather_to=[0,1], state_type="params") >>> # save optimizer state dict on rank 0 >>> paddle.incubate.distributed.utils.io.save(optimizer.state_dict(), path="path/to/save.pdopt", gather=0, state_type="opt") gather_toNr state_typez|must pass an arg state_type='params' or state_type='opt' to specify whether to save model state_dict or optimizer state_dict)paramsoptz8must pass an arg state_type='params' or state_type='opt'$zYou are saving z, while the path(z does not end with )z;Only DP and Sharding is supported now. However, current MP=z , PP=r z0only sharding stage 1/2 and DP are supported nowZmax_grouped_size3Gzstate_dict_keys:)max_sizezgathered_state_dict_keys:zSaving failed. Follwing are some suggestions: 1) pass the param max_grouped_size to turn the grouped size smaller (current value of max_grouped_size is zr) 2) if sharding stage is 1, use paddle.save rather than paddle.distributed.save 3) Concat the developers )getdistget_world_size_remove_not_supported_confpaddler isinstancestrresearchrwarningrZget_hybrid_communicate_groupZget_model_parallel_world_sizeZget_pipe_parallel_world_sizeZget_sharding_parallel_groupZget_data_parallel_groupZnranks _same_keyslisttupleintinfokeys_gather_state_dictget_rank RuntimeError) state_dictpathconfigsrr ZhcgZsharding_groupZdp_grouprZgathered_state_dictr&o/var/www/html/Deteccion_Ine/venv/lib/python3.10/site-packages/paddle/incubate/distributed/utils/io/dist_save.pyrs C         c csd}|D]\}}|t|t|kr!t|t|}qt||}td|i}t|}d}d}|t|krt|||t||} || |kr^|Vi}d}||||||<|d7}|| 7}|t|kr}|dkr}|V|t|ksAdSdS)zw Description: Generator of state dict groups to transfer.the size of each group is less than max_size. rzmax tensor size: rN) itemssys getsizeofmaxrdebugrrlen) r#rZmax_tensor_sizekvZ state_groupZk_listindexbitsZbsizer&r&r'_state_dict_groupss4    r2cCs |D] }t|dkr dSqdS)z& Check if all items are empty rFTr-)Z dict_listr/r&r&r' all_emptys  r4cCst|ttfs Jt|tr?td|sJd|t|dd}|ddkr/|d}|S|ddkr;|d}|S|d }|S) z Parse an integer or a mem size str to an integer convert xxxG to xxx * 1024^3 convert xxxM to xxx * 1024^2 convert xxxK to xxx * 1024^1 z ^[0-9]*[GMK]$zeWrong max_size 's format, the format ust be like 10K, 9M, 200G , etc, or an integer. However this is NGi@Mii)rrrrr)rnumr&r&r'_parse_mem_size_to_bitss    r9rc Cst|tttfs Jdt|tr|g}t|}|t|}tdt |}d}d}d}d|vr;| dd}d}d|vrG| dd}d}t ||||} t|t r[t ||||} n |dushJdt ||rn| | d<|rt|| d<| S) aq Description: Gather state dicts across all group ranks to dst, Depiring the same elements. including LR_Scheduler. Args: state_dict(dict): local state dict dst(int|list|tuple): ranks the state dicts are gathered to group(ProcessGroup): group across which the state dicts are gathered max_size(int|str): The max limitation of the gathered tensor group size transformered a time. Default is 3G bits. Each rank 's max tensor group before gathering is max_size // group.size Returns: Gathered state dict z,dst' type must be one of int, list and tuplezlen state_dict: len(state_dict)NFZmaster_weightsTZ LR_Schedulerz%Wrong type of master weights . type: )rrrrr9rrrr,copypop_grouped_gather_data_dictdicttype) r#dstgrouprZ state_dict_mwZhas_mwZhas_lrlroutputZmastersr&r&r'r s:       r c Csi}tdt||D]\}}z|||<Wqtdt|d|dd}i}tdt||D]W} g} |t| 7}td|dt|t | | |t |vr~| D]} | D]\}}td |d |j qg| | qatd td d | Ddt|q; g} i} tdt | | |t| rn:t |vr| D]} | D]\}}td |d |j q| | qtd tdd | Ddt|qtdt |vr t} |D]}tj||| d||<|||_q|SiS)a Description: Gather state data dict by groups. Args: state__data_dict(dict): local dict to transfer.The state_data_dict only contains the mapping: str->paddle.Tensor dst(int|list|tuple): ranks the state dicts are gathered to group(ProcessGroup): group across which the state dicts are gathered max_size(int|str): The max limitation of the gathered tensor group size transformered a time. Default is 3G bits. Each rank 's max tensor group before gathering is max_size // group.size Returns: Gatherd state_data_dict zlen state_tict_ : zthe object (type of z) of 'z!' is neither tensor nor parameterrzstart all gather ...zgen to gather: z / z gathered: z, z s list size: cs|]}t|VqdSNr3.0sr&r&r' Wz,_grouped_gather_data_dict..z output: Tz while TruecsrDrEr3rFr&r&r'rIirJzall gathered ...)place)rr,r-r(numpy TypeErrorr>rr2rall_gather_objectr!shapeupdatesumr4rZCPUPlacerZ to_tensorname) Zstate_data_dictr?r@rZ numpy_dictr.r/totalZ output_statestateZs_listrHrKr&r&r'r<)sb            r<cCsFt|}g}t|tj|||d|D] }||ks dSqdS)z Check whther all keys in each dict in the group are the same. Used in sharding strategy to determine whether a dict needs to be gathered. )r@FT)rrrrrrN)r#r@rZkey_listr.r&r&r'rxs  rcCs6dg}t|}|D] }||vr||dq |S)z? Remove the config values not supported by paddle.save Zuse_binary_formatN)r:rr;)r%Z__supported_by_save__Zconfigs_r.r&r&r'rs   r)r)r:rr)rZpaddle.distributed distributedrZpaddle.base.frameworkrrZ'paddle.distributed.fleet.utils.log_utilrZ save_for_autor__all__rr2r4r9r r<rrr&r&r&r's(       ' ;O