Protocol and Tool Support

Asynchronous connections

Use Protocol from an asynchronous application. Enter its async context before calling tools. See Quickstart for a complete connection example and Concepts for subprocess ownership and cancellation.

The transport guide explains command prefixes, local Python and prepared streams such as nc, including when to use from_command or from_subprocess.

class rmote.protocol.Protocol(reader, writer, *, decode_threshold=32768)

Bases: BaseProtocol

async __aenter__()

Return type:

Self

async __aexit__(exc_type, exc_val, exc_tb)

Return type:

None

__call__(tool, /, *args, **kwargs)

Overloads:

• self, tool (Callable[P, AsyncIterator[R]]), args (P.args), kwargs (P.kwargs) → AsyncGenerator[R, None]

• self, tool (Callable[P, Coroutine[Any, Any, R]]), args (P.args), kwargs (P.kwargs) → Coroutine[Any, Any, R]

• self, tool (Callable[P, R | Coroutine[Any, Any, R]]), args (P.args), kwargs (P.kwargs) → Coroutine[Any, Any, R]

Call a tool method on the peer.

An ordinary method gives a coroutine, so the caller awaits it. A method written as an async generator gives an async generator instead, so the caller reads it with async for and the items arrive as the remote side produces them. A bare break does not close that generator at once, so use contextlib.aclosing when the close has to be immediate.

A close stops the remote method at its next permission check, which happens when the method produces its next item. A method that waits for an item it never receives keeps running, so give such a method a separate operation that releases its resource.

get_id()

Return type:

getint

get_log_id()

Return type:

getint

handle_stream_credit(returned, packet_id)

Take the permission that a streaming consumer gave back.

A frame with STREAM and REQUEST always belongs to a stream this side produces, and one with STREAM and RESPONSE to a stream this side consumes. The direction tells the two registries apart, so the same number on both sides is never confused.

None means the consumer left, so the sender stops.

Return type:

None

handle_stream_item(flags, packet_id, payload, size)

Hand one item of a streaming response to its consumer.

This stays synchronous, because it runs in the receive loop. The budget that the consumer gives the sender already bounds the backlog, so the limits here only catch a peer that ignores the budget.

Return type:

None

async prepare_call(tool)

Make sure the tool is on the peer. Return the method id to call.

Raises:

ValueError – tool is neither a Tool method nor an importable module function.

Return type:

Text

async send_stream(items, packet_id)

Send the items of a streaming method, several of them in one packet.

A separate task writes what gathered, so the items that the producer makes while a packet is on the wire travel in the next one. Each item needs permission from the consumer before it reaches that task, so a consumer that reads slowly slows the producer and one that leaves stops it. The permission is awaited before the write lock is taken, so a waiting stream never holds the transport.

Raises:

ValueError – One item is above MAX_STREAM_ITEM. The consumer gets the failure, because an item that large could never find room.

Return type:

None

async stream(tool, /, *args, **kwargs)

Call a streaming tool method and yield its items as they arrive.

The method must be an async generator on the remote side. The items travel as responses, so the peer pushes without a request per item, and the items it has ready travel together in one packet. A large packet is fragmented, so a long stream does not hold the channel and ordinary calls keep their progress.

Closing the iterator tells the sender to stop. The task that produces the items is cancelled, so a generator that waits for data it never receives is interrupted as well, and the cleanup of the method runs in either case. Use contextlib.aclosing for a deterministic close on an early break.

Synchronous code inside the method cannot be interrupted: the cancellation arrives at the next await. A method that holds a resource between calls still needs its own operation to release it, exactly as an ordinary method does.

The items that the sender has ready travel in one packet, so a write on one side and a read on the other serve a burst instead of serving every item. One connection therefore carries many streams at the same price: with 64-byte items the cost is near 2 µs per item from one stream to thirty-two. An item that is alone travels at once, so a producer that waits between items keeps the latency of each one.

Equal streams do not finish together: the last of thirty-two finishes about 40 percent of the run after the first, because a stream that is served sends a whole batch. Where every stream must advance evenly, give each one its own connection.

Parameters:

• tool (Callable[[ParamSpec(P)], AsyncIterator[TypeVar(R)]]) – Method of a Tool class, defined as an async generator.

• *args (ParamSpecArgs) – Positional arguments of the method.

• **kwargs (ParamSpecKwargs) – Keyword arguments of the method.

Yields:

Items in the order the remote side produced them.

Raises:

• ConnectionError – The connection dropped while the stream was open.

• RuntimeError – The peer exceeded the stream budget or item limit.

• Exception – Whatever the remote method raised.

uncompressed(tool, /, *args, **kwargs)

