U da}@sddlZddlZddlmZmZddlmZddlmZm Z m Z m Z m Z m Z eeZGdddejZddZdd d ZGd d d eZGd ddeZGdddeZGdddeZddZdS)N)corecontext)ops) final_outputNodeTask TaskGroup TaskOutput WorkspaceTypecsNeZdZdZdddZddZddZfd d Zfd d Zd dZ Z S)Jobad A Job defines three TaskGroups: the `init_group`, the `epoch_group` and the `exit_group` which will be run by a JobRunner. The `init_group` will be run only once at startup. Its role is to initialize globally persistent blobs such as model weights, accumulators and data file lists. The `epoch_group` will be run in a loop after init_group. The loop will exit when any of the stop signals added with `add_stop_condition` is True at the end of an epoch. The download_group will be run only once, after all the executions of epoch_group finish. Its role is to collect the distribute scattered parameters back after training. The `exit_group` will be run only once at the very end of the job, the role of this group is to save the results of training in the end of the job. Jobs are context-driven, so that Tasks can be added to the active Job without having to explicitly pass the job object around. Example of usage: def build_reader(partitions): with Job.current().init_group: reader = HiveReader(init_reader, ..., partitions) Task(step=init_reader) with Job.current().epoch_group: limited_reader = ReaderWithLimit(reader, num_iter=10000) data_queue = pipe(limited_reader, num_threads=8) Job.current().add_stop_condition(limited_reader.data_finished()) return data_queue def build_hogwild_trainer(reader, model): with Job.current().init_group: Task(step=model.param_init_net) with Job.current().epoch_group: pipe(reader, processor=model, num_threads=8) with Job.current().exit_group: Task(step=model.save_model_net) with Job() as job: reader = build_reader(partitions) model = build_model(params) build_hogwild_trainer(reader, model) NcCsJ|pttjd|_|pt|_|p&t|_|p2t|_|ploggerinfoLoadr!rC)rnodesretrieve_from_epochr8 path_typetaskZ full_db_namerHrrrinits:  zCheckpointManager.initcCs|js t|jSr)rCr$fetchtolistrrrr blob_lists zCheckpointManager.blob_listc Cslt|dX}ttjg|jd}W5QRX|tt|}W5QRXt||_W5QRX|S)an Build a Task that will measure the time span of checkpoint operations, once operation is done, time can be read from _current_checkpoint_duration. Args: cp_op_name: A string name of the checkpoint operation. add_op: A functor to add the checkpoint operation. Returns: A task with timer. )name)Z counter_name) rrZ task_initZ TimerBeginr=Z task_exitZTimerGetAndEndrrG)rZ cp_op_nameadd_oprUZtimerZtime_span_blobrrr _timed_tasks   zCheckpointManager._timed_taskcCs8|jr"|jr"|jd||j<ntd|jdS)z Add one checkpoint stats into the stats. Args: stats: A dict of checkpoint stats that will be reported. rz&Failed to collect checkpoint stats: {}N)rFrGrWrOrPformat)rstatsrrrcollect_checkpoint_statss z*CheckpointManager.collect_checkpoint_statscsHt|jj|_|pjtdjfdd}d|S)z Build a Task that will be run by JobRunner when the job is to be resumed from a given epoch. This task will run a Load op that will load and deserialize all relevant blobs from a persistent storage. zLoading checkpoints from = %scs tjgjddddS)NTrK)rrQrYrFrrHrrrr[sz&CheckpointManager.load..add_opcheckpoint_load)r9r=r<rFr>rOrPr\)rr2r8rTr[rr`rloads  zCheckpointManager.loadcs<t|jj_tdjfdd}d|S)a Builds a Task that loads only the necessary blobs from a checkpoint of the given epoch. The necessary blobs are given in the blob_names argument. Args: blob_names: A list of strings. Each string is the name of a blob. epoch: The checkpoint epoch to load from. Returns: A Task which loads the specified blobs from the checkpoint of the given epoch. z Load from %scstjgjjddddS)NT)rLrHrMZallow_incomplete)rrQrFr>rr;rrrr[9sz.add_opcheckpoint_partial_loadr9r=r<rFrOrPr\)rr;r2r[rrcrload_blobs_from_checkpoint's z,CheckpointManager.load_blobs_from_checkpointc Cshtdt||j|jt>}td}tjg|gt||j|j|j dd| |W5QRX|S)NzCheck existence of %sFT)r9rHrM) rOrPr9r=r<rrZConstZDBExistsr>Z add_output)rr2rU existencerrrcheck_db_existsDs z!CheckpointManager.check_db_existscCs&i}|||jr"|j||dS)z Report checkpoint operation stats for current node. Args: action_name: A string of the name of checkpoint operation. N)r_r?report)r action_name all_statsrrrreport_checkpoint_statsRs z)CheckpointManager.report_checkpoint_statscs:t|jj_tdjfdd}d|S)z Build a Task that is run once after `init_group` and after each epoch is run. This will execute a Save ops to serialize and persist blobs present in the global workspace. z Saving to %scs tjgjjdddS)NT)rLrHrM)rZSaverYrFr>rrrrr[gsz&CheckpointManager.save..add_opcheckpoint_savere)rr2r[rrrsave^s zCheckpointManager.savecCs|jdk r|jj|ddSz Write metadata for checkpoint Args: epoch: An integer. The epoch-id for which checkpoint metadata is written Nr2r?writerr2rrrwrite_checkpoint_metadataps z+CheckpointManager.write_checkpoint_metadatacCs |}|jdk r|jj|d}|Sa^ Identify the epoch-id from which Job must resume Args: user_epoch: An integer. Optional parameter for user to explicitly identify the epoch-id to load checkpoint from Returns: epoch: the epoch-id to load checkpoints from or None if no checkpoints were written N) user_epochr? last_epochrrvrxrrrget_resume_from_epoch_id{s  z*CheckpointManager.get_resume_from_epoch_idcCsD|r ||_|r||_|jr@|jj|j|jt|jg|j|jddS)RSet parameters associated with CP manager Args: nodes: An array of nodes where this checkpoint manager is running. path_prefix: Used to construct db name or path where checkpoint files are stored. path_type: Indicate the type of path where checkpoint files are stored. r7rHZ node_namesr8rTN)rDrEr? set_paramsr<r>r0r=rrRr8rTrrrr}s  zCheckpointManager.set_paramscCs|jdk r|j|SdSdSaEReturns True if Checkpoint data is accessible Args: epoch: An integer. The epoch of the checkpoint. If None, it implies we need to check if checkpoint directory is accessible Returns: is_cp_accessible: A boolean. Returns True if Checkpoint data is accessible NTr? cp_accessiblersrrrrs  zCheckpointManager.cp_accessible)N)NNNN)NN)N)NN)N)r)r*r+r,rArrVrYr\r_rbrfrhrlrnrtrzr}rrrrrr:s(  &    r:c@s|eZdZdZdddZddZdddZdd d Zd d Zd dZ ddZ ddZ ddZ dddZ dddZd ddZdS)!MultiNodeCheckpointManagera? Coordinates checkpointing and checkpointing across multiple nodes. Each of `init`, `load` and `save` will build TaskGroups which will trigger checkpointing on each of the nodes involved in a distributed job. Args: db_prefix: The prefix used to construct full db name. Since `absolute_path` is set to True, this will be used as db_name in SaveOp. db_type: Type of database to use for storing checkpoint. metadata_handler: An optional object capable of reading/writing checkpoint info in storage of choice. NcCs(d|_||_||_||_d|_d|_dSr)_node_managersr<r>r?rDrE)rr7rHrIrrrrs z#MultiNodeCheckpointManager.__init__c Osl|jdk stdttjF}|jD],\}}t|||f||W5QRXq$|W5QRSQRXdS)Nzinit must be called first.)rr$rr r r)rfuncr kwZ task_groupnodemanagerrrr _task_groups   z&MultiNodeCheckpointManager._task_groupc Cs|jdk r,dd|jD|ks"tttjSg|_|D]>}t|,t|jt||j d}|j ||fW5QRXq6|j tj |g|||dS)NcSsg|] \}}|qSrr.0r_rrr sz3MultiNodeCheckpointManager.init..r7r1rH)rRrSr8rT) rr$rr r rr:r<r0r>r%rrV)rrRrSr8rTrrrrrrVs&   zMultiNodeCheckpointManager.initcCs|jtj|||dS)N)r8rT)rr:rb)rr2r8rTrrrrbs zMultiNodeCheckpointManager.loadc Cs|jdk r$dd|jD|ksntnJg|_|D]>}t|,t|jt||jd}|j||fW5QRXq.|jdk std|jD]f\}}||}| || d } | st dt||j|jdS|||} | | qt d d S) a:Loads the necessary blobs from the checkpoints to the current node. Args: blob_names: A list of strings. Each string is the name of a blob. epoch: An integer. The checkpoint epoch to load from. session: A Session object to execute the Load ops. NcSsg|] \}}|qSrrrrrrrszAMultiNodeCheckpointManager.load_blobs_locally..rzmust initialize node managersrzDB %s does not exist!Fz%Successfully loaded from checkpoints.T)rr$rr:r<r0r>r%rhrunr!rWrOrPr9r=rf) rrRr;r2sessionrrrZexistence_taskrgZ load_taskrrrload_blobs_locallys2       z-MultiNodeCheckpointManager.load_blobs_locallycCs4|jD](\}}t||krt||j|jSqdS)aReturns the DB name of the given node and the given epoch. The DB name is effectively the checkpoint path of the given node and the given epoch. Args: node_name: A string. The node name of interest. epoch: An integer. The epoch of the checkpoint. Returns: checkpoint_db_name: A string. The checkpoint path of the given node and the given epoch. N)rr0r9r=r<)rr1r2rrrrrget_ckpt_db_names z+MultiNodeCheckpointManager.get_ckpt_db_namecCsFi}|jD]\}}||q td||jrB|j||dS)a Report the checkpoint stats for all the nodes, we need to aggregate all the node's stats together so that we know which node's checkpoint operation dominates. Args: action_name: A string of the name of checkpoint operation. zcheckpoint stats: {}N)rr_rOdebugr]r?ri)rrjrkrrrrrrl%s  z2MultiNodeCheckpointManager.report_checkpoint_statscCs|tj|S)z Build a Task that will execute a Save ops to serialize and persist blobs present in the global workspace. )rr:rnrsrrrrn5szMultiNodeCheckpointManager.savecCs|jdk r|jj|ddSrorqrsrrrrt<s z4MultiNodeCheckpointManager.write_checkpoint_metadatacCs |}|jdk r|jj|d}|SrurwryrrrrzGs  z3MultiNodeCheckpointManager.get_resume_from_epoch_idcCsNdd|D|_|r||_|r$||_|jrJ|jj|j|j|j|j|jddS)r{cSsg|] }t|qSrr/)rrrrrr`sz9MultiNodeCheckpointManager.set_params..r|N)Z _node_namesrDrEr?r}r<r>r~rrrr}Ws z%MultiNodeCheckpointManager.set_paramscCs|jdk r|j|SdSdSrrrsrrrrms  z(MultiNodeCheckpointManager.cp_accessible)N)NNN)NN)N)NN)N)r)r*r+r,rrrVrbrrrlrnrtrzr}rrrrrrs    "  rc@seZdZdZddZdS)UploadTaskGroupBuilderz%A simple class to upload checkpoints.cCs tdS)aBuilds the task group to upload checkpoints. Args: epoch: An integer. The checkpoint epoch to be uploaded. checkpoint_manager: Can be a CheckpointManager for single machine or a MultiNodeCheckpointManager for multi-machine. The manager that initializes/saves/loads checkpoints. Raises: NotImplementedError: This base class only has the interface, the implementation will be in the subclasses. N)NotImplementedError)rr2checkpoint_managerrrrbuilds zUploadTaskGroupBuilder.buildN)r)r*r+r,rrrrrr}src@s2eZdZdZd ddZddZddZd d ZdS) JobRunnera Implement the runtime logic for jobs with checkpointing at the level of epoch. Can be used to run either single-host or distributed jobs. Job runner is a callable to be called once from the master, passing a session as an argument. This call will block until the Job execution is complete. If a checkpoint_manager is passed, checkpoints will be taken after initialization and after each epoch execution. If, in addition, `resume_from_epoch` is an epoch number, the corresponding checkpoint will be loaded and job execution will continue from the given epoch. In this case, the job's init_group will not be run. Refer to checkpoint_test.py for an example. NcCs||_||_||_||_dS)a"Initializes the JobRunner. Args: job: A Job object. The job to be executed. checkpoint_manager: Can be a CheckpointManager for single machine or a MultiNodeCheckpointManager for multi-machine. The manager that initializes/saves/loads checkpoints. resume_from_epoch: An integer. The epoch to resume from. upload_task_group_builder: A subclass of the UploadTaskGroupBuilder. Creates a task group to upload checkpoints. N)resume_from_epochrjobupload_task_group_builder)rrrrrrrrrszJobRunner.__init__cCs|jrF|jj|jd|j|j|_|jdk rFtd|j|jdk}|rb| |jj |jrtd| |jj |j|jd|r| d|n.ZStoppingzFinished trainingz"Finished uploading the checkpointsz#Finished downloading the parameterszFinished running the exit group)rr}rrrzrrOrPr]rr rVsave_checkpointsrbrlrranyrrrr)rrZ from_scratchr2rZupload_task_grouprrrtrainsf                  zJobRunner.traincCsF|jstdtd||j|j|||}|jd|S)aPLoads the necessary blobs from the checkpoints. Checkpoints store the snapshots of the workspace in each node. Sometimes we only need to load a subset of the blobs from the checkpoints. One common scenario is to load only the model blobs from the checkpoints for evaluation purpose. Given the names of the necessary blobs, this function goes over all the checkpoints of all the nodes, but only loads the blobs specified in the blob_names to the current workspace. Args: blob_names: A list of strings. Each string is the name of a blob. epoch: An integer. The checkpoint epoch to load from. session: A Session object to execute the load ops. Raises: ValueError: When the checkpoint manager is invalid. Checkpoint manager is Nonez#Loading checkpoint for epoch {} ...rd) r ValueErrorrOrPr]rrrrl)rr;r2rresultrrrload_blobs_from_checkpointss z%JobRunner.load_blobs_from_checkpointsc Cs|jstdzf|jjdd}|rhtd|||j||j|td|j dn t dWn4t k r}zt d||W5d}~XYnXdS) aTriggers operation to save checkpoints This method will trigger the Save ops to serialize and persist the blobs present in the global workspaace. Args: epoch: An integer. The checkpoint epoch-id that we are saving. session: A Session object to execute the save ops. Raises: ValueError: When the checkpoint manager is invalid. rNrpzSaving checkpoints for epoch {}zCheckpoints savedrmz$Checkpoint files cannot be accessed!z1Unable to write checkpoint for epoch {}. Error={}) rrrrOrPr]rrnrtrlwarning Exception)rr2rZ is_accessibleexrrrrs   zJobRunner.save_checkpoints)NNN)r)r*r+r,rrrrrrrrrs Hrc Cs|j,td}|jg|dd}t|dW5QRX|j.td}||}t||dd}W5QRX||dS) z^ Creates a task that will output True when a given number of epochs has finished. Zepoch_counter_initrJ)Z init_count)stepZepoch_countdown)rr!rN) r rr@Z CreateCounterrrZ CountDownr!r()rZ num_epochsZinit_netcounterZ epoch_netfinishedr&rrr epoch_limiter3s   r)N)r4loggingZ caffe2.pythonrrZcaffe2.python.net_builderrZcaffe2.python.taskrrrrr r getLoggerr)rOZManagedr r3r9objectr:rrrrrrrrs"  Z N%