o vi+@sbddlmZddlZddlZddlZddlmZddlmZm Z m Z ddl m Z GdddZ dS)) annotationsN)Iterator)AnyOptionalUnion)logc@seZdZdZ    d)d*d dZd+ddZd+ddZejd,d-ddZ d.ddZ d/d0dd Z d+d!d"Z d1d#d$Z d+d%d&Zd+d'd(ZdS)2OperationWatchdoga8A watchdog that monitors operations for hangs and collects performance statistics. This class provides a mechanism to detect when operations take longer than expected, which can help identify potential deadlocks or performance issues. It also collects statistics about operations to help identify bottlenecks. Attributes: warning_threshold: Time in seconds before warning about potential hangs. check_interval: Time in seconds between checks for hung operations. verbose_interval: Time in seconds between verbose logging. Xwarning_thresholdintcheck_intervalverbose_intervalnamestrreturnNonecCsJ||_||_||_||_i|_i|_t|_t |_ d|_ | dS)aaInitialize the watchdog. Args: warning_threshold: Time in seconds before warning about potential hangs. Defaults to 600 (10 minutes). check_interval: Time in seconds between checks. Defaults to 30. verbose_interval: Time in seconds between verbose logging. Defaults to -1 (disabled). N) _warning_threshold_check_interval_verbose_interval_name_ops_stats threadingLock_lockEvent _stop_event_threadstart)selfr rrrr"S/data/cameron/vidgen/cosmos-policy/cosmos_policy/_src/predict2/datasets/watchdog.py__init__(s   zOperationWatchdog.__init__cCs`|jdus |js.|jtj|jd|jdd|_|jt d|jddSdS)zsStart the watchdog monitoring thread. If the thread is already running, this method does nothing. NTZ_monitor_thread)targetdaemonr[z$] Watchdog monitoring thread started) ris_aliverclearrThread _monitor_looprr rdebugr!r"r"r#r Ds   zOperationWatchdog.startcCsr|j|jr5|jr7|jjdd|jr'tjd|jddddStjd|jddddSdSdS) zzStop the watchdog monitoring thread. This method is typically called when shutting down the application. g?)timeoutr'z2] Watchdog thread did not terminate within timeoutF rank0_onlyz$] Watchdog monitoring thread stoppedN) rsetrr(joinrwarningrr,r-r"r"r#stopOs  zOperationWatchdog.stopoperation_name descriptionverbose_first_nIterator[None]c cs8|dttd}t}|j||p|||dd|j|<Wdn1s-wYzdVWt|}||jkrXtjd|jd|d|d |d d dd |jB||jvre|j|=||jvrtd dddd|j|<|j|}|dd7<|d|7<t |d||d<||d<Wdn1swY|d kr|d|kr|d|d}tj d|jd|d|dd|d|d d|d d|dd d|dd d dd dSdSdSt|}||jkr tjd|jd|d|d |d d dd |jD||jvr|j|=||jvr(d dddd|j|<|j|}|dd7<|d|7<t |d||d<||d<Wdn 1sUwY|d kr|d|kr|d|d}tj d|jd|d|dd|d|d d|d d|dd d|dd d dd www)a}Context manager for monitoring an operation. This is the primary interface for using the watchdog. It automatically tracks the start and end time of the operation and updates statistics. Args: operation_name: Name/type of the operation to monitor. description: Optional description providing more context. verbose_first_n: If positive, print verbose logs for first N operations for operation_name. Yields: None Example: with watchdog.watch("data_fetch", "Fetching user data"): data = fetch_user_data(user_id) _iF)rdescr updatewarnedNr'z] Operation id: z , name: 'z' took .2fsr/rg)count total_timemax_time last_timer@rArBrCz ] name: 'z ', count z / z took zs. avg zs, max zs, last ) r timerrrrr3rrmaxinfo) r!r6r7r8op_id start_timedurationstatsavg_timer"r"r#watch\sz        J       JzOperationWatchdog.watchcCsdt}|j!|jD]\}}|d|kr||d<d|d<q WddS1s+wYdS)zSend a heartbeat for all operations of a given type. Use this inside long operations to prevent false warnings. Args: operation_name: The operation type to update. rr<Fr=N)rErritems)r!r6 current_timeZ_op_idopr"r"r# heartbeats "zOperationWatchdog.heartbeatN Optional[str]dict[str, Any]cCs|j^|r4||jvr+|j|}|ddkr"|d|d|d<|WdSiWdSi}|jD]\}}|||<|ddkrW|d|d||d<q;|WdS1sdwYdS)aGet statistics for operations. Args: operation_name: Get stats for specific operation, or None for all. Returns: Dictionary of operation statistics. For each operation, includes: - count: Number of completed operations - total_time: Total time spent in this operation type - max_time: Maximum time spent in a single operation - last_time: Time spent in the most recent operation - avg_time: Average time per operation (if count > 0) r@rrArLN)rrcopyrN)r!r6rKresultrr"r"r# get_statss$     $zOperationWatchdog.get_statscCs|}|stjd|jddddStjd|jddd|D]6\}}|ddkrZ|d |d}tjd|jd |d |dd |d d|dd d|dd d ddq$dS)zwPrint statistics for all operations. This is a convenience method that logs statistics at INFO level. r'z#] No operation statistics availableFr/Nz] Operation Statistics:r@rrAz] z: count=z, avg=r>zs, max=rBzs, last=rCr?)rVrrGrrN)r!rKrr?avgr"r"r# print_statss( zOperationWatchdog.print_statscCsx|j/t}i}|jD]\}}|d|d||d||dd||<q|WdS1s5wYdS)zList all currently active operations. Returns: Dictionary mapping operation IDs to information about active operations. rr;r r<)rr7 running_timeZtime_since_updateN)rrErrN)r!rOrUrHrPr"r"r#list_active_operationss   $z(OperationWatchdog.list_active_operationscCs6|j|jWddS1swYdS)zReset all operation statistics. This clears all accumulated statistics but does not affect active operations. N)rrr)r-r"r"r# reset_statss "zOperationWatchdog.reset_statscCsd}|jst}|jKt|jD];\}}||d}||jkrQ|dsQ||d}tj d|j d|dd|d d |d d |d d ddd|d<qWdn1s\wY|j dkrs|||j krs| |}|j |j|jrdSdS)zuMonitor registered operations for hangs. This is an internal method that runs in a separate thread. rr<r=r r'z] POTENTIAL HANG: 'rz' (r;z) has been running for z.1fzs total, with zs since last updateFr/TN)ris_setrErlistrrNrrr3rrrXwaitr)r!Zlast_verbose_timenowr:rPelapsedrAr"r"r#r+s2   zOperationWatchdog._monitor_loop)r r r r) r r rr rr rrrr)rr)r5r )r6rr7rr8r rr9)r6rrr)N)r6rRrrS)rrS)__name__ __module__ __qualname____doc__r$r r4 contextlibcontextmanagerrMrQrVrXrZr[r+r"r"r"r#rs"    B    r) __future__rrerrEcollections.abcrtypingrrr#cosmos_policy._src.imaginaire.utilsrrr"r"r"r#s