Overloads:

• self, tool (Callable[P, AsyncIterator[R]]), args (P.args), kwargs (P.kwargs) → AsyncGenerator[R, None]

• self, tool (Callable[P, Coroutine[Any, Any, R]]), args (P.args), kwargs (P.kwargs) → Coroutine[Any, Any, R]

• self, tool (Callable[P, R | Coroutine[Any, Any, R]]), args (P.args), kwargs (P.kwargs) → Coroutine[Any, Any, R]

Call without compressing the request, response or stream items.

All arguments belong to the tool, including a keyword named compressed. Concurrent calls keep their own compression policy. Stream consumers must close the returned generator just as for an ordinary stream call.

async wait_closed()

Return type:

None

Tool definitions and commands

Subclass Tool in a readable Python module. Import async_process from rmote.process and use await async_process(...) inside async tool methods to run commands without blocking other calls; set capture_output=True when the return value needs output. Built-in command tools use async methods, so direct local calls require await. Calls through Protocol and the synchronous Connection keep their normal calling conventions. The synchronous process helper remains available. See Subprocess helpers for subprocess helpers and Writing Tools for executable examples.

class rmote.protocol.Tool

Bases: object

rmote.protocol.inline(method)

Run a synchronous Tool method directly on the event loop, without a worker.

A synchronous method runs in a worker thread, because a call that waits would stop the loop and with it every other call of the connection. The hand-off to that thread costs about 30 us, which is more than a method that only reads a field, formats a value or returns a constant.

Mark such a method with this decorator, above or below staticmethod. Use it only when the method cannot wait: no file, no socket, no lock held by code that waits, and no long computation. The decorator does not enforce this requirement or change direct local calls.

>>> class Tiny(Tool):
...     @staticmethod
...     @inline
...     def answer() -> int:
...         return 42

Return type:

TypeVar(Method)

Streaming calls

A tool method written as an async generator streams its result. The items travel as responses with the STREAM flag set, and the ordinary response that follows them ends the stream. The remote side pushes the items as it produces them and does not wait for a request per item. The items that it has ready travel together in one packet, which the BATCH flag marks.

The call is native. protocol(Tool.method, ...) gives a coroutine for an ordinary method and an async iterator for an async generator:

async for item in protocol(Tool.items):
    handle(item)

Annotate the method as returning AsyncIterator[T], or the type checker reads the call as an ordinary one.

Frame fragmentation keeps a long stream from holding the channel, so ordinary calls keep their progress while a stream runs.

Flow control is a permission budget. The sender may have MAX_STREAM_WINDOW serialized bytes and MAX_STREAM_BACKLOG items in flight, and the consumer gives both back in batches when it asks for the next item. A slow consumer therefore slows the sender instead of filling memory. The count limit goes with the byte budget, so a long run of small items cannot pass it.

The budget counts the serialized size of an item before compression. That is not the memory of the Python object, which can be larger or smaller, but it is what the peer has to hold and what the wire carries.

An item above MAX_STREAM_ITEM is refused with a ValueError, because it could never find room. An item that is merely larger than the whole budget still goes out, alone, once nothing else is in flight.

The limits are enforced again on the receive side, as the last defence against a peer that ignores the budget. Such a peer ends the stream with a RuntimeError.

How many streams one connection carries

The items that a sender has ready travel in one packet, so a write on one side and a read on the other serve a burst instead of serving every item. The budget of the stream bounds that packet, so one of them holds at most 256 items or 4 MiB. Measured on a local transport with 64-byte items, 16 384 of them in total, by python -m benchmarks.streams, with both ways in alternating runs:

streams	µs per item	before	reads per item	before	local CPU µs	before
1	3.01	7.26	0.00	0.01	0.61	6.71
4	2.03	6.82	0.00	0.00	0.61	6.71
8	2.08	16.31	0.00	0.69	0.61	12.21
16	2.11	15.63	0.00	0.63	0.61	11.60
32	2.06	16.26	0.00	0.71	0.61	12.21

The “before” columns are the measurement of one packet per item. A connection then carried a handful of streams at full speed and twice the cost beyond it, because no stream filled its window any more and every item paid a syscall on both sides. Nothing of that is left: the cost per item no longer depends on how many streams share the connection.

Equal streams still do not finish together. The last of thirty-two finishes about 40 percent of the run after the first, against 33 percent with one packet per item, because a stream that is served sends a whole batch. The run itself is eight times shorter, so the difference between the first and the last finish fell from 87 ms to 15 ms. A smaller batch makes the finishes closer and the run longer: a limit of eight items per packet gives a spread of 17 percent and 3.66 µs per item. Where every stream must advance evenly, give each one its own connection.

