o i e;@sLdZddlZddlZddlmZddlmZddlmZddl m Z ddl m Z ddl mZdd l mZdd lmZdd lmZdd lmZdd lmZddlmZdaGdddejZd'ddZGdddeZddZ Gdddej!Z"   d(ddZ#ddZ$Gdd d ej%Z&Gd!d"d"eZ'Gd#d$d$eZ(Gd%d&d&eZ)dS))zCOperations for handling session logging and shutdown notifications.N) text_format) config_pb2) event_pb2)session)dtypes)errors)ops) array_ops) tf_logging)tpu_ops)session_run_hook) training_utilc@seZdZdZddZdS)CoordinatorResetErrorz/Raised when the monitored session should reset.cCstj|ddddS)Nz.Resetting session loop due to worker shutdown.)r AbortedError__init__selfrT/var/www/myenv/lib/python3.10/site-packages/tensorflow/python/tpu/session_support.pyr(szCoordinatorResetError.__init__N)__name__ __module__ __qualname____doc__rrrrrr%s rcCs"tj|j|j|r |dS|jdS)N)targetconfiggraph) session_libSessionsess_str_configr)rrrrr_clone_session-sr c@sXeZdZdZddZeddZddZdd Zdd d Z ddZ ddZ dddZ d S)WorkerHeartbeatManagerz:Manages the status/heartbeat monitor for a set of workers.cCs||_||_||_||_dS)aConstruct a new WorkerHeartbeatManager. (Prefer using `WorkerHeartbeatManager.from_devices` when possible.) Args: session: `tf.compat.v1.Session`, session to use for heartbeat operations. devices: `list[string]` Set of devices to connect to. heartbeat_ops: `list[tf.Operation]` Heartbeat operations. request_placeholder: `tf.Placeholder[String]` Placeholder used to specify the WorkerHeartbeatRequest protocol buffer. N)_session_devices_ops_request_placeholder)rrdevices heartbeat_opsrequest_placeholderrrrr7s  zWorkerHeartbeatManager.__init__c Cs|stdtd|tjdtjd}g}|D]}t|| t |Wdn1s3wYqt ||||S)z4Construct a heartbeat manager for the given devices.z3Trying to create heartbeat manager with no devices?z!Creating heartbeat manager for %sworker_heartbeat_request)namedtypeN) loggingerrorinfor placeholderrstringrdeviceappendr worker_heartbeatr!)rr&r(r'r1rrr from_devicesHs   z#WorkerHeartbeatManager.from_devicescCs t|jSN)lenr#rrrr num_workersZs z"WorkerHeartbeatManager.num_workerscCs0tdt||j|j|j|idS)zConfigure heartbeat manager for all devices. Args: message: `event_pb2.WorkerHeartbeatRequest` Returns: `None` z Configuring worker heartbeat: %sN) r,r.rMessageToStringr"runr$r%SerializeToString)rmessagerrr configure]s   z WorkerHeartbeatManager.configureN`cCsX|durt}tj|d}|jj|j|j|i|d}dd|D}t d||S)z6Ping all workers, returning the parsed status results.N) timeout_in_ms) feed_dictoptionscSsg|]}tj|qSr)rWorkerHeartbeatResponse FromString).0res_pbrrr ss z/WorkerHeartbeatManager.ping..zPing results: %s) rWorkerHeartbeatRequestr RunOptionsr"r9r$r%r:r,debug)rrequestr>r@resultsparsed_resultsrrrpingis   zWorkerHeartbeatManager.pingcCsh|}g}t||j|jD]\}}}|jtjkr |||fq|s%dSt|\}}t|j |||j S)zFPing all workers, returning manager containing lame workers (or None).N) rLzipr#r$ health_statusrOKr2r!r"r%)r ping_results lame_workers ping_responser1op bad_devicesbad_opsrrrrQzs   z#WorkerHeartbeatManager.lame_workerscCsdd|jS)NzHeartbeatManager(%s),)joinr#rrrr__repr__zWorkerHeartbeatManager.__repr__rcCs^td|tjtj|dtjtj|dd}||d|d}td|t |dS) z3Shutdown all workers after `shutdown_timeout_secs`.zShutting down %s. timeout_ms exit_code)watchdog_config shutdown_moder]$@z)Waiting %.2f seconds for worker shutdown.N) r,r.rrFWatchdogConfigSHUTDOWN_AFTER_TIMEOUTRequestedExitCoder<timesleep)rwait_time_in_msr]req sleep_secrrrshutdowns      zWorkerHeartbeatManager.shutdown)Nr=)r=r) rrrrr staticmethodr4r7r<rLrQrXrjrrrrr!4s  r!cCsB|}g}|D]}|j}d|vrd|vr||ddq|S)z7Return a list of devices for each worker in the system.z:TPU:0 coordinatorTPUCPU) list_devicesr*r2replace)rr&devices_that_support_heartbeatsr1r*rrrall_worker_devicessrrc@sReZdZdZ   dddZddd Zd d Zd d ZddZddZ ddZ dS)WatchdogManageraConfigures worker watchdog timer and handles periodic pings. Usage: # Ping workers every minute, shutting down workers if they haven't received # a ping after 1 hour. watchdog_manager = WatchdogManager( ping_interval=60, shutdown_timeout=3600 ) # Use as a context manager, resetting watchdog on context exit: with watchdog_manager: session.run(...) # Or setup globally; watchdog will remain active until program exit. watchdog_manager.configure_and_run() N< cCsPtj|||_||_d|_|j|_|j|_d|_ ||_ d|_ d|_ d|_ dS)aInitialize a watchdog manager. Args: session: Session connected to worker devices. A cloned session and graph will be created for managing worker pings. devices: Set of devices to monitor. If none, all workers will be monitored. ping_interval: Time, in seconds, between watchdog pings. shutdown_timeout: Time, in seconds, before watchdog timeout. TFN) threadingThreadr ping_intervalshutdown_timeoutdaemonrr_target_runningr#_graphr"_worker_manager)rrr&rxryrrrrs  zWatchdogManager.__init__FcCst|_tj|j|j|jd|_|jdurt |j|_|j t |j|j|_ Wdn1s5wY|rBd}tj}n|jd}tj}|j tjtj|d|ddS)z,Reset the graph, session and worker manager.)rrrNrarZ)r^r_)rGraphr}rrr{rr"r#rr as_defaultr!r4r~rNOT_CONFIGUREDryWAIT_FOR_COORDINATORr<rFrb)rstoppingr[r_rrr_reset_managers0      zWatchdogManager._reset_managercCs,td|j|j|d|_|dS)NzKEnabling watchdog timer with %d second timeout and %d second ping interval.T)r,r.ryrxrr|startrrrrconfigure_and_runs z!WatchdogManager.configure_and_runcCs(td|jddd|_|dS)NzStopping worker watchdog.T)rF)r,r.rr|rWrrrrstops   zWatchdogManager.stopcC |dSr5)rrrrr __enter__ zWatchdogManager.__enter__cCrr5)r)rexc_typeexc_valexc_tbrrr__exit__ rzWatchdogManager.__exit__c Csn|jr5z|jjddt|jWntjy/}zt d|| WYd}~nd}~ww|jsdSdS)N)rIz(Caught error while sending heartbeat: %s) r|r~rLrerfrxrOpErrorr,rHr)rerrrr9 s zWatchdogManager.run)Nrtru)F) rrrrrrrrrrr9rrrrrss   rsrtcCs4tdurt|d|}t||||atdSdS)zEStart global worker watchdog to shutdown workers on coordinator exit.Nr`) _WATCHDOGminrsr)rr&rxryrrrstart_worker_watchdogs rcCstdur tdadSdS)zStop global worker watchdog.N)rrrrrrstop_worker_watchdog)src@s2eZdZdZd ddZddZddZd d ZdS) GracefulShutdownHooka%Session hook that watches for shutdown events. If a shutdown is indicated, `saver.save(checkpoint_prefix)` is executed, and a SystemShutdown exception is raised to terminate the main session. If `saver` is None the `SAVERS` collection will be read to find a saver. `on_shutdown_hooks` is an optional list of functions that should be called after checkpointing. The function is called with (`run_context`, `all_workers`, `lame_workers`). If `heartbeat_group` is not specified, it will default to all CPU workers in the system. NcCs:||_||_|r |ng|_t|_d|_d|_d|_dS)NF) _saver_checkpoint_prefix_on_shutdown_hooksrrr}_workersr"_heartbeat_supported)rcheckpoint_prefixsaveron_shutdown_hooksrrrr@s  zGracefulShutdownHook.__init__c Cstdur|durtd|j`tdt||j|_ t |j t |j |_ |j dk|_|jrXz |j tjtjdWntjyWtdd|_YnwtdWddSWddSWddS1sxwYdS)NzuSaver defined but no global step. Run `get_or_create_global_step()` in your model definition to allow checkpointing.z"Installing graceful shutdown hook.r)r_zJTPU device does not support heartbeats. Failure handling will be disabled.FzANo workers support heartbeats. Failure handling will be disabled.)r get_global_stepr ValueErrorr}rr,r.r r"r!r4rrrr7rr<rrFrrInvalidArgumentErrorwarn)rtraining_sessioncoordrrrafter_create_sessionKs@    "z)GracefulShutdownHook.after_create_sessioncCsR|jr|jSttjj}|sdSt|ts|St|dkr%t ddS|dS)NzMultiple savers in the SAVERS collection. On-demand checkpointing will be disabled. Pass an explicit `saver` to the constructor to override this behavior.r) rrget_collection GraphKeysSAVERS isinstancelistr6r,r-)rsaversrrrrgs  zGracefulShutdownHook.savercCs~|jsdS|j}|r@td||r.td|j|j|j|jt ddntd|j D] }|||j|q6dSdS)Nz$ShutdownHook: lame workers found: %sz%ShutdownHook: saving checkpoint to %sT) global_step write_statezShutdownHook: no Saver defined.) rrrQr,r.rrsaverr rr)r run_context run_valuesrQfnrrr after_run{s*    zGracefulShutdownHook.after_run)NN)rrrrrrrrrrrrr1s    rc@ eZdZdZddZddZdS)ResetComputationzHook to reset a TPUEstimator computation loop. This hook shuts down all workers and resets the monitored session loop by throwing a CoordinatorResetError. cCdSr5rrrrrrzResetComputation.__init__cCs ~~|jddtdt)N*r\zResetting coordinator.)rjr,r.rrr all_workersrQrrr__call__s  zResetComputation.__call__Nrrrrrrrrrrr rc@r)ShutdownLameWorkersz~Shutdown lamed workers. Processing will continue normally (typically by waiting for the down workers to be restarted). cCrr5rrrrrrrzShutdownLameWorkers.__init__cCs|jdddSNrr\rjrrrrrrYzShutdownLameWorkers.__call__Nrrrrrrrrc@r)ShutdownAllWorkersz|Shutdown all workers. Processing will continue normally (typically by waiting for the down workers to be restarted). cCrr5rrrrrrrzShutdownAllWorkers.__init__cCs|jdddSrrrrrrrrYzShutdownAllWorkers.__call__Nrrrrrrrrr5)Nrtr)*rrvregoogle.protobufrtensorflow.core.protobufrtensorflow.core.utilrtensorflow.python.clientrrtensorflow.python.frameworkrrrtensorflow.python.opsr tensorflow.python.platformr r,tensorflow.python.tpu.opsr tensorflow.python.trainingr r rrrr objectr!rrrwrsrrSessionRunHookrrrrrrrrs<             kn d