o پi -@sdZddlmZddlZddlmZddlZddlmZm Z ddl m Z ddl m Z ddlmZdd lmZdd lmZdd lmZdd lmZeeZd&ddZd'ddZd(ddZd)ddZd*d"d#ZGd$d%d%ZdS)+af In-place weight updates for diffusion pipeline modules. This module provides WeightsUpdater, which swaps model weights at runtime without restarting the server. It is the diffusion-engine counterpart of the LLM engine's ModelRunner.update_weights_from_disk. Detailed usage of higher level API can be found in /python/sglang/multimodal_gen/test/server/test_update_weights_from_disk.py Key design decisions: - All-or-nothing with rollback: modules are updated sequentially. If any module fails (shape mismatch, corrupted file, etc.), every module that was already updated is rolled back by reloading its weights from pipeline.model_path (the last successfully-loaded checkpoint). On success, pipeline.model_path is updated to the new model_path so that future rollbacks target the latest good checkpoint, not the originally-launched model. - Rollback failures propagate: if rollback itself fails, the exception is not caught so the caller knows the model is in an inconsistent state. This matches the LLM engine behaviour. - Offload-aware: the diffusion LayerwiseOffloadManager replaces GPU parameters with torch.empty((1,)) placeholders while real weights live in consolidated pinned CPU buffers. A naive param.data.copy_() would fail with a shape mismatch. Instead, the updater dynamically detects active offload managers and writes new weights directly into their CPU buffers via update_cpu_weights(), bypassing the placeholders entirely. For any layer that happens to be prefetched on GPU at update time, the live GPU tensor is also updated so the change takes effect immediately. This requires no extra GPU memory and does not disturb the offload state. - DTensor-aware: parameters that have been distributed via torch.distributed.tensor are updated through distribute_tensor so that each shard is correctly placed on the right device mesh. ) annotationsN)Path)DTensordistribute_tensor) TeaCacheMixin)_list_safetensors_files)safetensors_weights_iterator)DiffusersPipeline)maybe_download_model)OffloadableDiTMixin) init_loggerreturndict[str, torch.nn.Module]cCsLt|tr|d}|dur|jdur|j}ni}n|j}dd|DS)zReturn updatable nn.Module components for the given pipeline. Works with both the native ComposedPipelineBase backend and the DiffusersPipeline wrapper. diffusers_pipelineNcSs$i|]\}}t|tjjr||qS) isinstancetorchnnModule).0nmrrh/home/ubuntu/.local/lib/python3.10/site-packages/sglang/multimodal_gen/runtime/loader/weights_updater.py Ns$z)get_updatable_modules..)rr get_module componentsmodulesitems)pipelinediffusers_piperawrrrget_updatable_modules@s  r! weights_dirstrcCs"t|}|s td|t|S)zAReturn a (name, tensor) iterator over safetensors in weights_dir.zNo safetensors files found in )rFileNotFoundErrorr)r"safetensors_filesrrr_get_weights_iterQsr&local_model_pathmodules_to_update!list[tuple[str, torch.nn.Module]] tuple[dict[str, str], list[str]]cCsVi}g}|D] \}}t||}|r!tt|r!t|||<q||q||fS)zCheck that every module has a weights directory with safetensors files. Returns: (weights_map, missing) where weights_map maps module name to its weights directory and missing lists modules without weight files. )rexistsrr#append)r'r( weights_mapmissing module_name_r"rrr_validate_weight_filesYs    r1moduletorch.nn.ModuleNonecsg}t|tr|jrdd|jD}|r>t|}t|D] }||qfdd|D}t|t| dSt|t| dS)zLoad weights into a module, handling offload-managed parameters. For offloaded modules, updates CPU buffers directly via update_cpu_weights(); non-offloaded parameters use in-place copy. cSsg|]}|jr|qSr)enabled)rrrrr vsz-_load_weights_into_module..c3s$|] \}}|vr||fVqdSNr)rrwoffloaded_namesrr }s"z,_load_weights_into_module..N) rr layerwise_offload_managersdictsetupdateupdate_cpu_weightsrload_weights_into_modelnamed_parameters)r2 weights_iteroffload_managers weight_dictmanager remainingrr9r_load_weights_into_modulensrH model_paramsr=cCs|D]E\}}||vr q||}|j|jkr$td|d|jd|jt|tr=t||j|j|j}|j |j q|j ||jqdS)z:Copy weights from weights_iter into model_params in-place.zShape mismatch for z: model=z , loaded=N) shape ValueErrorrrrtodtype device_mesh placements _local_tensorcopy_data)rCrIname loaded_weightparamdistributed_weightrrrrAs"    rAc@sFeZdZdZddZ  ddddZdddZd ddZd!ddZdS)"WeightsUpdatera In-place weight updates for diffusion pipeline modules. Args: pipeline: A ComposedPipelineBase (or DiffusersPipeline) instance whose modules will be updated. The pipeline's model_path attribute is used for rollback on failure. cCs ||_dSr7)r)selfrrrr__init__s zWeightsUpdater.__init__TN model_pathr# flush_cachebooltarget_moduleslist[str] | Noner tuple[bool, str]c Cstd|z||}Wnty.}ztt|dt|fWYd}~Sd}~ww|sId|dtt|j }t|d|fSzt |}Wnt yh}z dd|fWYd}~Sd}~wwt ||\}} | rd| d}t|d|fStd t |d d d d |D|||\} } ttj| r|r|D] \} } t| tr| qt| | | fS)z=Update model weights from disk without restarting the server.zUpdating weights from disk: FNz1No matching modules found for update. Requested: z. Available nn.Module(s): zFailed to download model: z9Cannot update weights: missing weight files for modules: z. No partial updates allowed.z Updating z modules: , css"|] \}}|d|VqdS)z <- Nr)rrprrrr;s z:WeightsUpdater.update_weights_from_disk..)loggerinfo_collect_modulesrKerrorr#listr!rkeysr Exceptionr1lenjoinr_apply_weightsgccollectrcuda empty_cacherrreset_teacache_state)rXrZr[r]r(e error_msgr'r-r.successmessager0r2rrrupdate_weights_from_disks\        z'WeightsUpdater.update_weights_from_diskr)csht|j|durt}nfdd|D}|r)td|dt|}fdd|DS)zResolve target_modules to (name, module) pairs. Raises: ValueError: If target_modules contains names not found in the pipeline. Ncsg|]}|vr|qSrr)rrrrrr6z3WeightsUpdater._collect_modules..z6Module(s) requested for update not found in pipeline: z. Available Module(s): csg|]}||fqSrr)rrSrvrrr6rw)r!rrfrgrK)rXr]namesunknownrrvrrds  zWeightsUpdater._collect_modulesr(r-dict[str, str]c Csg}|D]X\}}zt||}t||||Wqty\}z6||g}tjd|d|dt|d|d|d dd||d d |d|d fWYd }~Sd }~wwd |} ddt|d| dfS)z9Load weights into each module; rollback on first failure.z!Weight update failed for module 'z': z. Rolling back z( module(s) (including partially-loaded 'z'): .T)exc_infoFzFailed to update module 'z.. All modules rolled back to original weights.Nr`zUpdated z modules (z).) r&rHr,rhrbreri _rollbackrj) rXr(r-updated_modulesr/r2rCrq rollback_listrxrrrrks4      zWeightsUpdater._apply_weightsr~ list[str]r4cCsd|sdSt|jj}|D]#}|j|}|durq t||}|s$q tt|}t||q dS)zRestore updated_modules to original weights. If rollback itself fails the exception propagates so the caller knows the model is in an inconsistent state. N) r rrZrrr+r&r#rH)rXr~ original_pathrSr2r"rCrrrr}s     zWeightsUpdater._rollback)TN)rZr#r[r\r]r^r r_)r]r^r r))r(r)r-rzr r_)r~rr r4) __name__ __module__ __qualname____doc__rYrurdrkr}rrrrrWs : rW)r r)r"r#)r'r#r(r)r r*)r2r3r r4)rIr=r r4) r __future__rrlpathlibrrtorch.distributed.tensorrr,sglang.multimodal_gen.runtime.cache.teacacher*sglang.multimodal_gen.runtime.loader.utilsr1sglang.multimodal_gen.runtime.loader.weight_utilsr:sglang.multimodal_gen.runtime.pipelines.diffusers_pipeliner 6sglang.multimodal_gen.runtime.utils.hf_diffusers_utilsr 5sglang.multimodal_gen.runtime.utils.layerwise_offloadr 1sglang.multimodal_gen.runtime.utils.logging_utilsr rrbr!r&r1rHrArWrrrrs( (