When a stream is the right shape

The same ten thousand small items, over one local connection, in three shapes:

shape	total	per item
one call returning a list	4.5 ms	0.45 µs
a stream	91.1 ms	9.11 µs
one call per item	1 994 ms	199 µs

A call per item is never the answer: a round trip costs about 200 µs whatever it carries. One call with a list is the cheapest shape while the result fits in memory on both sides and the caller needs all of it: it serializes and compresses once. A stream earns its keep when the result must not be held whole, when the items appear over time, or when the consumer has to see the first ones before the last ones exist. The gap narrows as items grow, because the per-item cost stops dominating.

What a stream does not break

These properties were measured and need no change.

An ordinary call keeps its progress while a stream of 1 MiB items runs, because every packet travels as 64 KiB frames and the write lock is released between them. Latency of a call that returns at once, 60 samples on a free channel and during such a stream:

channel	minimum	p50	p95	maximum
free	0.181 ms	0.189 ms	0.215 ms	0.236 ms
a stream of 1 MiB items	0.184 ms	0.203 ms	0.310 ms	0.449 ms

A slow consumer costs the producer nothing beyond its own pace, and neither side spins while it waits. Two hundred items of 32 KiB with a pause per item:

pause per item	wall	the pauses alone	local CPU	remote CPU
none	0.006 s	0 s	0.010 s	0.006 s
1 ms	0.256 s	0.200 s	0.000 s	0.008 s
5 ms	1.211 s	1.000 s	0.040 s	0.031 s

The consumer returns permission in batches of half the budget, which a sweep of batch sizes showed to be on a plateau: returning it per item costs twice as much per item, a batch of 64 is within noise of the current one, and a batch of 255 loses 11 percent because it nearly equals the window of 256 items, so the sender runs out and waits. That sweep ran under load from other work on the machine, so its ratios are the reliable part. Do not change the rule.

To leave the loop tells the sender to stop, so an abandoned stream frees the peer. A bare break does not close the iterator at once, so use contextlib.aclosing when the close has to be immediate:

async with contextlib.aclosing(protocol(Tool.items)) as items:
    async for item in items:
        if enough(item):
            break

Closing the iterator sends a stop notification. That notification releases the sender’s permission and cancels the task that produces the items. The remote generator therefore receives the cancellation wherever it waits, its finally runs, and the task leaves the peer. A generator that waits for data it never receives ends the same way as one that keeps producing.

Two limits remain. Synchronous code inside the method cannot be interrupted: the cancellation arrives at the next await, so a method that blocks a thread keeps blocking it. And a resource that the method holds between calls needs its own call to release it, exactly as with an ordinary call:

key = await protocol(Vty.open, "/bin/sh")
async with contextlib.aclosing(protocol(Vty.output, key)) as output:
    async for chunk in output:
        if enough(chunk):
            break
# The session waits for more output, so a separate call releases it.
await protocol(Vty.close, key)

What the wire layer costs

Measured on a local transport with both sides on one quiet machine, best of 9, by python -m benchmarks.wire. The round trip of a call that returns its own argument, with the part this side spends before the bytes leave:

payload	round trip	serialize	send	everything else
8 B	167 µs	3.0 µs	3.1 µs	161 µs
100 B	159 µs	2.3	2.6	154
1 KiB	208 µs	3.6	13.9	190
16 KiB	244 µs	3.6	24.7	216
256 KiB	742 µs	12.0	230	501
1 MiB	2 167 µs	38.5	803	1 326

The floor is about 160 µs, of which this side spends 6: the rest is two passes through a pipe and the whole remote side, including the worker thread that a synchronous Tool method runs in. A method marked with rmote.protocol.inline() skips that thread.

One connection serves many calls at once, because a call waits for bytes rather than for the processor:

calls in flight	per call	calls per second
1	233 µs	4 300
8	69 µs	14 400
64	42 µs	23 900

Bytes per second for content that compresses and for content that does not. The sender refuses to compress dense data, so random bytes are faster than text:

payload	content	upload	download
1 MiB	compressible	845 MiB/s	871 MiB/s
1 MiB	random	1 150 MiB/s	1 162 MiB/s
8 MiB	compressible	1 001 MiB/s	981 MiB/s
8 MiB	random	1 385 MiB/s	1 284 MiB/s

Two parts of the wire layer were measured and are not worth tuning. The packet header costs 43 ns to pack and 33 ns to unpack, which is 0.06 percent of the round trip of a small call. Slicing a payload into frames costs 1.5 µs per frame, and the join on the other side costs less than that.

What a remote log record costs

