BVhQBdZddlmZddlmZddlmZddlmZddlmZddlm Z ddlm Z dd lm Z dd lm Z dd lm Z dd lmZdd lmZddlmZddlmZddlmZddlmZddlmZedgGddej4ZGddej8Zy)z"Synchronize replicas for training.)distribute_lib)indexed_slices)ops)tensor) array_ops)control_flow_ops) data_flow_ops) state_ops) variable_v1) variables) tf_logging) optimizer) queue_runner)session_manager)session_run_hook) deprecation) tf_exportztrain.SyncReplicasOptimizer)v1ceZdZdZej ddd dfd ZdZddZd Z d Z d Z d Z dd Z ddZxZS)SyncReplicasOptimizeraClass to synchronize, aggregate gradients and pass them to the optimizer. This class is deprecated. For synchronous training, please use [Distribution Strategies](https://github.com/tensorflow/tensorflow/tree/master/tensorflow/contrib/distribute). In a typical asynchronous training environment, it's common to have some stale gradients. For example, with a N-replica asynchronous training, gradients will be applied to the variables N times independently. Depending on each replica's training speed, some gradients might be calculated from copies of the variable from several steps back (N-1 steps on average). This optimizer avoids stale gradients by collecting gradients from all replicas, averaging them, then applying them to the variables in one shot, after which replicas can fetch the new variables and continue. The following accumulators/queue are created: * N `gradient accumulators`, one per variable to train. Gradients are pushed to them and the chief worker will wait until enough gradients are collected and then average them before applying to variables. The accumulator will drop all stale gradients (more details in the accumulator op). * 1 `token` queue where the optimizer pushes the new global_step value after all variables are updated. The following local variable is created: * `sync_rep_local_step`, one per replica. Compared against the global_step in each accumulator to check for staleness of the gradients. The optimizer adds nodes to the graph to collect gradients and pause the trainers until variables are updated. For the Parameter Server job: 1. An accumulator is created for each variable, and each replica pushes the gradients into the accumulators instead of directly applying them to the variables. 2. Each accumulator averages once enough gradients (replicas_to_aggregate) have been accumulated. 3. Apply the averaged gradients to the variables. 4. Only after all variables have been updated, increment the global step. 5. Only after step 4, pushes `global_step` in the `token_queue`, once for each worker replica. The workers can now fetch the global step, use it to update its local_step variable and start the next batch. Please note that some workers can consume multiple minibatches, while some may not consume even one. This is because each worker fetches minibatches as long as a token exists. If one worker is stuck for some reason and does not consume a token, another worker can use it. For the replicas: 1. Start a step: fetch variables and compute gradients. 2. Once the gradients have been computed, push them into gradient accumulators. Each accumulator will check the staleness and drop the stale. 3. After pushing all the gradients, dequeue an updated value of global_step from the token queue and record that step to its local_step variable. Note that this is effectively a barrier. 4. Start the next batch. ### Usage ```python # Create any optimizer to update the variables, say a simple SGD: opt = GradientDescentOptimizer(learning_rate=0.1) # Wrap the optimizer with sync_replicas_optimizer with 50 replicas: at each # step the optimizer collects 50 gradients before applying to variables. # Note that if you want to have 2 backup replicas, you can change # total_num_replicas=52 and make sure this number matches how many physical # replicas you started in your job. opt = tf.compat.v1.train.SyncReplicasOptimizer(opt, replicas_to_aggregate=50, total_num_replicas=50) # Some models have startup_delays to help stabilize the model but when using # sync_replicas training, set it to 0. # Now you can call `minimize()` or `compute_gradients()` and # `apply_gradients()` normally training_op = opt.minimize(total_loss, global_step=self.global_step) # You can create the hook which handles initialization and queues. sync_replicas_hook = opt.make_session_run_hook(is_chief) ``` In the training program, every worker will run the train_op as if not synchronized. ```python with training.MonitoredTrainingSession( master=workers[worker_id].target, is_chief=is_chief, hooks=[sync_replicas_hook]) as mon_sess: while not mon_sess.should_stop(): mon_sess.run(training_op) ``` NzThe `SyncReplicaOptimizer` class is deprecated. For synchronous training, please use [Distribution Strategies](https://github.com/tensorflow/tensorflow/tree/master/tensorflow/contrib/distribute).T) warn_oncec||}tt| ||tjd||||_||_d|_||_||_ ||_ t|||_ d|_ d|_d|_g|_y)aConstruct a sync_replicas optimizer. Args: opt: The actual optimizer that will be used to compute and apply the gradients. Must be one of the Optimizer classes. replicas_to_aggregate: number of replicas to aggregate for each variable update. total_num_replicas: Total number of tasks/workers/replicas, could be different from replicas_to_aggregate. If total_num_replicas > replicas_to_aggregate: it is backup_replicas + replicas_to_aggregate. If total_num_replicas < replicas_to_aggregate: Replicas compute multiple batches per update to variables. variable_averages: Optional `ExponentialMovingAverage` object, used to maintain moving averages for the variables passed in `variables_to_average`. variables_to_average: a list of variables that need to be averaged. Only needed if variable_averages is passed in. use_locking: If True use locks for update operation. name: string. Optional name of the returned operation. Nz?SyncReplicasV2: replicas_to_aggregate=%s; total_num_replicas=%sF)superr__init__logginginfo_opt_replicas_to_aggregate_gradients_applied_variable_averages_variables_to_average_total_num_replicasmax_tokens_per_step _global_step_sync_token_queue_chief_queue_runner_accumulator_list) selfoptreplicas_to_aggregatetotal_num_replicasvariable_averagesvariables_to_average use_lockingname __class__s b/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/training/sync_replicas_optimizer.pyrzSyncReplicasOptimizer.__init__sD!0 / TB LLI13DI"7D#D/D!5D1D 24IJDD!D $D  Dc:|jj|i|S)a`Compute gradients of "loss" for the variables in "var_list". This simply wraps the compute_gradients() from the real optimizer. The gradients will be aggregated in the apply_gradients() so that user can modify the gradients like clipping with per replica global norm if needed. The global norm with aggregated gradients can be bad as one replica's huge gradients can hurt the gradients from other replicas. Args: *args: Arguments for compute_gradients(). **kwargs: Keyword arguments for compute_gradients(). Returns: A list of (gradient, variable) pairs. )rcompute_gradientsr)argskwargss r2r5z'SyncReplicasOptimizer.compute_gradientss '499 & & 7 77r3c V |s td| td||_g}g}g}tj}t j }|j j|5tjddtjjg|jjd|_dddt!j"|j||_|j$g} t'j(t'j*|_tj.d|j05|D]\} } |j3| tj4| j45| |j3d dddUt7| t8j:rt=j>| j| jA| jBdz } |j3| jE| |j |j3| jG|jHnt7| tJjLs td t=jN| jd | jBdz } |j3| jQ| |j |j3| jS|jH|jTj3| | j4fdddtW||} tj4|j45tj.d 5|jXj[| |}ddddddtj4|j45tj.d 5t=j\d|jjd dd}||_/ddddddtj4|j45tj.d 5tj`|5jc}dddt!j"|j}tj`g5tejf|jhg|}jk|f}ddd|jlbtj`g5tj.d 5|jljo|jp}ddddddtsjtg|_;dddddd|jTD]E\}}tj4|5| j3|jy|ddddGtjz| |_>d|_?cdddS#1swY@xYw#1swYxYw#1swYxYw#1swYxYw#1swY3xYw#1swY8xYw#1swYxYw#1swY|xYw#1swY#xYw#1swY(xYw#1swYxYw#1swYxYw#1swYxYw#1swYyxYw)a(Apply gradients to variables. This contains most of the synchronization implementation and also wraps the apply_gradients() from the real optimizer. Args: grads_and_vars: List of (gradient, variable) pairs as returned by compute_gradients(). global_step: Optional Variable to increment by one after the variables have been updated. name: Optional name for the returned operation. Default to the name passed to the Optimizer constructor. Returns: train_op: The op to dequeue a token so the replicas can exit this batch and start the next one. This is executed by each replica. Raises: ValueError: If the grads_and_vars is empty. ValueError: If global step is not provided, the staleness cannot be checked. z!Must supply at least one variableNz*Global step is required to check stalenessrFsync_rep_local_step) initial_value trainable collectionsdtyper0z /grad_accum)shape shared_name) local_stepzUnknown grad type! sync_token_q)shapesr0r@ SetGlobalStepr0T)@ ValueErrorr%rno_opr get_strategyextendedcolocate_vars_withr VariableV1r GraphKeysLOCAL_VARIABLESr> base_dtype _local_stepr assignlocal_step_init_opr report_uninitialized_variablesglobal_variablesready_for_local_init_op name_scope_nameappenddevice isinstancerTensorr ConditionalAccumulator get_shaper0 apply_grad take_gradrr IndexedSlicesSparseConditionalAccumulatorapply_indexed_slices_gradtake_indexed_slices_gradr(ziprapply_gradients FIFOQueuer&control_dependenciesdequeuerfillr$ enqueue_manyr applyr!r QueueRunnerr'set_global_stepgroup chief_init_opr)r)grads_and_vars global_stepr0 train_opsaggregated_gradvar_list local_anchordistribution_strategychief_init_opsgradvar grad_accumaggregated_grads_and_vars update_opsync_token_queuetokentrain_optokenssync_opaccumdevs r2rgz%SyncReplicasOptimizer.apply_gradientss.  : ;; C DD#DIOH$))+L*779  ' ' : :< H&$//}}445!!,,$ &d&(..t/?/?MD--.N#,#K#K""$$&D  djj )H%B)$ ZZ # B \  " "4 (   B B $ .&== mmoHH}46J   Z22!1!133 4  " ":#7#7++$- .dN$@$@A344&CC "#((]2JLJ   ZAA!1!1B3 4  " ":#F#F++$- .  ' 'SZZ(@ A1 B BB8#&ox"@ ::k(( );3>>"+=;II--.G.9; ;; ::k(( )23>>"+=2  # #B$/$5$5$@$@+-)70>  @  "222 ::k(( ))3>>"+=)  % %i 0 -"**,% -##D$4$4e<  % %yk 2 =>>4#8#8"9;G&$116)<'  =  " " .'' 2 ,CNN24F ,--33**,G , ,$0#;#; wi$) #))&..6*% ZZ_ 6   ##O$5 6 6 66 ,11NDd $d QHH&&  B B:;;;; 2222 - - = = , , , ,))))( 6 6EHHsF+A W6>:Z8X ZE?X6Z X!X>X'Z-X76X*9X7'Z(Z>Y8Y %>Y8#4Y *Y8Y+ &Y =Y+ $Y8)Z1/Z #Z)Z6XX ZX XX' "Z*X4 /X77Y <ZY Y8Y Y8Y(#Y+ +Y5 0Y88Z =ZZ ZZ ZZ(cL|jdur td|jS)aReturns the QueueRunner for the chief to execute. This includes the operations to synchronize replicas: aggregate gradients, apply to variables, increment global step, insert tokens to token queue. Note that this can only be called after calling apply_gradients() which actually generates this queuerunner. Returns: A `QueueRunner` for chief to execute. Raises: ValueError: If this is called before apply_gradients(). Fz)Should be called after apply_gradients().)rrIr'r)s r2get_chief_queue_runnerz,SyncReplicasOptimizer.get_chief_queue_runnerTs* %' B CC  # ##r3c:|jj|i|S)a=Return a slot named "name" created for "var" by the Optimizer. This simply wraps the get_slot() from the actual optimizer. Args: *args: Arguments for get_slot(). **kwargs: Keyword arguments for get_slot(). Returns: The `Variable` for the slot if it was created, `None` otherwise. )rget_slotr6s r2rzSyncReplicasOptimizer.get_sloths  499  t .v ..r3c6|jjS)zFetches a list of optimizer variables in the default graph. This wraps `variables()` from the actual optimizer. It does not include the `SyncReplicasOptimizer`'s local step. Returns: A list of variables. )rr rs r2r zSyncReplicasOptimizer.variablesvs 99    r3c:|jj|i|S)aReturn a list of the names of slots created by the `Optimizer`. This simply wraps the get_slot_names() from the actual optimizer. Args: *args: Arguments for get_slot(). **kwargs: Keyword arguments for get_slot(). Returns: A list of strings. )rget_slot_namesr6s r2rz$SyncReplicasOptimizer.get_slot_namess  $499 # #T 4V 44r3c>|jdur td|j|jz }|dk(r |j}n||krtd||fz|dkDrt j |j j 5t jd5tj|g|j }|jj|f}ddddddStjd }|S#1swY+xYw#1swYSxYw) aReturns the op to fill the sync_token_queue with the tokens. This is supposed to be executed in the beginning of the chief/sync thread so that even if the total_num_replicas is less than replicas_to_aggregate, the model can still proceed as the replicas can compute multiple steps per variable update. Make sure: `num_tokens >= replicas_to_aggregate - total_num_replicas`. Args: num_tokens: Number of tokens to add to the queue. Returns: An op for the chief/sync replica to fill the token queue. Raises: ValueError: If this is called before apply_gradients(). ValueError: If num_tokens are smaller than replicas_to_aggregate - total_num_replicas. Fz>get_init_tokens_op() should be called after apply_gradients().rDzBToo few tokens to finish the first step: %d (given) vs %d (needed)rrCNno_init_tokensrH)rrIrr"rr[r%rXrrkr&rlrrJ)r) num_tokens tokens_neededr init_tokenss r2get_init_tokens_opz(SyncReplicasOptimizer.get_init_tokens_ops%( %'  J LL//$2J2JJMR..j m #  N } % & ''A~ ::d''.. /E1CE d.?.?@,,996)D EE %**0@Ak  EEE s$D>DDD DDct|||S)zECreates a hook to handle SyncReplicasHook ops such as initialization.)_SyncReplicasOptimizerHook)r)is_chiefrs r2make_session_run_hookz+SyncReplicasOptimizer.make_session_run_hooks %dHj AAr3)NNNF sync_replicas)NN)rD)__name__ __module__ __qualname____doc__r deprecatedrr5rgrrr rrr __classcell__)r1s@r2rr)sw\|;  J #'!%$( #5  5 n8${z$( / ! 5'RBr3rc"eZdZdZdZdZdZy)rz>A SessionRunHook handles ops related to SyncReplicasOptimizer.c.||_||_||_y)aCreates hook to handle SyncReplicasOptimizer initialization ops. Args: sync_optimizer: `SyncReplicasOptimizer` which this hook will initialize. is_chief: `Bool`, whether is this a chief replica or not. num_tokens: Number of tokens to add to the queue. N)_sync_optimizer _is_chief _num_tokens)r)sync_optimizerrrs r2rz#_SyncReplicasOptimizerHook.__init__s*DDN!Dr3c|jjdur td|jr|jj|_|jj |_|jj|_ |jj|j|_ y|jj|_|jj |_d|_ d|_ y)NFzLSyncReplicasOptimizer.apply_gradient should be called before using the hook.)rrrIrrq_local_init_oprW_ready_for_local_init_opr _q_runnerrr_init_tokens_oprTrs r2beginz _SyncReplicasOptimizerHook.begins ..%7    ~~ 00>>d    6 6 #++BBDdn!11DD   d!00CCd    6 6 #dn!dr3cvtj|j|d\}}|s%td|jj d||j |j|j|j |j|j |jj||ddyy)z.Runs SyncReplicasOptimizer initialization ops.z8Model is not ready for SyncReplicasOptimizer local init.zXInit operations did not make model ready for SyncReplicasOptimizer local_init. Init op: z , error: NT)coorddaemonstart) r_readyr RuntimeErrorrr0runrrcreate_threads)r)sessionrlocal_init_successmsgs r2after_create_sessionz/_SyncReplicasOptimizerHook.after_create_sessions-44 %%wBD      # #S * ++ KK##$ ' kk$&&' ~~! nn## t4$9"r3N)rrrrrrrrBr3r2rrsF ""&9r3rN)rtensorflow.python.distributertensorflow.python.frameworkrrrtensorflow.python.opsrrr r r r tensorflow.python.platformr rtensorflow.python.trainingrrrrtensorflow.python.utilr tensorflow.python.util.tf_exportr OptimizerrSessionRunHookrrBr3r2rs )76+.+2/+-+<0367.6 ,-.PBI//PB/PBf 19!1!@!@19r3