U /e% @s`ddlmZddlmZmZddlmZddlmZddl m Z ddl m Z GdddeZ d S) ) annotations) AwaitableCallable) parse_bytes) ShardsBuffer)ResourceLimiter) log_errorscsNeZdZdZedZedZddddd fd d Zd d ddddZZ S)CommShardsBufferaAccept, buffer, and send many small messages to many workers This takes in lots of small messages destined for remote workers, buffers those messages in memory, and then sends out batches of them when possible to different workers. This tries to send larger messages when possible, while also respecting a memory bound **State** - shards: dict[str, list[bytes]] This is our in-memory buffer of data waiting to be sent to other workers. - sizes: dict[str, int] The size of each list of shards. We find the largest and send data from that buffer State ----- memory_limit: str A maximum amount of memory to use across the process, like "1 GiB" This includes both data in shards and also in network communications max_connections: int The maximum number of connections to have out at once max_message_size: str The maximum size of a single message that we want to send Parameters ---------- send: callable How to send a list of shards to a worker Expects an address of the target worker (string) and a payload of shards (list of bytes) to send to that worker z2 MiBz100 MiBN z-Callable[[str, list[bytes]], Awaitable[None]]zResourceLimiter | Noneint)sendmemory_limiterconcurrency_limitcstj||tjd||_dS)N)r rmax_message_size)super__init__r rr )selfr r r __class__>/tmp/pip-unpacked-wheel-g426oqom/distributed/shuffle/_comms.pyr3s zCommShardsBuffer.__init__strz list[bytes]None)addressshardsreturnc s>t.|d|||IdHW5QRXW5QRXdS)z,Send one message off to a neighboring workerr N)rtimer )rrrrrr_process@s zCommShardsBuffer._process)Nr ) __name__ __module__ __qualname____doc__rrZ memory_limitrr __classcell__rrrrr s# r N) __future__rtypingrrZ dask.utilsrZdistributed.shuffle._diskrZdistributed.shuffle._limiterrZdistributed.utilsrr rrrrs