A record that a Tool method writes reaches the local handlers of the caller. It travels inside the response of that call, so it needs no packet of its own. Measured by python -m benchmarks.logs on the same quiet machine, with one record per call and a local handler that keeps nothing. Both ways share one process and the rounds alternate, so the load of the machine hits them equally:

threads	the record travels	calls per second	share of the run without records	loop CPU per record	records with a packet
8	with the response	11 982	0.80	9.1 µs	4%
8	in its own packet	11 444	0.77	14.1 µs	70%
8	filtered on arrival	12 873	0.86	4.8 µs	3%
32	with the response	17 723	0.86	3.5 µs	3%
32	in its own packet	17 666	0.86	4.0 µs	19%
32	filtered on arrival	18 484	0.90	3.0 µs	3%

A record keeps a packet of its own when it cannot wait for the response: when another batch is on the wire, or when the batch is above the limits below.

The level of the local logger is tested where the packet arrives. A record that no local handler accepts is dropped there, so it costs no hand-off to the delivery thread and no logging.LogRecord: raise the level of rmote.remote.<name> for the records you do not want, and they become almost free. That price, 0.86 of the run at eight threads, is also what the wire alone costs: with the delivery replaced by an empty function the share is the same. A record a handler does want therefore costs 0.06 of the run more, which is the hand-off and the work of the handlers; they run in the delivery thread and compete with the loop thread for the interpreter.

A burst of more than RemoteLogHandler.ATTACH_RECORDS records, or one above ATTACH_BYTES of text, keeps a packet of its own, so a response never becomes a fragmented packet because of the records it carries. A record of a call that runs long waits no more than ATTACH_DELAY for a response and then travels alone. Records of one logger always arrive in the order they were written.

Frame fragmentation

A logical packet travels as one or more frames. Every frame repeats the logical flags of the packet. The FRAGMENT flag marks each frame except the last one, so the receiver knows that more data follows.

BaseProtocol.FRAGMENT_SIZE sets the largest payload in one frame. The sender serializes the whole packet first, then writes the fragments and compresses each body as it goes. It releases the write lock between frames. Packets with a different packet_id therefore interleave on the wire, and one large transfer cannot hold the channel for its whole duration. A frame that arrives above FRAGMENT_SIZE, before or after inflation, is refused: no sender of this protocol writes one.

The receiver keeps one buffer per packet_id in a rmote.protocol.FragmentBuffer. A packet that completes first is returned first, in any fragment order. The buffer enforces MAX_REASSEMBLY_BYTES and MAX_PARTIAL_PACKETS and raises ValueError when a peer passes a limit.

Fragments of one packet_id must come from one task. The receiver joins them in arrival order, so concurrent senders on the same id would mix their bytes.

Frame compression

Each frame body is compressed on its own, against the history of the connection. One zlib dictionary serves each direction, at BaseProtocol.COMPRESSION_LEVEL, and every compressed body ends with a sync flush. A body that repeats the words of an earlier one therefore carries a reference instead of the words, which is what makes a small packet cheap: the method name, the flags and the shape of a result are the same in every call.

COMPRESSED marks the bodies that went through the dictionary. The receiver inflates exactly those, in the order they arrive, before the fragments of a packet are joined. The bootstrap and the ready boundary travel plain, and the boundary opens the codec of each direction.

Small incoming payloads decode on the event loop. Pickle decoding above decode_threshold (32 KiB of decoded bytes by default) uses the default executor. Pass this keyword to the Protocol constructor or its factories, or to Connection.from_local, from_command or from_ssh. Factories apply the value to both endpoints. Positive integers set the threshold; -1 keeps all receive decoding on the event loop, including fragmented packets. Zero and values below -1 are rejected. For example:

async with await Protocol.from_command("ssh", "-T", "host", decode_threshold=-1) as remote:
    ...

A single frame inflates inline, bounded by FRAGMENT_SIZE (64 KiB). Once decoded fragments of a packet exceed the threshold, subsequent frames decompress and decode off the loop in groups of at most BaseProtocol.DECODE_FRAMES (16 frames), stopping at the first completed packet. The read lock preserves dictionary order and waits for an active decoder before propagating cancellation. Stream credits and reassembly limits still bound buffering.

Packets carrying a MODULES source envelope install and unpickle on the event loop. Later large packets can import registered modules and run reconstruction hooks in an executor thread; these hooks must not require a running event loop. The hand-off costs throughput for large batches in exchange for shorter loop pauses. python -m benchmarks.receive measures both; use --burst 38 to replay simultaneous bursts at the same average rate, and --rate 0 for peak throughput. Compare --decode-threshold -1 with the default to measure the tradeoff.

