BVhdZddlZddlmZddlmZddlmZddlmZddlm Z ddl m Z dd l m Z dd l m Z dd l mZdd lmZe j"d dZe j"ddZdZedgGddZy)z?Provides a utility class for preemption detection and recovery.N)logging)detect_platform)PlatformDevice)context) monitoring) AbortedError)CancelledError) InternalError)UnavailableError) tf_exportzD/tensorflow/api/distribution_strategy/preemption_watcher_initializedz'Counter for usages of PreemptionWatcherz@/tensorflow/api/distribution_strategy/preemption_watcher_handledzICounter for number of preempions catched and handled by PreemptionWatcher TF_DEFAULT_PREEMPTION_NOTICE_KEYz)distribute.experimental.PreemptionWatcher)v1c2eZdZdZdZedZdZdZy)PreemptionWatcheraWatch preemption signal and store it. Notice: Currently only support Borg TPU environment with TPUClusterResolver. This class provides a way to monitor the preemption signal during training on TPU. It will start a background thread to watch the training process, trying to fetch preemption message from the coordination service. When preemption happens, the preempted worker will write the preemption message to the coordination service. Thus getting a non-empty preemption message means there is a preemption happened. User can use the preemption message as a reliable preemption indicator, and then set the coordinator to reconnect to the TPU worker instead of a fully restart triggered by Borg. For example, a training process with preemption recovery will be like: ```python keep_running = True preemption_watcher = None while keep_running: try: # Initialize TPU cluster and stratygy. resolver = tf.distribute.cluster_resolver.TPUClusterResolver() tf.config.experimental_connect_to_cluster(resolver) tf.tpu.experimental.initialize_tpu_system(resolver) strategy = tf.distribute.TPUStrategy(resolver) # PreemptionWatcher must be created after connected to cluster. preemption_watcher = tf.distribute.experimental.PreemptionWatcher() train_model(strategy) keep_running = False except Exception as e: if preemption_watcher and preemption_watcher.preemption_message: preemption_watcher.block_until_worker_exit() keep_running = True else: raise e ``` Attributes: preemption_message: A variable to store the preemption message fetched from the coordination service. If it is not None, then there is a preemption happened. platform: A PlatformDevice to indicate the current job's platform. Refer to failure_handling_util.py for the definition of enum class PlatformDevice. cPd|_t|_|jtjk7r!t j d|jytjjdtj|jdjy)Nz3Preemption watcher does not support environment: %sT)targetdaemon)_preemption_messager _platformr INTERNAL_TPUrwarning*_preemption_watcher_initialization_counterget_cell increase_by threadingThread_watch_preemption_keystartselfs p/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/distribute/failure_handling/preemption_watcher.py__init__zPreemptionWatcher.__init__]ss#D$&DN ~~444 oo ?199;GGJd88FLLNc|jS)zReturns the preemption message.)rr s r"preemption_messagez$PreemptionWatcher.preemption_messageis  # ##r$ctjdtjjt}t j jdtjd||_y)NzWatching preemption signal.rzPreemption signal received.) rinforget_config_key_value_PREEMPTION_KEY_preemption_handling_counterrrr)r!messages r"rz'PreemptionWatcher._watch_preemption_keynsR LL./oo44_EG ))+77: LL./&Dr$cP|jtjk7ry tjj dy#t $r.}d|j vrtjdYd}~yd}~wtttf$rtjdYywxYw)aBlock coordinator until workers exit. In some rare cases, another error could be raised during the preemption grace period. This will cause the coordinator to reconnect to the same TPU workers, which will be killed later. It prevents the coordinator to reconnect to new TPU workers, and falls back to a hard restart. To avoid this situation, this method will block the coordinator to reconnect until workers exit. This method will be a no-op for non-TPU platform. NBLOCK_TILL_EXITz$Coordination service is not enabled.zWorkers exited.) rrrrr)r r,rr(rr r )r!es r"block_until_worker_exitz)PreemptionWatcher.block_until_worker_exitus ~~444 & oo,,->? & /qyy @  ll$%% .*: ;& ll$%&s#A B% $A66,B%$B%N) __name__ __module__ __qualname____doc__r#propertyr&rr0r$r"rr,s,-^ O $ $'&r$r)r4rabslrCtensorflow.python.distribute.failure_handling.failure_handling_utilrrtensorflow.python.eagerrr"tensorflow.python.framework.errorsrr r r tensorflow.python.util.tf_exportr Counterrr+r*rr6r$r"r=sF_^+.;=<?6.@Z-?-?J-.* 2z11FO  5 62>\&\&?\&r$