IGNITE-27977 Refactor bytes serialization for DataStreamerRequest - #13454
Open
anton-vinogradov wants to merge 10 commits into
Open
IGNITE-27977 Refactor bytes serialization for DataStreamerRequest#13454anton-vinogradov wants to merge 10 commits into
anton-vinogradov wants to merge 10 commits into
Conversation
The stream receiver was marshalled by hand: the streamer produced the blob, and the processor unmarshalled it with a class loader it had just built. The pair is now an @Marshalled field, so the generated marshaller owns both directions. The class is marked @UseBinaryMarshaller: the receiver is a user class, and the hand-written call used ctx.marshaller(), which is the same schema-aware marshaller the annotation selects. The wire format is unchanged - updaterBytes stays @order(3). The message stays a DeferredUnmarshalMessage. Its class loader does not come from a carried deployment alone: with forced local deployment it is the grid class loader. The processor therefore passes the loader explicitly and keeps the read inside its own try, so a missing deployment is still answered to the sender instead of leaving it waiting for a timeout. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The class javadoc and the comment at the read said the same thing twice. The javadoc now states what the message is, and the comment states why the read waits for this point. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The field holds a user object, and the request is printed on the sending side under debug logging. The blob it replaced printed as bytes, and GridToStringBuilder rethrows whatever a field toString throws, so a user toString could now break the logging path. GridJobExecuteRequest excludes its user objects the same way. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Handing the message the receiver instead of its bytes cost a marshal per batch, where the streamer used to marshal once and reuse the result. Measured on one node: 281 ns for IsolatedUpdater, 810 ns for StreamTransformer.from(ep), 9.3 us for a receiver holding 10K of state, against 76 us to build and marshal a batch of 512 entries - up to 12% of a batch, and a larger share of a small one. The streamer keeps the bytes the generated marshaller produced for the first request and hands them to the next one, which the marshaller then keeps instead of producing its own. It reuses a result rather than deciding how to obtain it, so the marshaller stays the one codegen picks. The bytes are paired with the receiver they belong to, so a receiver replaced mid-stream invalidates them by itself - no separate cache reset that a concurrent send could race with. This also closes the older mismatch, where the cache was never invalidated at all and a replaced receiver took effect locally but not remotely. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Holding the receiver in the request meant its serialized form belonged to the request, while the object belonged to the streamer. A request is one batch, the receiver lives for the whole stream, so keeping the bytes cost either a marshal per batch or a cache beside the streamer - a cache that had to be invalidated by hand and published safely. The receiver now travels in StreamReceiverMessage, where the object and its bytes sit together and live exactly as long as the receiver does. The streamer holds one instance and puts it into every request, so the generated marshaller fills the bytes for the first batch and the rest find them already there. Replacing the receiver builds another instance, which invalidates the old bytes by construction. This also removes an older race: the receiver field was mutated from the user thread and read by the sending ones without being volatile. The wire format of the request changes: field 3 is now a nested message rather than a byte array. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The streamer exposes receiver(StreamReceiver) yet read the field back through rcvr(), and the request named its carrier field after what the getter returns rather than after what it holds. Paired accessors in these classes share a name - allowOverwrite(), skipStore(), keepBinary() - so the getter is receiver() now, and the carrier is updaterMsg. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The field is written by the batch that is marshalled first and read by the rest, and those batches leave on different threads. Without the keyword a reader could see the reference before the contents, skip the marshalling and send a half-written array. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The collection was held in a local only because the test cleared the static field in a finally block, before the assertions ran. Clearing it in afterTest, next to the other static cleanup of this class, removes both the local and the try/finally. The local inside the SPI stays and is now explained: it reads the volatile field once, since the field is cleared while nodes that are still stopping keep sending through it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Clearing after the test made the field nullable, which cost a null check and a local copy in the SPI, and my comment there claimed a race that did not exist - afterTest clears the field once the grids are already stopped. A final collection cleared in beforeTest gives each test the same clean start with none of that. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Read the receiver once where the deployment aware is built, instead of calling the getter three times in a row; keep the explicit type argument on individual() that the rewrite had dropped; unwrap the sent message once in the test SPI; and say in the request javadoc that the excluded field carries a user object rather than being one. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Possible compatibility issues. Please, check rolling upgrade casesThis PR modifies protected classes (with Order annotation). Affected files:
|
Contributor
Author
|
/runall 🚀 RunAll queued — build 9267111 · live progress & verdict: Ignite PR Checker. The verdict lands here when the run finishes. |
Contributor
Author
|
Ignite PR Checker verdict · RunAll build 9267111 · 147 suites ran, 0 reused
Everything below is what it did manage to say. 🔎 No blockers found — but the run above can't prove the PR is clean. 16 pre-existing/flaky tests filtered out. Re-run once the above is sorted out. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
DataStreamerRequest.updaterByteswas a hand-marshalled blob:DataStreamerImplproducedthe bytes while building the request, and
DataStreamProcessorunmarshalled them with aclass loader it had just resolved.
Carrying the receiver is not new here.
StreamReceiverextendsSerializablebecause itruns on the node that owns the data, and its bytes already travelled in this very message,
in every batch. What changes is who marshals them.
Why this matters
A field that marshals itself picks its own marshaller, so it stays outside whatever the
transport decides — which is what IGNITE-28940 has to decide in one place. The generated
marshaller is that place.
updaterByteswas the last such blob on this path.Change
The receiver travels in
StreamReceiverMessage: an@Marshalledpair of the user objectand its bytes, marked
@UseBinaryMarshallersince it is the one holding a user class. Thehand-written call used
ctx.marshaller()— the very marshaller that annotation selects(
IgniteKernalpasses it as the schema-aware one), so the bytes are produced as before.Putting the two in one carrier is what keeps the object and its serialized form on the same
lifetime. A request is one batch; the receiver lives for the whole stream. The streamer
holds one carrier and puts it into every request, so the generated marshaller fills the
bytes for the first batch and the rest find them in place. Replacing the receiver builds
another carrier, which invalidates the old bytes by construction — there is no cache to
reset, and no window where a reset races with a send.
DataStreamProcessorreads the message throughMessageMarshalling.unmarshaland drops itsown
Marshallerfield.Why not marshal per batch
Measured on one node: 281 ns to marshal
IsolatedUpdater(12 B), 810 ns forStreamTransformer.from(ep)(299 B), 9.3 µs for a receiver holding 10K of state (10270 B),against 76 µs to build and marshal a batch of 512 entries. Marshalling per batch would cost
up to 12% of a full batch and a larger share of a small one, so the saving the old cache
gave is kept — it just no longer picks the marshaller.
DataStreamerImplSelfTest#testReceiverMarshalledOncePerStreamerpins this down: it collectsthe carrier of every request that leaves through the communication SPI, where the message is
already marshalled, and requires all of them to share one byte array. Building a carrier per
request makes it fail.
Why the read stays with the consumer
The message remains a
DeferredUnmarshalMessage. Its class loader does not follow from acarried deployment alone: with forced local deployment it is the grid class loader, and
otherwise it comes from the global deployment of the sender. So the processor passes the
loader explicitly and keeps the read inside its existing
try, where a missing deploymentis reported back to the sender instead of leaving it waiting for a timeout.
Wire format
Field 3 of the request becomes a nested message instead of a byte array, and the carrier
takes the next free id in the datastreamer group.
Along the way
sending ones without being
volatile; holding it in one carrier fixes that publication.The bytes inside the carrier are
volatilefor the same reason: the batch that marshalsfirst writes them, the others read them, and a reader seeing the reference before the
contents would skip the marshalling and send a half-written array.
replaced mid-stream took effect locally, where the live field is read, but not remotely,
where the stale bytes kept going. Both paths now see the same object.
@GridToStringExcludein the request: it holds a user object, the requestis printed under debug logging on the sending side, and
GridToStringBuilderrethrowswhatever a field
toStringthrows.GridJobExecuteRequestexcludes its user objects thesame way.
What this changes for the entries
Reading the message reads all of it, so
entriesnow pass through the generated marshalleron the receiving side too, with the cache object context resolved from
cacheIdand thedeployment class loader.
DataStreamerUpdateJobstill unmarshals them once more under theglobal loader; that pass is now a no-op, since
CacheObject.unmarshalonly acts when thevalue is absent. It is kept because the job also runs on the local path, where no message
was ever unmarshalled, and because the same loop carries the security permission checks.
The streamer also still marshals keys and values by hand before sending. That is not a
duplicate of the generated marshal: it uses the cache object context the streamer holds from
its creation, whereas the generated code resolves the context by
cacheIdand would skip thework if the cache were destroyed in between.
Verified
123 tests, no failures: the whole
processors.datastreamerpackage — withIgniteDataStreamerPerformanceTestleft out as an endless benchmark that always times out —plus
P2PStreamingClassLoaderTest,P2PClassLoadingFailureHandlingTest,GridP2PContinuousDeploymentSelfTest,ClassLoadingProblemExceptionTest,MessageMarshalOnceTest,IgniteCoreMessagesSerializationTest,DirectMarshallingMessagesTestandMessageProcessorTest. Alsomvn checkstyle:check -Pcheckstyle -pl modules/core.🤖 Generated with Claude Code