A body whose first DENSITY_SAMPLE bytes use more than DENSITY_LIMIT byte values travels raw and leaves the dictionary untouched, so random and already compressed data costs one short sample. No trial compression runs, because a copy of the dictionary costs more than the pass it would test: the deflate state holds its whole window. A body that passes the sample and still does not shrink therefore travels compressed and grows by the few bytes of its deflate block.

A tool that knows its data cannot shrink says so, and the call carries that decision to the request, the response and the items of a stream:

value = await remote.uncompressed(Tool.method, *args, **kwargs)
async for item in remote.uncompressed(Tool.items):
    handle(item)

FileSync decides from the name and the size of the file, so an archive or an image is not compressed twice while text and JSON still are.

Measured on a local transport by python -m benchmarks.wire and a byte count of the same calls, against the policy that compressed each packet on its own, in alternating runs:

workload	packet policy	frame codec
100 empty calls	201–203 µs, 10 700 B	210 µs, 3 000 B
100 calls of 1 KiB JSON	225–227 µs, 20 300 B	210–213 µs, 4 177 B
5 echoes of 1 MiB of text	2.60–2.72 ms, 34 000 B	2.62–2.64 ms, 37 938 B
5 echoes of 1 MiB of random	1.89–1.93 ms, 5 245 220 B	1.91–1.95 ms, 5 244 800 B
calls per second, 1 in flight	3 956–4 107	3 864–3 922
calls per second, 8 in flight	17 093–17 293	15 038–16 178
calls per second, 64 in flight	30 985–31 260	27 047
stream items, 8 streams	1.71–1.75 µs	1.77 µs

Small traffic costs three to five times fewer bytes, because the dictionary answers for what repeats. The price is the deflate pass of every frame: one call in flight loses 3 percent of the calls per second, eight lose 8 percent and sixty-four lose 13 percent. Bulk transfers and streams keep their speed, and bulk text costs 12 percent more bytes, because the codec flushes every frame while one gzip pass had the whole packet. Random data is unchanged: both policies refuse to compress it.

A local transport carries bytes for free, so these numbers are the cost without the gain. The gain appears where the link is slower than the processor; that case is not measured here.

A connection that needs the old policy sets FRAME_CODEC to False on its protocol class. Both sides run the same source, so they always agree; a peer that was started separately must use the same setting.

A compressed body that does not end at a flush boundary, inflates above FRAGMENT_SIZE, or carries bytes after the end of its stream ends the connection with a ValueError. The dictionary of that direction cannot be repaired, because it no longer matches the sender.

Protocol internals

The clients use these helpers for bootstrap, source transfer, and packet handling. Application code normally calls tools through a connection.

class rmote.protocol.BaseProtocol(reader, writer, *, decode_threshold=32768)

Bases: object

BOUNDARY = b'PROTOCOL READY\n'

COMPRESSION_INLINE = 262144

COMPRESSION_LEVEL = 3

COMPRESSION_RATIO = 0.95

COMPRESSION_SAMPLE = 8192

COMPRESSION_THRESHOLD = 1024

DECODE_FRAMES = 16

DECODE_INLINE = 32768

DECOMPRESSION_INLINE = 32768

DENSITY_LIMIT = 128

DENSITY_SAMPLE = 256

FRAGMENT_SIZE = 65536

FRAME_CODEC: ClassVar[bool] = True

ITEM_HEADER = Struct('>I')

MAGIC = b'RMOTE'

MAX_PARTIAL_PACKETS = 256

MAX_REASSEMBLY_BYTES = 268435456

MAX_START_FAILURE_LINES = 10

MAX_START_FAILURE_TEXT = 2000

MAX_STREAM_BACKLOG = 256

MAX_STREAM_ITEM = 67108864

MAX_STREAM_WINDOW = 4194304

PACKET_HEADER = Struct('>5sIIQ')

classmethod batch(frames)

Join pickled items into one payload, each behind its length.

Return type:

bytes

static check_decode_threshold(value)

Return type:

None

async collect_stderr(reader)

Keep the last lines of reader until it ends.

Return type:

None

compressed(payload)

Compress payload when that pays. None keeps the original bytes.

Three tests, from the cheapest to the most expensive: the byte values of a short sample, a compression pass over a larger sample, and the result of the full pass. The last one is unconditional, so a packet never grows on the wire.

Return type:

bytes | None

decode_frames(frames)

Inflate a bounded group in wire order and decode its final packet.

Only the last frame can complete a packet. A source envelope is handed back as bytes, so its installation and decoding remain on the loop. The caller holds read_lock until this worker finishes.

Return type:

Packet | None

async static decode_in_thread(function, *args)

Run decoding off the loop without releasing its state on cancellation.

