[
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.
was:
*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
> 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.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
