o civ)@sddlZddlZddlZddlZddlmZmZmZmZm Z m Z ddl Z ddl m Z ddlmZddlmZddlmZddlmZmZmZmZmZmZddlmZmZdd lmZm Z dd l!m"Z"dd l#m$Z$m%Z%m&Z&m'Z'dd l(m)Z)dd l*m+Z+ddl,m-Z-m.Z.ddl/m0Z0ddl1m2Z2ddl3m4Z4m5Z5m6Z6m7Z7ddl8m9Z9ddl:m;Z;ddlm?Z?ddl@mAZAddlBmCZCmDZDddlEmFZFddlGmHZHmIZIddlJmKZKeLeMZNeIGdddZOdS)N)AnyCallableDictListOptionalUnion)env_bool) usage_lib) ActorHandle)tag_train_v2_trainer) BackendConfig Checkpoint DataConfigResult RunConfig ScalingConfig)+_RESUME_FROM_CHECKPOINT_DEPRECATION_WARNING$_TRAINER_RESTORE_DEPRECATION_WARNING)RAY_CHDIR_TO_TRIAL_DIRRAY_TRAIN_ENABLE_STATE_TRACKING)!_GET_METADATA_DEPRECATION_MESSAGE)AcceleratorSetupCallbackBackendSetupCallbackDatasetsSetupCallbackWorkingDirectorySetupCallback) GenDataset)_initialize_env_callbacks)ControllerMetricsCallbackWorkerMetricsCallback)StateManagerCallback)UserCallbackHandler)DEFAULT_RUN_CONTROLLER_AS_ACTORMETRICS_ENABLED_ENV_VARRUN_CONTROLLER_AS_ACTOR_ENV_VARget_env_vars_to_propagate)RayTrainCallback)TrainRunContext)TrainController)create_failure_policy)create_scaling_policy)ObjectRefWrapperconstruct_train_func) UserCallback) Deprecated DeveloperAPI)NodeAffinitySchedulingStrategyc@seZdZdZddddddddddeegdfeegdffdeedeedee dee d eee e fd ee d eed eee effd dZdefddZdeefddZdefddZdeefddZeeddZeeddZdS)DataParallelTraineraBase class for distributed data parallel training on Ray. This class supports the SPMD parallelization pattern, where a single training function is executed in parallel across multiple workers, and different shards of data are processed by each worker. N)train_loop_configbackend_configscaling_config run_configdatasetsdataset_configresume_from_checkpointmetadatatrain_loop_per_workerr1r2r3r4r5r6r7r8c Cs|pt|_||_||_|pt|_|pt|_|pi|_|p!t |_ t |j|j|j|j|j|j d|_ |durs  zADataParallelTrainer._create_default_callbacks..)user_callbacksr<cSsg|] }t|ts|qSr@rRrTr@r@rArWs)rrr2r3rrr5r;extendrrrappendr"rrr<rrr4rJr )r?rJaccelerator_setup_callbackbackend_setup_callbackdatasets_setup_callback working_directory_setup_callbackrun_config_callbacksrXr@r@rArMsL        z-DataParallelTrainer._create_default_callbackscKsttt}|r9tjdttdddtidt }|jdi|}| |t |j t |j St di|}t | | S)NrF)node_idsoftenv_vars)num_cpusscheduling_strategy runtime_envr@)rr#r!rayremoter/get_runtime_context get_node_idr$r'_register_sigint_handlergetrun get_resultasyncio)r?controller_init_kwargsrun_controller_as_actorcontroller_actor_cls controllerr@r@rArKs(  z2DataParallelTrainer._initialize_and_run_controllerrrcs$dfdd}ttj|dS)z=Register SIGINT handler so user Ctrl C gracefully aborts run.rcsttdd7dkrtdtddkr8z tjWdStjj y7tdYdSwdS)NuReceived SIGINT. Gracefully aborting the training run — this may take a few seconds. To forcefully abort immediately, you can send a different signal, such as SIGKILL.zGReceived SIGINT at least 3 times. Forcefully aborting the training run.r) loggerinfosysexitrfrkabortrg exceptionsActorDiedError)signumframerr sigint_countr@rAsigint_handlers  zDDataParallelTrainer._register_sigint_handler..sigint_handlerN)signalSIGINT)r?rrrr@r~rArjsz,DataParallelTrainer._register_sigint_handlercOtt)z[Deprecated] Restores a Train experiment from a previously interrupted/failed run. This method is deprecated and will be removed in a future release. r=rclsargskwargsr@r@rArestorezDataParallelTrainer.restorecOr)z[Deprecated] Checks if a Train experiment can be restored from a previously interrupted/failed run. This method is deprecated and will be removed in a future release. rrr@r@rA can_restorerzDataParallelTrainer.can_restore)__name__ __module__ __qualname____doc__rrrrr rrstrrrr rrBrrQrr%rMrKr r'rj classmethodr-rrr@r@r@rAr0;sN    (%4 r0)Prnloggingrrwtypingrrrrrrrfray._private.ray_constantsrray._private.usager ray.actorr ray.air._internal.usager ray.trainr r rrrrray.train.base_trainerrrray.train.constantsrrray.train.contextr ray.train.v2._internal.callbacksrrrr)ray.train.v2._internal.callbacks.datasetsr-ray.train.v2._internal.callbacks.env_callbackr(ray.train.v2._internal.callbacks.metricsrr.ray.train.v2._internal.callbacks.state_managerr.ray.train.v2._internal.callbacks.user_callbackr ray.train.v2._internal.constantsr!r"r#r$)ray.train.v2._internal.execution.callbackr%(ray.train.v2._internal.execution.contextr&+ray.train.v2._internal.execution.controllerr'1ray.train.v2._internal.execution.failure_handlingr(/ray.train.v2._internal.execution.scaling_policyr)ray.train.v2._internal.utilr*r+ray.train.v2.api.callbackr,ray.util.annotationsr-r.ray.util.scheduling_strategiesr/ getLoggerrrur0r@r@r@rAsB