IGNITE-28853 Optimize serialization and deserialization of CompessedMessage

[wernerdv]

Thank you for submitting the pull request to the Apache Ignite.

In order to streamline the review of the contribution
we ask you to ensure the following steps have been taken:

The Contribution Checklist

• There is a single JIRA ticket related to the pull request.
• The web-link to the pull request is attached to the JIRA ticket.
• The JIRA ticket has the Patch Available state.
• The pull request body describes changes that have been made.
  The description explains WHAT and WHY was made instead of HOW.
• The pull request title is treated as the final commit message.
  The following pattern must be used: IGNITE-XXXX Change summary where XXXX - number of JIRA issue.
• A reviewer has been mentioned through the JIRA comments
  (see the Maintainers list)
• The pull request has been checked by the Teamcity Bot and
  the green visa attached to the JIRA ticket (see tab PR Check at TC.Bot - Instance 1 or TC.Bot - Instance 2)

Notes

• How to Contribute
• Coding abbreviation rules
• Coding Guidelines
• Apache Ignite Teamcity Bot

If you need any help, please email dev@ignite.apache.org or ask anу advice on http://asf.slack.com #ignite channel.

[ignitetcbot]

TCBot Test Analysis

• TeamCity: --> Run :: All Results
• Branch: pull/13329/head
• Base: master
• Tested commit: fa51508

Possible Blockers (0)

No blockers found.

New Tests (2)

• Basic 1: 1 tests
  • IgniteBasicTestSuite: CompressedMessageTest.testReadFailsOnNullChunk - PASSED
• Cache 13: 1 tests
  • IgniteCacheTestSuite13: SystemViewCacheExpiryPolicyTest.testCacheViewExpiryPolicy[factory=javax.cache.configuration.FactoryBuilder$SingletonFactory@689faf79, actual=SingletonFactory [expiryPlc=EternalExpiryPolicy [create=ETERNAL]]] - PASSED

[wernerdv]

Results of the JmhDirectMessageReaderBenchmark:

• master:

Benchmark                                          (entries)   Mode  Cnt     Score      Error  Units
JmhDirectMessageReaderBenchmark.compressedMessage         30  thrpt    5  4595,593 ± 7257,161  ops/s
JmhDirectMessageReaderBenchmark.compressedMessage        500  thrpt    5  1582,391 ±  476,917  ops/s

• current branch:

Benchmark                                          (entries)   Mode  Cnt      Score       Error  Units
JmhDirectMessageReaderBenchmark.compressedMessage         30  thrpt    5  47589,313 ± 23902,679  ops/s
JmhDirectMessageReaderBenchmark.compressedMessage        500  thrpt    5   3216,279 ±   586,795  ops/s

[anton-vinogradov]