Cancelling an executor future cannot stop its function. Keep the caller’s read lock until it finishes, including repeated cancellation, so the next frame never races a worker that still advances the zlib dictionary.

Return type:

Any

decode_packet(payload, flags, packet_id)

Decode one reassembled packet; install source envelopes on the loop.

Return type:

Packet

dense(payload)

True when a short sample uses too many byte values to shrink.

This is a guess, and a cheap one. It can refuse a payload that would have shrunk a little, which costs bytes but no time. It cannot make a packet grow, because the result of every pass is checked as well.

Return type:

bool

encode(value)

Pickle one value, and report the sources the peer does not have.

The caller decides where the sources travel. One value of its own carries them in the envelope of its packet; the items of a batch share one envelope for all of them.

Return type:

tuple[bytes, WSGIEnvironment[Text, ModuleSource]]

async classmethod from_command(*argv, python='python3', stderr=-1, env=None, start_new_session=False, tls=None, decode_threshold=32768)

Bootstrap a Protocol over any command that starts a remote Python REPL.

The interpreter and its flags are appended to argv. For example, from_command("ssh", "-T", "host") runs ssh -T host python3 -qui. An empty argv starts the interpreter on the local host.

Parameters:

• *argv (Text) – Transport command and its arguments. It must pass stdin and stdout through without changing the bytes.

• python (Text) – Python executable to start at the far end.

• stderr (getint) – Where to redirect the remote stderr. The default pipe is read by this side: its last lines explain a failed start, and the rest is dropped. Pass DEVNULL to discard it, or a descriptor to let the caller see it live.

• env (WSGIEnvironment[Text, Text] | None) – Environment for the transport command. None inherits it.

• start_new_session (bool) – Isolate the transport from local terminal signals.

• tls (TLSConfig | None) – Optional controller certificate and remote trust settings.

• decode_threshold (getint) – Decoded packet bytes above which receive uses a worker, on both endpoints. Default 32 KiB; -1 disables workers.

Return type:

Self

Returns:

A connected instance. The caller must still enter its context.

async classmethod from_ssh(host, *, user=None, port=None, identity=None, python='python3', ssh_options=None, stderr=-1, tls=None, decode_threshold=32768)

Bootstrap a Protocol over SSH.

Parameters:

• host (Text) – Remote host, optionally in user@host form.

• user (Text | None) – Remote username (-l). Overrides any user embedded in host.

• port (getint | None) – SSH port (-p).

• identity (Text | None) – Path to an SSH identity file (-i).

• python (Text) – Python executable on the remote host.

• ssh_options (list[Text] | None) – Extra arguments inserted before the host in the ssh command (e.g. ["-o", "StrictHostKeyChecking=no"]).

• stderr (getint) – Where to redirect remote stderr. The default pipe is read by this side, so a refusal of ssh appears in the error of the start.

Return type:

Self

async classmethod from_stdio(logging_level=20, *, decode_threshold=32768)

Return type:

Self

async classmethod from_subprocess(process, *, tls=None, decode_threshold=32768)

Return type:

Self

async pack(payload)

Compress off the loop only when the work pays for the hand-off.

The density check costs about a microsecond, so it runs here: a large payload that cannot shrink is refused without paying for a worker thread at all.

Return type:

bytes | None

async read_boundary()

Wait for the ready boundary, keeping what the peer said before it.

A peer that cannot start writes its reason to this stream, and its interpreter can write a traceback here too. Those lines go into the error, so the caller learns the cause instead of only the symptom.

The boundary also opens the codec of this direction: everything before it travels plain, and every frame after it can carry a compressed body.

Return type:

None

async receive()

Read frames until one logical packet is complete, then return it.

Frames of different packet_id can interleave. Fragments are buffered per packet_id, so a packet that completes first returns first.

Small payloads decode inline; large payloads use the default executor. Source envelopes install and unpickle on the event loop. Cancellation waits for an active worker before releasing the read lock, because its zlib dictionary and fragment buffers belong to this direction.

Return type:

Packet

static report_start_failure(stream, report, error)

Report why this side cannot start, on both channels it still has.

One plain line goes to the protocol stream, before the ready boundary, so the peer reports a cause instead of an unexplained end of connection. The complete traceback goes to the original stderr, for whoever reads the transport. Raw descriptors are used, because the failure can be the channel setup itself.

This writes to the protocol stream only while the connection is being abandoned, so it cannot corrupt a packet. A failed report is ignored: the original failure is what matters.

Return type:

None

async send(packet, flags, packet_id)

Return type:

None

async send_frame(payload, flags, packet_id)

Write one frame as a single unit. Hold the write lock for that frame.

