Regression: All subscribers are currently zero-copy.

[griffinmilsap]

Background

Publishers serialize messages to subscribers via TCP or SHM (and sometimes via a module-level dictionary for "local" mode). Backend process then yields a deepcopy of that message to interested subscribers, to do .. well just about anything with.

Subscribers have a feature that allows them to receive a "zero-copy" view of the incoming message WITHOUT that deepcopy. All subscribers in that process (and in the case of SHM channels, all subscribers on the machine) have a (hopefully read-only) view of the same chunk of memory which results in a FAST transfer with the danger that if any subscriber mutates the message, all other subscribers will see that mutation -- which is most-likely REALLY bad and difficult to debug. The idea was to give a higher-performance shortcut to developers who PROMISE not to mutate the object, and they OPT-IN to this functionality by creating an InputStream with zero_copy = True

But...

In the current dev branch zero_copy is ignored. BackendProcess just uses recv_zero_copy every time.

ezmsg/src/ezmsg/core/backendprocess.py

Lines 409 to 425 in eaf8068

	while True:
	if not callables:
	sub.close()
	await sub.wait_closed()
	break
	async with sub.recv_zero_copy() as msg:
	try:
	for callable in list(callables):
	try:
	await callable(msg)
	except (Complete, NormalTermination):
	callables.remove(callable)
	finally:
	del msg
	if len(callables) > 1:
	await asyncio.sleep(0)

This wasn't always the case; As can be seen in this commit from 2022:

ezmsg/ezmsg/core/backendprocess.py

Lines 206 to 213 in 0504747

	while True:
	if getattr(task, ZERO_COPY_ATTR) == False:
	msg = await sub.recv()
	await handle_message(msg)
	else:
	async with sub.recv_zero_copy() as msg:
	await handle_message(msg)
	del msg

Reviewing the git blame, it looks like I introduced this issue sometime in early 2023. I believe I was implementing single-process mode.

What does this mean to me?

This has a few implications:

1. ezmsg was running FASTER than it really should have been. Unless your code is religiously using zero-copy (like most of ezmsg-sigproc), the fix to this issue will result in a deepcopy on every message your subscriber receives, which will negatively affect performance in the name of safety/guardrails.
2. Any system that uses ezmsg 3.2.0+ will have this issue and any subscriber could mutate an incoming message, resulting in message corruption for all other subscribers.
3. A nuance: BackendProcess was still honoring zero_copy = True for a niche deepcopy on publish which would only be triggered if your coroutine yields the same object that it was given. Basically, since the MessageCache was locally transferring the very same object your coroutine publishes, we didn't want that to be a zero-copy view of a message we recieved from an upstream publisher. BUT subscribers were receiving a zero-copy view without even requesting it, so this deepcopy wasn't being triggered on publish.

What is the fix?

We call recv instead of recv_zero_copy when subscribers do not declare zero_copy = True. This internally handles the deepcopy and performance metrics take a hit.

[griffinmilsap]

So individual TASKS define zero_copy in the API; it isn't something set on the InputStream the same way leaky is. There's a one-to-one mapping between OutputStream and Publisher, and since ezmsg 3.2.0, there is now a one-to-one mapping between InputStream and Subscriber. In ezmsg 3.2.0 this was a significant performance enhancement because of the redundant telemetry per-subscriber.

Before ezmsg 3.2.0, there was a one-to-one mapping between input @ez.subscriber coroutine methods and Subscriber, so we would set zero_copy on individual tasks/coroutines.

Possible Fixes

1. A minimal change: we would use recv if the subscriber is leaky, or recv_zero_copy if it isn't, then deepcopy the message again for every coroutine that doesn't define zero_copy = True, so that multiple coroutines in the same unit cannot mutate the same message for each other. This is a LOT of deep copies (sometimes two deep copies if a leaky subscriber does not define a coroutine with ez.subscriber(... zero_copy. = True)
2. A less-minimal change: revert the change in ezmsg 3.2 and add a separate Subscriber per ez.subscriber which is almost zero-performance hit now with MessageChannels. I think this is the proper way to address this, but will not be a trivial change.
3. A huge breaking-change: add the zero_copy kwarg to InputStream where one might expect it to be anyway, but this would break a lot of existing code. This would unfortunately have the same issue where multiple coroutines that get messages from the same InputStream will still share the same messages and could mutate the messages for each other.
4. Wontfix: Things seem... remarkably stable as-is. And this has been the implementation for a long time now. Maybe we deprecate zero_copy and all subscribers remain performant without guardrails, and .. we document the dangers of mishandling input messages.

@cboulay @pperanich @KonradPilch thoughts?

[griffinmilsap]

After sleeping on this, I think the outcome might be wontfix but with the addition of some new debug features

Perhaps we could add some sort of safe mode (enabled with an environment variable) that deep copies every message going to any ez.subscriber coroutine. If your system crashes or acts weird without "safe mode", but the behavior resolves with "safe mode" then you know that one of your coroutines is mutating its input. There could be an additional equality check on the message before and after the ez.subscriber is called, and we can post warnings if the message has mutated.

[cboulay]

When I read the proposed solutions initially, I was already leaning toward the proposed wontfix.

I think we might have to promote some documentation around this, and provide some patterns on how to avoid mutating the input, and on how to write unit tests to verify that the input is not being mutated.
Sadly, I can't think of a generic function or utility to inspect a unit or its processor to make sure it is non-mutating, because each unit will have its own unique requirements for inputs and might have different codepaths that need to be tested.

[griffinmilsap]

Just talked with Preston and he also indicates that he thinks the wontfix route is best too. He suggested that it may be worthwhile to just default zero_copy to True, but this means we continue to support this kwarg at the task level which is counter-intuitive given the subscriber is parameterized at the InputStream/Unit level. I'm still in favor of just deprecating and removing the kwarg.

Preston also indicated that there should be a way to disable the additional deepcopy on publish if a task attempts to republish the exact same object it receives via zero-copy, and additionally handle the removal of this guard rail with documentation.