Reviewed the whole change (the CompressedMessage/serializer part matches my #13325, so the findings there carry over; the DirectMessageReader part — curStream cache and tmpReader reuse — is new and I looked at it from scratch). The curStream cache itself checks out: all three mutation points (setBuffer / beforeNestedRead / afterNestedRead) keep it in sync, depth only ever changes inside try/finally in DirectByteBufferStream, and every production reset() is followed by setBuffer before the next read. The benchmark compiles with -Pbenchmarks and runs. Found the following, roughly by severity.

1. tmpReader reuse: one partial inner read poisons every subsequent compressed field on the connection (reproduced)

reset() is a shallow reset: DirectMessageState.reset() only zeroes stack[0].state and asserts pos == 0. It does not clear the per-stream parse state (msgTypeDone, half-read msg, readSize/readItems, map iteration, tmpArr) — those are only cleared when a readMessage/readMap finishes successfully — and backward(false) intentionally preserves the state of abandoned deeper items for resume.

The old code was immune by construction (new DirectMessageReader(...) per compressed field). With reuse, this sequence breaks: a CompressedMessage whose inflated payload is logically incomplete (incompatible peer, corruption, understated dataSize — see #4) makes fun.apply(tmpReader) return with lastFinished == false and no exception, so the session stays alive. The next, fully valid compressed field on the same per-connection reader reuses the poisoned tmpReader: readMessage sees the stale msgTypeDone == true, skips the type header of the new payload and resumes the old half-read message at a wrong offset → the valid field reads as null with lastRead == false forever (outer parser waits for bytes that never come), or a silently corrupted message is assembled.

I verified this with a quick repro against the branch classes: after one truncated payload, a following valid field fails to deserialize; with a fresh reader (old behavior) the same field reads fine.

The clean fix falls out of an invariant this method already has: tmpReader's buffer is always the complete payload, so a partial inner read is never "need more bytes" — it is always corruption. Instead of propagating it as lastRead = false:

T res = fun.apply(tmpReader);

if (!tmpReader.state.item().stream.lastFinished()) {
    tmpReader = null; // Don't reuse a reader with half-read state.

    throw new IgniteException("Failed to deserialize compressed payload: " +
        "uncompressed data ended unexpectedly [dataSize=" + msg0.dataSize() + ']');
}

lastRead = true;

This also turns the silent protocol desync from #4 into a localized, diagnosable error, and dropping tmpReader matters because the session is not guaranteed to die: the legacy selector loop (IGNITE_NO_SELECTOR_OPTS=true, GridNioServer.processSelectedKeys) logs the exception and keeps the session — only the optimized path closes it.

2. tmpReader retains the last uncompressed payload for the connection lifetime

tmpReader is a field of the per-connection reader (session meta, GridDirectParser), and after the read its stack[0].stream keeps referencing the ByteBuffer.wrap(uncompressed). For exchange traffic (GridDhtPartitionsFullMessage on large topologies) that's megabytes per connection held until the next compressed message on that connection or TCP session close; on the coordinator — one payload per node connection. The old per-call reader made it garbage immediately. One line after fun.apply:

tmpReader.setBuffer(EMPTY_BUF); // static final ByteBuffer EMPTY_BUF = ByteBuffer.wrap(new byte[0]);

Carried over from #13325 (identical CompressedMessage/serializer code)

These three have no benign trigger — the sender computes dataSize = buf.remaining() correctly by construction — but the reworked receive path made the wire dataSize load-bearing, while master was insensitive to it (tmpBuf grew from actually received chunks, readAllBytes() sized the output, dataSize was assert-only). So on a corrupted stream the failure quality degrades vs master, and the guards are cheap:

3. dataSize > 0 + immediate finalChunk → bare NPE. readFrom() legitimately returns true with chunks == null when the first finalChunk boolean is true (state 1 exits before state 2 runs) — the new null-chunk guard doesn't cover this path. uncompress() then returns null and the caller hits ByteBuffer.wrap(null) → NPE in the NIO worker. Master threw IgniteException (EOFException: Unexpected end of ZLIB input stream) on the same input. Fix in uncompress():

if (chunks == null)
    throw new IgniteException("Compressed stream is truncated [expected=" + dataSize + ", inflated=0]");

(The null return contract is dead: the caller already short-circuits dataSize() == 0.)

4. Understated dataSize truncates silently. Both inflate loops are bounded by off < dataSize, so a valid stream that inflates to more than dataSize is cut short and the off != dataSize check can't fire — it only covers the "shorter" direction, while master's assert uncompressedData.length == dataSize covered both. The consequence is the partial-read cascade from #1. Besides the reader-level guard above, the invariant is cheap to restore locally with a probe inflate after the main loop (hoist the chunk index out of the for):

byte[] probe = new byte[1];

while (!inflater.finished()) {
    if (inflater.needsInput()) {
        if (i == chunks.size())
            break;

        inflater.setInput(chunks.get(i++));
    }

    if (inflater.inflate(probe, 0, 1) > 0)
        throw new IgniteException("Compressed stream is longer than expected [expected=" + dataSize + ']');
}

A well-formed stream just consumes the deflate terminator and finishes here. Worth a unit test next to testReadFailsOnNullChunk: valid compressed payload with the dataSize field patched smaller → expect IgniteException.

5. dataSize drives allocations before any validation. Negative values: dataSize <= -20480 → new ArrayList<>(msg.dataSize / CHUNK_SIZE + 1) → IllegalArgumentException: Illegal Capacity; -1..-20479 → new byte[dataSize] → NegativeArraySizeException. Huge values: dataSize = Integer.MAX_VALUE + one tiny chunk → ~2 GB upfront allocation → OutOfMemoryError → failure handler. Minimal guard in readFrom state 0:

if (msg.dataSize < 0)
    throw new IgniteException("Invalid compressed message data size: " + msg.dataSize);

Optionally bound the allocation by what was actually received (raw deflate can't expand beyond ~1032:1): before new byte[dataSize], check dataSize <= totalCompressedLen * 1032L + 64.

6. Benchmark nit

@Setup relies on assert finished after writeMessage, but asserts are off in the JMH fork by default — if the message ever outgrows the 4 MB buffer (larger entries param), the benchmark will happily measure deserialization of a truncated stub. Make it explicit: if (!finished) throw new IllegalStateException("Message doesn't fit the buffer");

🤖 Generated with Claude Code

[anton-vinogradov]

Prepared ready-to-merge fixes for everything above: wernerdv#3 (targets this PR's branch, so merging it brings the fixes straight here).

Beyond the six review items, self-reviewing the fixes surfaced two more issues that are also addressed there:

• an exception thrown from inside the payload deserialization (not just a silent partial read) left the reused tmpReader with half-read state — the drop is now done in a finally;
• a lying dataSize could still inflate the chunk-list capacity allocation in readFrom() (~1.7 MB of Object[] for a ~15-byte wire message) — the pre-size is now clamped.

Verification: CompressedMessageTest 7/7 (5 new cases), DirectMarshallingMessagesTest green, -Pcheckstyle clean, benchmarks module compiles. Known accepted limitations are listed in the fix PR description.

🤖 Generated with Claude Code