The frame codec compresses the body here, because the dictionary of the direction must move in the order the bodies reach the wire. The work stays on the loop: the lock serializes it anyway, so a worker thread would only add its hand-off to the price.

Return type:

None

async send_payload(payload, flags, packet_id)

Write one logical packet. Split it into fragments when it is large.

Every fragment carries the logical flags. The FRAGMENT flag marks all fragments except the last. The write lock is released between fragments, so packets with a different packet_id interleave instead of waiting.

One packet_id must be sent by one task only. The peer joins its fragments in arrival order, and concurrent senders would mix their bytes.

Return type:

None

async send_serialized(payload, flags, packet_id)

Compress a serialized payload when that pays, then write it.

A streaming sender serializes the item itself, because it needs the size for its budget. This method takes that work and does not repeat it.

The frame codec compresses the body of each frame instead, so this method does nothing while that codec is installed.

Return type:

None

serialize(value, flags)

Put source definitions outside the object pickle, before it is decoded.

Return type:

tuple[bytes, Flags, set[Text]]

classmethod split(payload)

Decode the items of one batch payload, in the order they were sent.

Raises:

ValueError – The lengths do not describe the payload, which means a broken peer. The connection cannot continue after that.

Return type:

list[Any]

classmethod start_failure(said, transport=())

Build the reason a peer did not reach the ready boundary.

Two streams carry evidence. The protocol stream holds what the peer said before the boundary. The stderr of the transport holds what the interpreter or the transport itself said, and that is the only evidence of a failure before our code runs: a missing module, a syntax error of an old interpreter, or ssh that refuses to connect. The tail of each one is kept, because the cause of a traceback is its last line.

Return type:

Text

async start_tls(context, *, server_side, server_hostname=None, timeout=10.0)

Establish TLS before exchanging any rmote boundaries or frames.

Return type:

None

async unpack(payload)

Decompress, off the loop only when the work pays for the hand-off.

The decision uses the compressed size, so the limit stays low: the work is proportional to the result, which a high ratio makes much larger than the frame that carries it.

Return type:

bytes

watch_stderr(reader)

Read the stderr of the transport for as long as it lasts.

The lines of the start go into the failure of the handshake. The reading continues after the start, because a transport that writes more would otherwise fill the pipe and block on its own write. Only the last lines are kept, so a chatty host costs no memory.

Return type:

None

worth_compressing(payload)

Report whether a sample of payload shrinks enough to compress it all.

A sample of the head decides for the whole payload. Random and already compressed data keep their size, and a pass over them only spends processor time.

Return type:

bool

async write_boundary()

Write the ready boundary, then compress what follows it.

Return type:

None

class rmote.protocol.Flags(*values)

Bases: IntFlag

BATCH = 1024

COMPRESSED = 1

EXCEPTION = 32

FRAGMENT = 128

LOG = 64

MODULES = 512

REQUEST = 2

RESPONSE = 4

RPC = 16

STREAM = 256

SYNC = 8

class rmote.protocol.FragmentBuffer(max_bytes, max_packets)

Bases: object

Reassembly buffer for packets that arrive in more than one frame.

Frames of different packets can interleave, so each packet_id keeps its own buffer. The limits stop a faulty peer from exhausting memory.

buffers: dict[int, list[bytes]]

clear()

Drop every buffered fragment.

Return type:

None

sizes: dict[int, int]

store(packet_id, chunk)

Append one non-final fragment. Raise ValueError when a limit is passed.

Return type:

None

take(packet_id, tail)

Join the stored fragments with the final chunk. Release the buffer.

Return type:

bytes

class rmote.protocol.Packet(payload, flags, packet_id, size)

Bases: object

One complete packet that arrived from the peer.

size is the length of the serialized payload before compression. A streaming consumer needs it, to give the sender back the budget it spent.

flags: Flags

packet_id: int

payload: Any

size: int

class rmote.protocol.StreamCredit(max_bytes, max_items, bytes_in_flight=0, items_in_flight=0, released=False, ready=<factory>)

Bases: object

Permission to send the items of one streaming response.

The budget is the serialized size of the items in flight, measured before compression. That is not the memory of the Python objects, which can be larger or smaller, but it is what the peer has to hold and what the wire carries.

A count limit goes with the budget, so a long run of small items cannot pass it. The consumer gives both back as it takes the items.

bytes_in_flight: int

fits(size)

True when one more item of size bytes may go out now.

An item larger than the whole budget goes out alone, once nothing else is in flight. It therefore never waits for room that cannot appear.

Return type:

bool

give_back(size, items)

Take back the budget of items the consumer has taken.

Raises:

