BVh{dZddlZddlZddlZddlZddlZddlmZddlm Z ddlm Z ddlm Z ddl mZddl mZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZddlmZddlmZddl m!Z!ddl"m#Z#ddl$m%Z%ddl&m'Z'dZ(dZ)dZ*dZ+dZ,dZ-dZ.dZ/e%dgGd d!e0Z1Gd"d#e1Z2Gd$d%e1Z3Gd&d'e1Z4Gd(d)e1Z5d*Z6e%d+gGd,d-e0Z7e7Z8y).axModule for `PreemptionCheckpointHandler`. This is currently under development and the API is subject to change. PreemptionCheckpointHandler reduces loss of training progress caused by termination (preemption or maintenance) of workers in multi-worker synchronous training and avoid surfacing an error indistinguishable from application errors to the job scheduler or users. N)gen_check_preemption_op) checkpoint)checkpoint_context)checkpoint_management)distribute_lib)multi_worker_util)failure_handling_util)context) constant_op)dtypes)errors)file_io) variables)gfile) tf_logging) tf_contextlib) deprecated) tf_export) doc_controlsRUN_TO_CHECKPOINTLAST_RUN_TO_CHECKPOINTTERMINATED_WORKERRECEIVED_SIGNALcheckpointed_runs STOP_WATCHER TF_DEFAULT_PREEMPTION_NOTICE_KEYcDtjj|}tjj|}dt |z}tjj ||}t j|tjj ||S)zz0BorgTerminationConfig.__init__..schhrlr-rr2)r8r3r4r5r6default_exit_fns r+r9zBorgTerminationConfig.__init__s1 #9D*O-oDL$)DDLr-r:r;r@r-r+rLrLs " r-rLc eZdZdZ ddZy)BorgTPUTerminationConfigrMNch||_|xstj|_|xsd|_||_yrD)r3r default_tpu_exit_fnr4r5r6r7s r+r9z!BorgTPUTerminationConfig.__init__s3 #9DG3GGDL$)DDLr-r:r;r@r-r+rVrVs " r-rVcr|s t}|tjjur6t |j |j |j|jS|tjjur6t|j |j |j|jS|tjjur6t|j |j |j|jSt|j |j |j|jS)zAComplete un-filled fields of TerminationConfig based on platform.)r0r PlatformDeviceGCE_GPUrBr3r4r5r6GCE_CPUrI INTERNAL_TPUrVrL)platform_devicetermination_configs r+ _complete_config_for_environmentr`s *,-<<DDD "#5#L#L#5#=#=#5#B#B#5#=#= ?? />>FFF "#5#L#L#5#=#=#5#B#B#5#=#= ?? />>KKK #$6$M$M$6$>$>$6$C$C$6$>$> @@ !11""$6$C$C"" $$r-z3distribute.experimental.PreemptionCheckpointHandlerceZdZdZ ddZdZdZdZdZdZ d Z d Z d Z d Z d Zeeddej$dZdZdZdZdZej2dZdZdZdZdZdZdZ y)PreemptionCheckpointHandlera8Preemption and error handler for synchronous training. Note: This API only supports use with `tf.distribute.MultiWorkerMirroredStrategy` and `tf.distribute.TPUStrategy`. A `PreemptionCheckpointHandler` coordinates all workers to save a checkpoint upon receiving a preemption signal. It also helps disseminate application error messages accurately among the cluster. When a `PreemptionCheckpointHandler` object is created, it restores values from the latest checkpoint file if any exists. Right after the initialization, the object starts to watch out for termination signal for any member in the cluster. If receiving a signal, the next time the worker executes `PreemptionCheckpointHandler.run`, the `PreemptionCheckpointHandler` will align all workers to save a checkpoint. Then, if an `exit_fn` is configured via `tf.distribute.experimental.TerminationConfig`, it will be invoked. Otherwise, the process will simply exit and later the platform should restart it. Note: We advise users of `tf.distribute.MultiWorkerMirroredStrategy` who choose to configure their own `exit_fn` in `tf.distribute.experimental.TerminationConfig` to include a `sys.exit(CODE_OR_MESSAGE)` in the `exit_fn` so that after the restart, all workers can initialize communication services correctly. For users of `tf.distribute.TPUStrategy`, if they do not wish to do a cluster restart but would like an in-process restart (i.e., keep the coordinator alive and re-do the steps to connect to cluster, initialize TPU system, and make the `TPUStrategy` object), they could configure the `exit_fn` to a no-op. For users of `tf.distribute.MultiWorkerMirroredStrategy`, the core API is `PreemptionCheckpointHandler.run`: ```python strategy = tf.distribute.MultiWorkerMirroredStrategy() trained_epoch = tf.Variable(initial_value=tf.constant(0, dtype=tf.dtypes.int64), name='epoch') step_in_epoch = tf.Variable(initial_value=tf.constant(0, dtype=tf.dtypes.int64), name='step_in_epoch') with strategy.scope(): dataset, model, optimizer = ... checkpoint = tf.train.Checkpoint(optimizer=optimizer, model=model, trained_epoch=trained_epoch, step_in_epoch=step_in_epoch) preemption_checkpoint_handler = tf.distribute.experimental.PreemptionCheckpointHandler(cluster_resolver, checkpoint, checkpoint_dir) while trained_epoch.numpy() < NUM_EPOCH: while step_in_epoch.numpy() < STEPS_PER_EPOCH: # distributed_train_function contains a call to strategy.run. loss += preemption_checkpoint_handler.run(distributed_train_function, args=(next(iterator),)) # For users of MultiWorkerMirroredStrategy, usually # STEPS_PER_TRAIN_FUNCTION = 1. step_in_epoch.assign_add(STEPS_PER_TRAIN_FUNCTION) ... epoch.assign_add(1) step_in_epoch.assign(0) ``` For users of `tf.distribute.TPUStrategy`, the core APIs are `PreemptionCheckpointHandler.run` and `PreemptionCheckpointHandler.watch_preemption_scope`: ```python strategy = tf.distribute.TPUStrategy(tpu_cluster_resolver) # Rest of TPU init omitted, see documentation for TPUSTrategy. with preemption_checkpoint_handler.watch_preemption_scope(): while trained_epoch.numpy() < NUM_EPOCH: while step_in_epoch.numpy() < STEPS_PER_EPOCH: # distributed_train_function contains a call to strategy.run. loss += preemption_checkpoint_handler.run(distributed_train_function, args=(next(iterator),)) # For users of TPUStrategy, usually STEPS_PER_TRAIN_FUNCTION >> 1 since # clustering multiple steps within a tf.function amortizes the overhead # of launching a multi-device function on TPU Pod. step_in_epoch.assign_add(STEPS_PER_TRAIN_FUNCTION) ... epoch.assign_add(1) step_in_epoch.assign(0) ``` Not all interruptions come with advance notice so that the `PreemptionCheckpointHandler` can handle them, e.g., those caused by hardware failure. For a user who saves checkpoints for these cases themselves outside the `PreemptionCheckpointHandler`, if they are using a `tf.train.CheckpointManager`, pass it as the `checkpoint_or_checkpoint_manager` argument to the `PreemptionCheckpointHandler`. If they do not have a `tf.train.CheckpointManager` but are directly working with `tf.train.Checkpoint`, we advise saving the checkpoints in the directory that's passed as the `checkpoint_dir` argument. In this way, at the program beginning, `PreemptionCheckpointHandler` can restore the latest checkpoint from the directory, no matter it's saved by the user themselves or saved by the `PreemptionCheckpointHandler` before preemption happens. **A note on the platform:** `PreemptionCheckpointHandler` can only handle the kind of termination with advance notice. For now, the API recognizes the termination signal for CPU, GPU, and TPU on Google Borg and CPU and GPU on the Google Cloud Platform. In these cases, `PreemptionCheckpointHandler` will automatically adopt the correct preemption/maintenance notification detection mechanism. Users of other platforms can configure a detection monitoring behavior through the `tf.distribute.experimental.TerminationConfig`. Customization for the exit behavior and grace period length could also be done here. Ncft|tjr|stjd||_||_||_||_tj|_ t|j|j }|j|_|j|_|j"|_|j&|_d|_|jtj,j.tj,j0fvrt3j4dnu|jtj,j6k(r|j9n=|r+d|j;j=vr t?d|jAt3jBdy)a:Creates the `PreemptionCheckpointHandler`. Args: cluster_resolver: a `tf.distribute.cluster_resolver.ClusterResolver` object. You may also obtain it through the `cluster_resolver` attribute of the distribution strategy in use. checkpoint_or_checkpoint_manager: a `tf.train.CheckpointManager` or a `tf.train.Checkpoint`. If you are using a `tf.train.CheckpointManager` to manage checkpoints outside the `PreemptionCheckpointHandler` for backup purpose as well, pass it as `checkpoint_or_checkpoint_manager` argument. Otherwise, pass a `tf.train.Checkpoint` and the `PreemptionCheckpointHandler` will create a `tf.train.CheckpointManager` to manage it in the `checkpoint_dir`. checkpoint_dir: a directory where the `PreemptionCheckpointHandler` saves and restores checkpoints. When a `PreemptionCheckpointHandler` is created, the latest checkpoint in the `checkpoint_dir` will be restored. (This is not needed if a `tf.train.CheckpointManager` instead of a `tf.train.Checkpoint` is passed as the `checkpoint_or_checkpoint_manager` argument.) termination_config: optional, a `tf.distribute.experimental.TerminationConfig` object to configure for a platform other than Google Borg or GCP. zEWhen a checkpoint is passed, a checkpoint_dir must be passed as well.TzQPreemptionCheckpointHandler does not support usage with TPU or CPU device on GCP.pszjPreemptionCheckpointHandler does not supportusage with tf.distribute.experimental.ParameterServerStrategy.z4PreemptionCheckpointHandler initialized or restored.N)" isinstancecheckpoint_lib Checkpointr InvalidArgumentError_cluster_resolver_termination_config!_checkpoint_or_checkpoint_manager_checkpoint_dirr detect_platform_platform_devicer`r3_termination_watcher_fnr4_exit_fnr5 _grace_periodr6_save_fn _local_moderZGCE_TPUr\loggingwarningr]_initialize_for_tpu_strategy cluster_specas_dictNotImplementedError2_initialize_for_mirrored_and_multi_worker_mirroredinfo)r8cluster_resolver checkpoint_or_checkpoint_managerr&r_completed_termination_configs r+r9z$PreemptionCheckpointHandler.__init__s<2 ++-5C  ' '), --.D1D-MD*)D1AACD#C t77$  %;;  188DM5BBD088DMD ,,44,,44!oo &  / / < < = '') d&6&C&C&E&M&M&OO! M   ==? LLGHr-cd|_d|_d|_|j|jj d|_y)z*>*.*@*@*H*HJ   &r-ctjdtjtj|jy)NzStart watcher for local signal.)rur|signalSIGTERM_sigterm_handler_fnrs r+rz6PreemptionCheckpointHandler._start_watching_for_signals' LL23 MM&..$":":;r-ctj|_tj|jd|j zd|_tjd|j jy)Nz!WorkerTerminationSignalWatcher-%sTrz%Start polling for termination signal.) rr+_poll_termination_signal_thread_should_stopr_poll_termination_signalrrrur|rrs r+rzAPreemptionCheckpointHandler._start_polling_for_termination_signals_7@7HD4+4+;+;,, 043F3F F,D( LL89((..0r-c |jjs |jry|jrnt j dN|j y)z:Poll maintenance notice and notify peers if receiving one.Nr)ris_setrrotimesleep_maybe_set_received_own_sigtermrs r+rz4PreemptionCheckpointHandler._poll_termination_signalsR  9 9 @ @   + + % % '  jjm  ((*r-cB|jrTtjd|jt j|_|j jy tjjt|jtjd|jt j|_|j jy#tj$r#tjd|jYywxYw)zr-c"t|ddr tjjtt tjjtt|jjd|_ tjdyy#t j t jf$rYt$r+}tjdt|zYd}~d}~wwxYw#t j t jf$rYt$r+}tjdt|zYd}~d}~wwxYw#|jjd|_ tjdwxYw)z0Stop the thread that is _watch_step_to_save_key.rNzQIgnoring error when shutting down _stop_cluster_wise_termination_watcher_thread: z0Shut down watcher for peer's termination signal.)rr r_INITIAL_RUN_COUNT_KEY_STOP_WATCHING_CLUSTER_VALUEr rUnavailableError Exceptionrur|r#_FINAL_RUN_COUNT_KEYrr$)r8es r+-_stop_cluster_wise_termination_watcher_threadzIPreemptionCheckpointHandler._stop_cluster_wise_termination_watcher_threadsZt?FQ.. "$@  J../C/K M 55::<8<5 HI=G '')@)@ A Q  GILQP Q QQ'')@)@ A  Q GILQP Q QQ 55::<8<5 HIsR,B ,C; #C8C8 !C33C8;#EE E(!E EEE8FcD|j|jy)N)rrrs r+__del__z#PreemptionCheckpointHandler.__del__s668--/r-z Runs a training function with error and preemption handling. This function handles the preemption signal from any peer in the cluster by saving the training progress and exiting gracefully. It will also broadcase any program error encountered during the execution of `distributed_train_function` to all workers so that they can raise the same error. The `distributed_train_function` argument should be a distributed train function (i.e., containing a call to `tf.distribute.Strategy.run`). For `tf.distribute.MultiWorkerMirroredStrategy` users, we recommend passing in a single-step `distributed_train_function` to `PreemptionCheckpointHandler.run` so that the checkpoint can be saved in time in case a preemption signal or maintenance notice is sent. Besides the preemption and error handling part, `PreemptionCheckpointHandler.run(distributed_train_function, *args, **kwargs)` has the same effect and output as `distributed_train_function(*args, **kwargs)`. `distributed_train_function` can return either some or no result. The following is a shortened example: ```python @tf.function def distributed_train_step(iterator): # A distributed single-step training function. def step_fn(inputs): # A per-replica single-step training function. x, y = inputs ... return loss per_replica_losses = strategy.run(step_fn, args=(next(iterator),)) return strategy.reduce( tf.distribute.ReduceOp.SUM, per_replica_losses, axis=None) for epoch in range(preemption_handler.total_run_calls // STEPS_PER_EPOCH, EPOCHS_TO_RUN): iterator = iter(multi_worker_dataset) total_loss = 0.0 num_batches = 0 for step in range(preemption_handler.total_run_calls % STEPS_PER_EPOCH, STEPS_PER_EPOCH): total_loss += preemption_handler.run(distributed_train_step) num_batches += 1 train_loss = total_loss / num_batches print('Epoch: %d, train_loss: %f.' %(epoch.numpy(), train_loss)) train_accuracy.reset_states() ``` Args: distributed_train_function: A (single-step) distributed training function. *args: args for `distributed_train_function`. **kwargs: kwargs for `distributed_train_function`. Raises: Program error encountered by any member in the cluster while executing the `distributed_train_function`, or any error from the program error propagation process. Returns: Result of running the `distributed_train_function`. )rnr rZr] _run_for_tpurtr\_run_for_multi_worker_mirroredr8distributed_train_functionargskwargss r+runzPreemptionCheckpointHandler.run%s\  / / < < =T  9 KD KF KK  ,,44,,44# ( 8 88 0T 0 0 $'+/5r-cFtjt||i|S)z?PreemptionCheckpointHandler.run implementation for TPUStrategy.preemption_key)rcheck_preemptionPREEMPTION_KEYrs r+rz(PreemptionCheckpointHandler._run_for_tpus ,,NK %t 6v 66r-c` |jtj}||i|}tj|z }|xjdz c_|j||jz |jz z|_|S#tj $r}|j s~tjd|| tjj|j|j#t$r!}tjd||Yd}~d}~wwxYwd}~wwxYw)z8PreemptionCheckpointHandler.run implementation for MWMS.r$Propagating error to cluster: %r: %s.Ignoring error during error propagation: %r:%sN)&_check_preemption_and_maybe_checkpointrrrr OpErrorrsrur|r report_error_to_cluster error_codemessager) r8rrrrun_begin_timeresult new_run_timerexs r+rz:PreemptionCheckpointHandler._run_for_multi_worker_mirroreds   113yy{n)4:6:fYY[>1l 1!%!9!9 11 1T5F5F=G"Gd M >>    ;QB Q //  3 3ALL!)) L  Q ,,GR P P  Q  sBBB D-#D(8C:9D(: D$DD(D$$D((D-c|jtjjk(r: t j 5t jtdddyy|jtjj$tjj&fvry|j(|i||xj*dz c_d|_y#1swYzxYw#tj$rr}|jjdrLtjdt j|j |i||j#nYd}~yd}~wwxYw)a$Saves a checkpoint if a preemption signal has been made available. This is an alternative API for `PreemptionCheckpointHandler.run` and `PreemptionCheckpointHandler.watch_preemption_scope`. This method works for both `tf.distribute.MultiWorkerMirroredStrategy` and `tf.distribute.TPUStrategy`. However, **for TPUStrategy, this method will add a synchronization point between workers and the coordinator** and thus may have performance implication. If this is a concern, use the combination of `PreemptionCheckpointHandler.watch_preemption_scope` and `PreemptionCheckpointHandler.run` instead. ```python strategy = tf.distribute.TPUStrategy(tpu_cluster_resolver) # initialization omitted with strategy.scope(): # Save in the checkpoint. trained_step = tf.Variable(initial_value=tf.constant(0, dtype=tf.dtypes.int64), name='trained_step', aggregation=tf.VariableAggregation.ONLY_FIRST_REPLICA) checkpoint_manager = tf.train.CheckpointManager(checkpoint, directory, max_to_keep=1) preemption_handler = tf.distribute.experimental.PreemptionCheckpointHandler(cluster_resolver, checkpoint_manager) while trained_step.numpy() < NUM_STEPS: # Train STEPS_IN_FUNCTION steps at once. train_multi_step_function() trained_step.assign_add(STEPS_IN_FUNCTION) preemption_handler.save_checkpoint_if_preempted() ``` Args: *args: args for `tf.train.CheckpointManager.save()` to save checkpoint. **kwargs: kwargs for `tf.train.CheckpointManager.save()` to save. rNCtype.googleapis.com/tensorflow.distributed_runtime.WorkerPreemption/Clearing preemption error to save checkpoint...rr)rnr rZr]r async_scoperrrr AbortedErrorexperimental_payloadsgetrur|async_clear_error_save_checkpointrprtr\rrr)r8rr abort_errors r+save_checkpoint_if_preemptedz8PreemptionCheckpointHandler.save_checkpoint_if_preempteds9F ,,99 :  " - ! 2 2+ - - -&   ,,44,,44#  2d114B6B 1!"d9 - -   , , 0 0 R  ,,H I  # # % $   0 0 --/  s/CCCCCE#1A(EE#c#K|jtjjk(r$ t j 5ddddyy dy#1swYxYw#t j$rp}|jjdrJtjdt j|j|jnYd}~yd}~wwxYw#t j$r}|j s~tjd|| t jj#|j$|j&#t($r!}tjd||Yd}~d}~wwxYwd}~wwxYww)aSyncs error and maybe save checkpoint for usage with TPUStrategy. Note: Usage with `tf.distribute.MultiWorkerMirroredStrategy` does not need this API. Example usage: ```python with preemption_checkpoint_handler.watch_preemption_scope(): while trained_step.numpy() < NUM_STEPS: # distributed_train_function contains a call to strategy.run. loss += preemption_checkpoint_handler.run(distributed_train_function, args=(next(iterator),)) trained_step.assign_add(STEPS_PER_TRAIN_FUNCTION) ``` In this workflow, `PreemptionCheckpointHandler.run` will flag preemption signal received, and `watch_preemption_scope` will handle the preemption signal by saving a checkpoint and then either exit to restart or execute a user-passed `exit_fn` in `tf.distribute.experimental.TerminationConfig`. If no preemption signal is received during execution of ops and function inside the scope, `watch_preemption_scope` ensures the completion of all async op and function execution when exiting and will raises exceptions if async execution results in an error state. Yields: None Nrrrr)rnr rZr]r rr rrrrur|rrrprrsrrrr)r8rrrs r+watch_preemption_scopez2PreemptionCheckpointHandler.watch_preemption_scopesE<  5 D D Q QQ  "      #      , , 0 0 R  ,,H I  # # %    ! --/    ^^ ,,=q! D S OO  5 5allAII N  S LLI2r R R  S s(F A AA F C&F AA C#3A&CF C##F &F 9#F8EF FE;6F;FFF  F ctjj|jjdj dt jd|jtjjk7r%|jj|jtj}t!j"5|j$r|j$|i|n|j&j(|i|dddtj}t jd|j&j*||z |_y#1swYRxYw)z&Saves the checkpoint and exit program.z-PreemptionCheckpointHandler Saving Checkpointrz:PreemptionCheckpointHandler: Starting saving a checkpoint.NzCheckpoint finished at path %s)rrrrnrrrur|r rZr]rassignrr monotonicrpreemption_save_contextrrrsaver_checkpoint_time)r8rr start_timeend_times r+rz,PreemptionCheckpointHandler._save_checkpoint"s::CC ""799DQ LLMN  5 D D Q QQ $$T%9%9:!J  3 3 5=  t&v&+&&++T  . .t~~  # # > > H HI K ,,G I --/ ,,@ A-1$ *  ( ( . . 0# <&  # # * * ,  ) )  , ,tyy{ : ,,K M D--12o llJK  #2  &&**, ../C/> @ ^%9?K*77$$113$$..0,& Pa OO  2 2#$A&:%;1QC@ B LLKQ O P  MMOS -r-c<|jdkxs |jS)zEReturn whether to exit: exit if no grace period or grace period ends.r)rqrrs r+rz)PreemptionCheckpointHandler._time_to_exits *   ! # H(H(HHr-c|jdkDrA|js4d}|j|jz||jzdzz |_yyy)zDSet up at the beginning of a countdown period for long grace period.rN)rqrrrr )r8 buffer_factors r+rz^PreemptionCheckpointHandler._setup_countdown_if_has_grace_period_and_not_already_counting_downs^ Ad&F&Fm  ) )D,>,> > $22 2Q 6 7 ''Gr-c(~~|jy)z?Upload the to-be-preempted worker's id to coordination service.N)r)r8signumframes r+rz/PreemptionCheckpointHandler._sigterm_handler_fns((*r-ctjjt}|tk7r0||_|j j tdtd|j}tjj|dtjd||jdkDrtjjt}|tk7rxtdtd|j}tjj|dtjd||j j ||_yyyy)zWatch out for step-to-save config key and acknowledge. All workers, including the one to be preempted, execute this function to get step-to-save. r1zFPreemptionCheckpointHandler: %s set, preemption awareness acknowledgedrzOPreemptionCheckpointHandler: %s acknowledged, final checkpoint timing received.N)r r rrrrrrrrrur|rqr)r8 step_valueack_keyfinal_step_values r+rz3PreemptionCheckpointHandler._watch_step_to_save_keys8"778NOJ 11",d $$((*#$A&<%=Qt?R?R>STg oo,,Wc: ll ./68   a "??,AA " ; ;'(*>)?qATAT@UV' //  0 0# > ,,,-4 6  ( ( , , .%5$ " < '2r-)NN)!r<r=r>r?r9rwr{rrrrrrrrpropertyrrdo_not_generate_docsrrrrrrcontextmanagerrrrrrrrr@r-r+rbrbPssr#"& QIf`(D 4<1 + BI JD0  dLN$$%N 0Zx7 8C#J8 8t20aPFI. 8+ *6r-rb)9r?rrrQrr.tensorflow.core.distributed_runtime.preemptionrtensorflow.python.checkpointrrfrrtensorflow.python.distributerr-tensorflow.python.distribute.failure_handlingr tensorflow.python.eagerr tensorflow.python.frameworkr r r tensorflow.python.lib.iortensorflow.python.opsrtensorflow.python.platformrrrutensorflow.python.utilr"tensorflow.python.util.deprecationr tensorflow.python.util.tf_exportrtensorflow.tools.docsrrrrrrrrr,objectr0rBrIrLrVr`rbWorkerPreemptionHandlerr@r-r+r2s  RE;>7:O+3..,+,<096.-/-$)-3% 62>bb?bN/* /  -  0 $v @RH_6&_6I_6F6r-