/usr/local/lib64/python3.6/site-packages/caffe2/python/__pycache__
Edit: /usr/local/lib64/python3.6/site-packages/caffe2/python/__pycache__/checkpoint.cpython-36.pyc (27973B)
3
Ege} @ s d dl Z d dlZd dlmZmZ d dlmZ d dlmZm Z m
Z
m Z m Z m
Z
ejeZG dd dejZdd Zdd d
ZG d d d eZG d
d deZG dd deZG dd deZdd ZdS ) N)corecontext)ops) final_outputNodeTask TaskGroup
TaskOutput
WorkspaceTypec sN e Zd ZdZdddZdd Zdd Z fd d
Z fd d Zd
d Z 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)
Nc C sJ |pt tjd| _|pt | _|p&t | _|p2t | _|p
r ZGetAllBlobNamesr5 r9 r8 r: loggerinfoLoadr r? )r nodesretrieve_from_epochr4 path_typetaskZ full_db_namerD r r r init s*
zCheckpointManager.initc C s | j s
t| j j j S )N)r? r! fetchtolist)r r r r blob_list s
zCheckpointManager.blob_listc C sl t |dX}tj tjg | jd}W dQ R X | tj tj|}W dQ R X t|| _W dQ R X |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_nameN) r r Z task_initZ
TimerBeginr9 Z task_exitZTimerGetAndEndr rC )r Z
cp_op_nameadd_oprQ timerZtime_span_blobr r r _timed_task s
z CheckpointManager._timed_taskc C s8 | j r"| jr"| jj d || j < ntjdj| j dS )z
Add one checkpoint stats into the stats.
Args:
stats: A dict of checkpoint stats that will be reported.
r z&Failed to collect checkpoint stats: {}N)rB rC rS rK rL format)r statsr r r collect_checkpoint_stats s
z*CheckpointManager.collect_checkpoint_statsc sH t |jj|_|p j tjdj fdd}jd|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.
z Loading checkpoints from = %sc s t jg j j ddd d S )NT)rG rD rH rI )r rM rU rB r )rD r r r rW s z&CheckpointManager.load..add_opcheckpoint_load)r5 r9 r8 rB r: rK rL rY )r r. r4 rP rW r )rD r r load s
zCheckpointManager.loadc s< t |jj_tjdj fdd}jd|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 %sc s t jg jjddd d S )NT)rG rD rH Zallow_incomplete)r rM rB r: r )r7 r r r rW 9 s z.add_opcheckpoint_partial_load)r5 r9 r8 rB rK rL rY )r r7 r. rW r )r7 r r load_blobs_from_checkpoint' s z,CheckpointManager.load_blobs_from_checkpointc C sh t jdt|| j| j t >}tjd}tjg |gt|| j| j| j dd |j
| W d Q R X |S )NzCheck existence of %sFT)r5 rD rH ) rK rL r5 r9 r8 r r ZConstZDBExistsr: Z
add_output)r r. rQ existencer r r check_db_existsD s
z!CheckpointManager.check_db_existsc C s& i }| j | | jr"| jj|| 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_statsr r r report_checkpoint_statsR s
z)CheckpointManager.report_checkpoint_statsc s: t | j j _tjd j fdd} jd|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 %sc s t j j g j jdd d S )NT)rG rD rH )r ZSaverU rB r: r )r r r rW g s
z&CheckpointManager.save..add_opcheckpoint_save)r5 r9 r8 rB rK rL rY )r r. rW r )r r save^ s zCheckpointManager.savec C s | j dk r| j j|d dS )z
Write metadata for checkpoint
Args:
epoch: An integer. The epoch-id for which checkpoint metadata is
written
N)r. )r; write)r r. r r r write_checkpoint_metadatap s
z+CheckpointManager.write_checkpoint_metadatac C s |}| j dk r | j j|d}|S )a^
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_epoch)r;
last_epoch)r rk rl r r r get_resume_from_epoch_id{ s
z*CheckpointManager.get_resume_from_epoch_idc C sD |r
|| _ |r|| _| jr@| jj| j| jt| jg| j | jd dS )aR Set 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.
)r3 rD
node_namesr4 rP N)r@ rA r;
set_paramsr8 r: r, r9 )r rN r4 rP r r r ro s
z CheckpointManager.set_paramsc C s | j dk r| j j|S dS dS )aE Returns 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
NT)r;
cp_accessible)r r. r r r rp s
z CheckpointManager.cp_accessible)N)NNNN)NN)N)NN)N)r& r' r( r) r= r rR rU rY r\ r^ r` rb rf rh rj rm ro rp r r r r r6 s&
!
r6 c @ s| e Zd ZdZdddZdd Zd ddZd d d
Zd d Zd
d Z dd Z
dd Z dd Z d ddZ
d ddZd 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.
Nc C s( d | _ || _|| _|| _d | _d | _d S )N)_node_managersr8 r: r; r@ rA )r r3 rD rE r r r r s z#MultiNodeCheckpointManager.__init__c O sd | j d k stdttj>}x6| j D ],\}}t| ||f|| W d Q R X q&W |S Q R X d S )Nzinit must be called first.)rr r! r r
r r )r funcr kwZ
task_groupnodemanagerr r r _task_group s
z&MultiNodeCheckpointManager._task_groupc
C s | j d k r,dd | j D |ks"tttjS g | _ xF|D ]>}t|, t| jt|| j d}| j j
||f W d Q R X q8W | j tj |g|||dS )Nc S s g | ] \}}|qS r r ).0ru _r r r
s z3MultiNodeCheckpointManager.init..)r3 r- rD )rN rO r4 rP )
rr r! r r
r r r6 r8 r, r: r" rw rR )r rN rO r4 rP ru rv r r r rR s"
z MultiNodeCheckpointManager.initc C s | j tj|||dS )N)r4 rP )rw r6 r^ )r r. r4 rP r r r r^ s
z MultiNodeCheckpointManager.loadc
C s | j dk r$dd | j D |ksrtnNg | _ xF|D ]>}t|, t| jt|| jd}| j j||f W dQ R X q0W | j dk stdxn| j D ]d\}}|j|}|j | |j
d j } | st j
dt||j|j dS |j||}
|j |
qW t j
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.
Nc S s g | ] \}}|qS r r )rx ru ry r r r rz s zAMultiNodeCheckpointManager.load_blobs_locally..)r3 r- rD z must initialize node managersr zDB %s does not exist!Fz%Successfully loaded from checkpoints.T)rr r! r r6 r8 r, r: r" rb runr rS rK rL r5 r9 r` ) r rN r7 r. sessionru rv ry Zexistence_taskra Z load_taskr r r load_blobs_locally s.
z-MultiNodeCheckpointManager.load_blobs_locallyc C s4 x.| j D ]$\}}t||krt||j|jS qW dS )a Returns 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)rr r, r5 r9 r8 )r r- r. ru rv r r r get_ckpt_db_name s z+MultiNodeCheckpointManager.get_ckpt_db_namec C sJ i }x | j D ]\}}|j| q W tjdj| | jrF| jj|| 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 r\ rK debugrZ r; rc )r rd re ry rv r r r rf % s z2MultiNodeCheckpointManager.report_checkpoint_statsc C s | j tj|S )z
Build a Task that will execute a Save ops to serialize and persist
blobs present in the global workspace.
)rw r6 rh )r r. r r r rh 5 s z MultiNodeCheckpointManager.savec C s | j dk r| j j|d dS )z
Write metadata for checkpoint
Args:
epoch: An integer. The epoch-id for which checkpoint metadata is
written
N)r. )r; ri )r r. r r r rj < s
z4MultiNodeCheckpointManager.write_checkpoint_metadatac C s |}| j dk r | j j|d}|S )a^
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)rk )r; rl )r rk rl r r r rm G s
z3MultiNodeCheckpointManager.get_resume_from_epoch_idc C sN dd |D | _ |r|| _|r$|| _| jrJ| jj| j| j| j | j| jd dS )aR Set 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.
c S s g | ] }t |qS r )r, )rx ru r r r rz ` s z9MultiNodeCheckpointManager.set_params..)r3 rD rn r4 rP N)Z _node_namesr@ rA r; ro r8 r: )r rN r4 rP r r r ro W s z%MultiNodeCheckpointManager.set_paramsc C s | j dk r| j j|S dS dS )aE Returns 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
NT)r; rp )r r. r r r rp m s
z(MultiNodeCheckpointManager.cp_accessible)N)NNN)NN)N)NN)N)r& r' r( r) r rw rR r^ r} r~ rf rh rj rm ro rp r r r r rq s
"
rq c @ s e Zd ZdZdd ZdS )UploadTaskGroupBuilderz%A simple class to upload checkpoints.c C s
t dS )a Builds 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)r r. checkpoint_managerr r r build s
z UploadTaskGroupBuilder.buildN)r& r' r( r) r r r r r r } s r c @ s2 e Zd ZdZd ddZdd Zdd Zd 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.
Nc C s || _ || _|| _|| _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_epochr jobupload_task_group_builder)r r r r r r r r r s zJobRunner.__init__c C s | j rF| j j| jj d | j j| j| _| jdk rFtjdj| j | jdk}|rb|j | jj
| j rtjd |j | j j | jj | jd |r| j d| n.ZStoppingzFinished trainingz"Finished uploading the checkpointsz#Finished downloading the parametersz Finished running the exit group)r ro r r rm r rK rL rZ r{ r
rR save_checkpointsr^ rf r r anyr r r r )r r| Z from_scratchr. r Zupload_task_groupr r r train s\
zJobRunner.trainc C sF | j stdtjdj| | j j| jj |||}| j jd |S )aP Loads 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.
zCheckpoint manager is Nonez#Loading checkpoint for epoch {} ...r_ ) r
ValueErrorrK rL rZ r} r r rf )r r7 r. r| resultr r r load_blobs_from_checkpoints s z%JobRunner.load_blobs_from_checkpointsc C s | j stdyf| j jdd}|rhtjdj| |j| j j| | j j| tjd | j j d n
tj
d W n4 t k
r } ztj
dj|| W Y dd}~X nX dS ) a Triggers 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.
zCheckpoint manager is NoneN)r. z Saving checkpoints for epoch {}zCheckpoints savedrg z$Checkpoint files cannot be accessed!z1Unable to write checkpoint for epoch {}. Error={}) r r rp rK rL rZ r{ rh rj rf warning Exception)r r. r| Z
is_accessibleexr r r r s
zJobRunner.save_checkpoints)NNN)r& r' r( r) r r r r r r r r r s
H r c C s | j , tjd}|jg |d d}t|d W dQ R X | j. tjd}|j|}t||dj d }W dQ R X | j| dS ) z^
Creates a task that will output True when a given
number of epochs has finished.
Zepoch_counter_initrF )Z
init_count)stepNZepoch_countdown)r r r ) r
r r< Z
CreateCounterr r Z CountDownr r% )r Z
num_epochsZinit_netcounterZ epoch_netfinishedr# r r r
epoch_limiter3 s
r )N)r0 loggingZ
caffe2.pythonr r Zcaffe2.python.net_builderr Zcaffe2.python.taskr r r r r r
getLoggerr& rK ZManagedr r/ r5 objectr6 rq r r r r r r r s"
Z
N %