U /e@srddlmZddlZddlmZmZddlmZddlm Z e e Z GdddeZ dd Zd d Zd d ZdS)) annotationsN)coerce_to_addressconnect)SchedulerPlugin)dumps_functionc@s"eZdZdZdddZddZdS) EventStreamz Maintain a copy of worker eventsNcCsd|_g|_|r||dS)Nr)namebufferZ add_plugin)self schedulerr G/tmp/pip-unpacked-wheel-g426oqom/distributed/diagnostics/eventstream.py__init__szEventStream.__init__cOs0|dkr,||d<|dks |dkr,|j|dS)N processingkeyZmemoryZerred)r append)r rstartfinishargskwargsr r r transitionszEventStream.transition)N)__name__ __module__ __qualname____doc__rrr r r r r s rcCsg|j|_}|S)N)r )r esr r r r swap_buffersrcCs|j|jddS)N)r)Z remove_pluginr)r rr r r teardown!srcsBt|}t|IdH}|dtttt|ttdIdH|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)rrwriterrrr)addressr"Zcommr r r eventstream%s r%) __future__rloggingZdistributed.corerrZdistributed.diagnostics.pluginrZdistributed.workerr getLoggerrloggerrrrr%r r r r s