[
https://issues.apache.org/jira/browse/IGNITE-29206?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dmitry Werner updated IGNITE-29206:
-----------------------------------
Description:
*Problem*
DirectMessageWriter serializes @Compress fields and pending raw fields into a
reusable heap scratch buffer (tmpBuf) through a nested tmpWriter. The whole
field is accumulated there first, and only then compressed or copied out:
- The buffer starts at 10 KB and doubles until the whole field fits, copying
its content on each step.
- It is never shrunk, and the nested tmpWriter streams keep a reference to it
too.
A writer is long-lived:
- communication writers are stored in the NIO session meta and live as long as
the session;
- the discovery ServerImpl.cliMsgSer lives as long as the node.
So every writer ends up holding a buffer as large as the biggest field it has
ever serialized, roughly up to twice that field's raw size. Growing it also
allocates and copies about twice the field size, and produces large (humongous
in G1) arrays.
The exchange coordinator is hit hardest. It sends GridDhtPartitionsFullMessage
to every node, and on large clusters compressed fields such as locParts can
reach tens of megabytes of raw data. After a PME, each of the N communication
sessions keeps such a buffer. On large clusters this can add up to hundreds of
megabytes of heap that is never released. Other nodes are affected the same way
by GridDhtPartitionsSingleMessage, and discovery by DiscoveryDataPacket.
In addition, if serialization into the temporary buffer throws, tmpWriter keeps
stale nested state: map/collection iterators, keyDone flags and per-item state.
reset() only resets the top state item, not the stream fields. A later field
written through the same writer would continue from that stale state and
produce corrupted output. Current callers tear the writer down after such an
error, so this is hardening of the writer's own invariant.
*Fix*
- Stream serialization through a fixed buffer. The temporary buffer has a fixed
size of 10 KB and never grows. The field is serialized into it piece by piece,
the same way messages are written into a socket buffer. Each time the buffer
fills up, or the field is finished, its content is handed to a piece consumer,
and the buffer is reused for the next piece.
- Compress on the fly. For @Compress fields the consumer is the new
CompressedMessage.Compressor. It deflates every piece as it comes, so the raw
data is never held in one buffer. The raw size is counted along the way. It is
written before the chunks, but the whole CompressedMessage is built before
anything goes to the wire, so the wire format is unchanged. The old
CompressedMessage(ByteBuffer, int) constructor is removed.
- Raw fields. For raw fields the pieces are collected into a byte array.
- Failure handling. If serialization fails, tmpWriter is dropped, so a writer
with half-written state is never reused. This mirrors how DirectMessageReader
drops tmpReader.
- Effect. Memory retained per writer is a constant 10 KB instead of the largest
field ever sent. Large fields no longer pay for buffer growth and copying.
Small fields (under 10 KB, the vast majority) are serialized and compressed
exactly as before.
The read side (DirectMessageReader) is unchanged. The decompressed payload is
allocated per message, releasePayload() resets the stream references after
reading, and tmpReader is dropped on failure.
*Tests*
- CompressedMessageTest#testTemporaryBufferDoesNotGrow writes a small message
and then a large GridDhtPartitionsFullMessage through the same writer. The
compressed size of the large one exceeds the buffer capacity, so it is
serialized in several pieces. The test checks that the temporary buffer keeps
its initial capacity and that the message reads back correctly.
- CompressedMessageTest#testWriterReusableAfterFailedCompressedSerialization
makes serialization of a compressed field fail in the middle of a nested map.
It checks that the next message written through the same writer reads back
correctly.
- DirectMarshallingMessagesTest#testLargeRawField serializes a raw field larger
than the temporary buffer, so it is written in several pieces, and checks that
it reads back unchanged.
- Existing tests are adapted to CompressedMessage.Compressor. Serialization
tests pass: CompressedMessageTest, DirectMarshallingMessagesTest,
RollingUpgradeMessageSerializationTest, *MessageSerializationTest
was:
*Problem*
DirectMessageWriter serializes @Compress fields and pending raw fields into a
reusable heap scratch buffer (tmpBuf) with a nested tmpWriter. The data is then
compressed or copied out of it. A writer is long-lived:
- communication writers are stored in the NIO session meta and live as long as
the session;
- the discovery ServerImpl.cliMsgSer lives as long as the node.
This causes one problem and one robustness gap.
1. The oversized buffer is retained per writer.
- The buffer starts at 10 KB and doubles until the whole field fits.
- It is never shrunk.
- The nested tmpWriter streams keep a reference to it too.
So every writer ends up holding a buffer as large as the biggest field it has
ever serialized, roughly up to twice that field's raw size.
The exchange coordinator is hit hardest. It sends GridDhtPartitionsFullMessage
to every node, and on large clusters compressed fields such as locParts can
reach tens of megabytes of raw data. After a PME, each of the N communication
sessions keeps such a buffer. On large clusters this can add up to hundreds of
megabytes of heap that is never released. Other nodes are affected the same way
by GridDhtPartitionsSingleMessage, and discovery by DiscoveryDataPacket.
2. A half-written tmpWriter would be reused after a failure.
If serialization into the temporary buffer throws, tmpWriter keeps stale nested
state: map/collection iterators, keyDone flags and per-item state. reset() only
resets the top state item, not the stream fields. A subsequent compressed field
written through the same writer would continue from that stale state and
produce corrupted output, and the buffer would stay pinned as well.
Current callers tear the writer down after such an error: NIO closes the
session, and discovery closes the session or stops the worker. So this is
hardening of the writer's own invariant, not a fix for a reproducible failure.
The read side (DirectMessageReader) has neither issue. The decompressed payload
is allocated per message, releasePayload() resets the stream references after
reading, and tmpReader is dropped on failure.
*Fix*
- Once the temporary buffer has been used, release it if its capacity is above
320 KB (a doubling step of the initial 10 KB capacity). "Used" means deflated
into the CompressedMessage chunks or copied into a RawField. For compressed
fields the release also happens if compression fails.
- On release, reset the nested tmpWriter and all its state-item streams to a
shared empty buffer, so nothing keeps the large array reachable. This mirrors
DirectMessageReader#releasePayload().
- If serialization into the temporary buffer fails, drop both tmpWriter and
tmpBuf, so a writer with half-written state is never reused. This mirrors how
DirectMessageReader drops tmpReader.
- Buffers up to 320 KB are still reused, so small and medium fields keep
serializing with no per-message allocations.
- An oversized field now grows a fresh buffer each time. That copying is small
compared to serializing and compressing such a field.
- The wire format is unchanged.
*Tests*
- CompressedMessageTest#testLargeTemporaryBufferReleased writes a small
message, then a large GridDhtPartitionsFullMessage twice, through the same
writer. After each write it checks that neither the writer nor its tmpWriter
streams hold a buffer above the limit, and that the message reads back
correctly.
- CompressedMessageTest#testWriterReusableAfterFailedCompressedSerialization
makes serialization of a compressed field fail in the middle of a nested map.
It checks that tmpWriter and tmpBuf are dropped and that the next message
written through the same writer reads back correctly.
- Both new tests fail without the fix.
- Existing serialization tests pass: CompressedMessageTest,
DirectMarshallingMessagesTest, RollingUpgradeMessageSerializationTest,
*MessageSerializationTest.
> DirectMessageWriter accumulates whole compressed and raw fields in an
> ever-growing temporary buffer
> ---------------------------------------------------------------------------------------------------
>
> Key: IGNITE-29206
> URL: https://issues.apache.org/jira/browse/IGNITE-29206
> Project: Ignite
> Issue Type: Bug
> Reporter: Dmitry Werner
> Assignee: Dmitry Werner
> Priority: Major
> Time Spent: 10m
> Remaining Estimate: 0h
>
> *Problem*
> DirectMessageWriter serializes @Compress fields and pending raw fields into a
> reusable heap scratch buffer (tmpBuf) through a nested tmpWriter. The whole
> field is accumulated there first, and only then compressed or copied out:
> - The buffer starts at 10 KB and doubles until the whole field fits, copying
> its content on each step.
> - It is never shrunk, and the nested tmpWriter streams keep a reference to it
> too.
> A writer is long-lived:
> - communication writers are stored in the NIO session meta and live as long
> as the session;
> - the discovery ServerImpl.cliMsgSer lives as long as the node.
> So every writer ends up holding a buffer as large as the biggest field it has
> ever serialized, roughly up to twice that field's raw size. Growing it also
> allocates and copies about twice the field size, and produces large
> (humongous in G1) arrays.
> The exchange coordinator is hit hardest. It sends
> GridDhtPartitionsFullMessage to every node, and on large clusters compressed
> fields such as locParts can reach tens of megabytes of raw data. After a PME,
> each of the N communication sessions keeps such a buffer. On large clusters
> this can add up to hundreds of megabytes of heap that is never released.
> Other nodes are affected the same way by GridDhtPartitionsSingleMessage, and
> discovery by DiscoveryDataPacket.
> In addition, if serialization into the temporary buffer throws, tmpWriter
> keeps stale nested state: map/collection iterators, keyDone flags and
> per-item state. reset() only resets the top state item, not the stream
> fields. A later field written through the same writer would continue from
> that stale state and produce corrupted output. Current callers tear the
> writer down after such an error, so this is hardening of the writer's own
> invariant.
> *Fix*
> - Stream serialization through a fixed buffer. The temporary buffer has a
> fixed size of 10 KB and never grows. The field is serialized into it piece by
> piece, the same way messages are written into a socket buffer. Each time the
> buffer fills up, or the field is finished, its content is handed to a piece
> consumer, and the buffer is reused for the next piece.
> - Compress on the fly. For @Compress fields the consumer is the new
> CompressedMessage.Compressor. It deflates every piece as it comes, so the raw
> data is never held in one buffer. The raw size is counted along the way. It
> is written before the chunks, but the whole CompressedMessage is built before
> anything goes to the wire, so the wire format is unchanged. The old
> CompressedMessage(ByteBuffer, int) constructor is removed.
> - Raw fields. For raw fields the pieces are collected into a byte array.
> - Failure handling. If serialization fails, tmpWriter is dropped, so a writer
> with half-written state is never reused. This mirrors how DirectMessageReader
> drops tmpReader.
> - Effect. Memory retained per writer is a constant 10 KB instead of the
> largest field ever sent. Large fields no longer pay for buffer growth and
> copying. Small fields (under 10 KB, the vast majority) are serialized and
> compressed exactly as before.
> The read side (DirectMessageReader) is unchanged. The decompressed payload is
> allocated per message, releasePayload() resets the stream references after
> reading, and tmpReader is dropped on failure.
> *Tests*
> - CompressedMessageTest#testTemporaryBufferDoesNotGrow writes a small message
> and then a large GridDhtPartitionsFullMessage through the same writer. The
> compressed size of the large one exceeds the buffer capacity, so it is
> serialized in several pieces. The test checks that the temporary buffer keeps
> its initial capacity and that the message reads back correctly.
> - CompressedMessageTest#testWriterReusableAfterFailedCompressedSerialization
> makes serialization of a compressed field fail in the middle of a nested map.
> It checks that the next message written through the same writer reads back
> correctly.
> - DirectMarshallingMessagesTest#testLargeRawField serializes a raw field
> larger than the temporary buffer, so it is written in several pieces, and
> checks that it reads back unchanged.
> - Existing tests are adapted to CompressedMessage.Compressor. Serialization
> tests pass: CompressedMessageTest, DirectMarshallingMessagesTest,
> RollingUpgradeMessageSerializationTest, *MessageSerializationTest
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
