Workers, the Scheduler, and Clients communicate by sending each other Python objects (such as Protocol messages or user data). The communication layer handles appropriate encoding and shipping of those Python objects between the distributed endpoints. The communication layer is able to select between different transport implementations, depending on user choice or (possibly) internal optimizations.
The communication layer lives in the
Communication addresses are canonically represented as URIs, such as
tcp://127.0.0.1:1234. For compatibility with existing code, if the
URI scheme is omitted, a default scheme of
tcp is assumed (so
127.0.0.1:456 is really the same as
The default scheme may change in the future.
The following schemes are currently implemented in the
tcpis the main transport; it uses TCP sockets and allows for IPv4 and IPv6 addresses.
tlsis a secure transport using the well-known TLS protocol over TCP sockets. Using it requires specifying keys and certificates as outlined in TLS/SSL.
inprocis an in-process transport using simple object queues; it eliminates serialization and I/O overhead, providing almost zero-cost communication between endpoints as long as they are situated in the same process.
Some URIs may be valid for listening but not for connecting.
For example, the URI
tcp:// will listen on all IPv4 and IPv6 addresses
and on an arbitrary port, but you cannot connect to that address.
Higher-level APIs in
distributed may accept other address formats for
convenience or compatibility, for example a
(host, port) pair. However,
the abstract communications layer always deals with URIs.
There are a number of top-level functions in
to help deal with addresses:
Split address into its scheme and scheme-dependent location string.
>>> parse_address('tcp://127.0.0.1') ('tcp', '127.0.0.1')
If strict is set to true the address must have a scheme.
>>> unparse_address('tcp', '127.0.0.1') 'tcp://127.0.0.1'
Canonicalize address, adding a default scheme if necessary.
>>> normalize_address('tls://[::1]') 'tls://[::1]' >>> normalize_address('[::1]') 'tcp://[::1]'
Apply scheme-specific address resolution to addr, replacing all symbolic references with concrete location specifiers.
In practice, this can mean hostnames are resolved to IP addresses.
>>> resolve_address('tcp://localhost:8786') 'tcp://127.0.0.1:8786'
Return a hostname / IP address identifying the machine this address is located on.
In contrast to get_address_host_port(), this function should always succeed for well-formed addresses.
>>> get_address_host('tcp://22.214.171.124:80') '126.96.36.199'
The basic unit for dealing with established communications is the
A message-oriented communication object, representing an established communication channel. There should be only one reader and one writer at a time: to manage current communications, even with a single peer, you must create distinct
Messages are arbitrary Python objects. Concrete implementations of this class can implement different serialization mechanisms depending on the underlying transport’s characteristics.
Close the communication immediately and abruptly. Useful in destructors or generators’
Close the communication cleanly. This will attempt to flush outgoing buffers before actually closing the underlying transport.
This method is a coroutine.
Return whether the stream is closed.
Return backend-specific information about the communication, as a dict. Typically, this is information which is initialized when the communication is established and doesn’t vary afterwards.
The local address. For logging and debugging purposes only.
The peer’s address. For logging and debugging purposes only.
Read and return a message (a Python object).
This method is a coroutine.
- deserializers : Optional[Dict[str, Tuple[Callable, Callable, bool]]]
An optional dict appropriate for distributed.protocol.deserialize. See Serialization for more.
Write a message (a Python object).
This method is a coroutine.
- msg :
- on_error : Optional[str]
The behavior when serialization fails. See
distributed.protocol.core.dumpsfor valid values.
You don’t create
Comm objects directly: you either
incoming communications, or
connect to a peer listening for connections:
connect(addr, timeout=None, deserialize=True, connection_args=None)¶
Connect to the given address (a URI such as
tcp://127.0.0.1:1234) and yield a
Commobject. If the connection attempt fails, it is retried until the timeout is expired.
listen(addr, handle_comm, deserialize=True, connection_args=None)¶
Create a listener object with the given parameters. When its
start()method is called, the listener will listen on the given address (a URI such as
tcp://0.0.0.0) and call handle_comm with a
Commobject for each incoming connection.
handle_comm can be a regular function or a coroutine.
Listener objects expose the following interface:
An address this listener can be contacted on. This can be different from listen_address if the latter is some wildcard address such as ‘tcp://0.0.0.0:123’.
The listening address as a URI string.
Start listening for incoming connections.
Stop listening. This does not shutdown already established communications, but prevents accepting new ones.
Extending the Communication Layer¶
Each transport is represented by a URI scheme (such as
backed by a dedicated
Backend implementation, which provides
entry points into all transport-specific routines.
A communication backend, selected by a given URI scheme (e.g. ‘tcp’).
Get a host name (normally an IP address) identifying the host the address is located on. loc is a scheme-less address.
Get the (host, port) tuple of the scheme-less address loc. This should only be implemented by IP-based transports.
Get a connector object usable for connecting to addresses.
get_listener(loc, handle_comm, deserialize, **connection_args)¶
Get a listener object for the scheme-less address loc.
Get the local listening address suitable for reaching loc.
Resolve the address into a canonical form. loc is a scheme-less address.
Simple implementations may return loc unchanged.