B °F.\Šã@sŒddlmZmZmZddlZddlmZddlmZddl m Z m Z ddl m Z e e¡ZGd d „d eƒZd d „Zd d„Zejdd„ƒZdS)é)Úprint_functionÚdivisionÚabsolute_importN)Úgené)ÚSchedulerPluginé)ÚconnectÚcoerce_to_address)Údumps_functionc@s"eZdZdZddd„Zdd„ZdS)Ú EventStreamz" Maintain a copy of worker events NcCsg|_|r| |¡dS)N)ÚbufferZ add_plugin)ÚselfÚ scheduler©rúBlib/python3.7/site-packages/distributed/diagnostics/eventstream.pyÚ__init__szEventStream.__init__cOs0|dkr,||d<|dks |dkr,|j |¡dS)NZ processingÚkeyZmemoryZerred)r Úappend)rrÚstartZfinishÚargsÚkwargsrrrÚ transitionszEventStream.transition)N)Ú__name__Ú __module__Ú __qualname__Ú__doc__rrrrrrr s r cCsg|j|_}|S)N)r )rÚesr rrrÚ swap_buffersrcCs| |¡dS)N)Z remove_plugin)rrrrrÚteardown$srccsDt|ƒ}t|ƒV}| dttƒttƒ|ttƒdœ¡Vt |¡‚dS)aƒ Open a TCP connection to scheduler, receive batched task messages The messages coming back are lists of dicts. Each dict is of the following form:: {'key': 'mykey', 'worker': 'host:port', 'status': status, 'compute_start': time(), 'compute_stop': time(), 'transfer_start': time(), 'transfer_stop': time(), 'disk_load_start': time(), 'disk_load_stop': time(), 'other': 'junk'} Where ``status`` is either 'OK', or 'error' Parameters ---------- address: address of scheduler interval: time between batches, in seconds Examples -------- >>> stream = yield eventstream('127.0.0.1:8786', 0.100) # doctest: +SKIP >>> print(yield read(stream)) # doctest: +SKIP [{'key': 'x', 'status': 'OK', 'worker': '192.168.0.1:54684', ...}, {'key': 'y', 'status': 'error', 'worker': '192.168.0.1:54684', ...}] Zfeed)ÚopZsetupZfunctionÚintervalrN) r r Úwriter r rrrZReturn)Zaddressr!ZcommrrrÚ eventstream(s r#)Z __future__rrrZloggingZtornadorZpluginrZcorer r Zworkerr Z getLoggerrZloggerr rrÚ coroutiner#rrrrÚs