0FieddlmZddlZddlmZmZddlmZddlm Z ej e Z GddeZ dZd Zd ZdS) ) annotationsN)coerce_to_addressconnect)SchedulerPlugin)dumps_functionc eZdZdZddZdZdS) EventStreamz Maintain a copy of worker eventsNcTd|_g|_|r||dSdS)Nr )namebuffer add_plugin)self schedulers Clib/python3.11/site-packages/distributed/diagnostics/eventstream.py__init__zEventStream.__init__s<!   '   & & & & & ' 'cp|dkr-||d<|dks|dkr|j|dSdSdS)N processingkeymemoryerred)r append)rrstartfinishargskwargss r transitionzEventStream.transitionsU L F5M!!Vw%6%6 ""6***** ! %6%6rN)__name__ __module__ __qualname____doc__rrrrr r s=**'''' +++++rr c$g|jc|_}|Sr)r )resr s r swap_bufferr&sBIBIv Mrc<||jdS)N)r ) remove_pluginr )rr%s rteardownr)!s! )))))rc Kt|}t|d{V}|dtttt |tt dd{V|S)aOpen 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 = await eventstream('127.0.0.1:8786', 0.100) # doctest: +SKIP >>> print(await read(stream)) # doctest: +SKIP [{'key': 'x', 'status': 'OK', 'worker': '192.168.0.1:54684', ...}, {'key': 'y', 'status': 'error', 'worker': '192.168.0.1:54684', ...}] Nfeed)opsetupfunctionintervalr))rrwriterr r&r))addressr/comms r eventstreamr3%s4 ((G!! ! ! ! ! ! !D **#K00&{33 &x00      Kr) __future__rloggingdistributed.corerrdistributed.diagnostics.pluginrdistributed.workerr getLoggerrloggerr r&r)r3r#rrr;s""""""77777777::::::------  8 $ $ + + + + +/ + + +  ***%%%%%r