ValueError – The peer returned more than it holds, or a negative amount. Both mean a broken peer, so the stream must not go on.

Return type:

None

items_in_flight: int

max_bytes: int

max_items: int

ready: Event

release()

Stop the sender, because the consumer left or the link is gone.

Return type:

None

released: bool

async spend(size)

Wait for permission for one item of size bytes.

Return type:

bool

Returns:

True when the item may go out. False when the sender must stop, because the consumer left or the connection is gone.

rmote.protocol.streaming_method(tool)

True when tool is an async generator, also behind a descriptor.

A static method or a class method hides the function in __func__.

Return type:

bool

rmote.protocol.tool_to_dict(cls, known=())

Describe a file-based class by its module bundle, or an inline class by source.

known names the modules the peer already holds. Every class of a package shares one bundle, so its sources travel once for a connection and the classes that follow carry only what is still missing.

Return type:

WSGIEnvironment[Text, Any]

rmote.protocol.tool_key(definition)

Give the name a tool definition is registered under, on either side.

A definition that names a module keeps the module and the qualified name, so two modules can hold a class of the same name. An inline definition carries source and no module, so its own name is the key. Both sides take the key from the definition, and they cannot disagree about it.

Return type:

Text

rmote.protocol.tool_from_dict(data, context=None)

Restore file-based classes through normal imports; execute inline source separately.

Return type:

type[Tool]

rmote.protocol.bootstrap_packer(code)

Return type:

bytes

class rmote.protocol.ModuleBundle

Bases: object

Build a source bundle for a module, or a package and its descendants.

A module can declare __tool_package__ = 'my_package' to include that package instead. Imports outside this boundary must exist on the target. Only regular filesystem packages and Python source files are supported.

classmethod build(module, seen=None)

Return the bundle of module, through the cache when it can.

A call without seen asks for a whole bundle, which depends only on the root and is therefore cached. A recursive call carries the modules its caller already has, so its result is partial and is not cached.

The caller receives a copy, because the cached mapping is read-only.

Return type:

WSGIEnvironment[Text, ModuleSource]

classmethod bundle(cls, module)

Cache the whole bundle of a root; restart after source edits.

Treat the returned mapping and its source records as read-only.

Return type:

MappingProxyType[Text, ModuleSource]

classmethod collect(cls, root)

Cache sources for the root resolved by build; restart after source edits.

Treat the returned mapping and its source records as read-only.

Return type:

MappingProxyType[Text, ModuleSource]

static declared_module(value)

Return type:

ModuleType | None

static declaring(module)

Return module when it declares a package to transfer.

Return type:

ModuleType | None

classmethod expand(module, seen)

Collect the sources of a root and of everything it references.

Return type:

WSGIEnvironment[Text, ModuleSource]

static owner_module(owner)

Cache the declaring module of a class or a function.

The answer depends only on the owner, and a module does not gain the declaration while the process runs. The cache therefore holds one entry for each type that a payload carries, not one for each value.

Return type:

ModuleType | None

classmethod references(source, module, seen=None)

Find explicitly transferable dependencies referenced by Tool source.

Return type:

tuple[WSGIEnvironment[Text, ModuleSource], WSGIEnvironment[Text, tuple[Text, Text]]]

classmethod root(module)

Resolve the package whose bundle carries module.

Every module of one package resolves to the same root, so the cache below holds one entry for the package and not one for each caller.

Return type:

ModuleType

static source(path, *, package)

Return type:

ModuleSource

class rmote.protocol.ModuleLoader

Bases: MetaPathFinder, Loader

Import transferred sources in memory using Python’s normal import locks.

Sources are immutable within a remote interpreter. Failed modules follow ordinary import semantics: they can be retried; successful dependencies stay loaded. Parent packages outside the bundle are empty namespaces if absent.

create_module(spec)

Return a module to initialize and into which to load.

This method should raise ImportError if anything prevents it from creating a new module. It may return None to indicate that the spec should create the new module.

Return type:

None

exec_module(module)

Return type:

None

find_spec(fullname, path=None, target=None)

Return type:

ModuleSpec | None

get_source(fullname)

Expose in-memory source for tracebacks and inspect.getsource.

Return type:

Text

classmethod install(sources)

Return type:

ModuleLoader

classmethod load(name, sources)

Return type:

ModuleType

lock: ClassVar[RLock] = <unlocked _thread.RLock object owner=0 count=0>

sources: dict[str, ModuleSource]

class rmote.protocol.ModulePickler(stream, known)

Bases: Pickler

Discover explicitly transferable modules in arguments and return values.

persistent_id(value)

Return type:

None

sources: dict[str, ModuleSource]