[
https://issues.apache.org/jira/browse/IGNITE-29206?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dmitry Werner updated IGNITE-29206:
-----------------------------------
Description:
*Motivation*
DirectMessageWriter serializes @Compress and pending raw fields by accumulating
the whole field in a reusable heap buffer (tmpBuf), which doubles until the
field fits and is never shrunk. Writers are long-lived (per NIO session for
communication, per node for discovery), so each one retains a buffer up to ~2x
the largest field it has ever serialized. Growing it also copies the data and
allocates humongous arrays.
The exchange coordinator is hit hardest: compressed fields of
GridDhtPartitionsFullMessage can reach tens of megabytes, and after a PME every
communication session keeps such a buffer, which can add up to hundreds of
megabytes of heap on large clusters. The retention was introduced in
IGNITE-28836.
Also, if serialization throws, tmpWriter keeps stale nested state that a later
field could continue from. Current callers drop the writer after an error, so
this is hardening.
*Changes*
The temporary buffer has a fixed size of 10 KB. The field is serialized
into it piece by piece, like into a socket buffer, and each piece is handed to
a consumer.
@Compress fields are deflated on the fly by the new
CompressedMessage.Compressor, so the raw data is never held in one buffer. The
wire format is unchanged.
Raw fields are collected piece by piece and joined into a single byte array.
If serialization fails, tmpWriter is dropped, as DirectMessageReader
already does with tmpReader.
Effect: each writer retains a constant 10 KB instead of the largest field ever
sent. Small fields (the vast majority) fit in a single piece, as before.
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.
> Serialize compressed and raw fields in DirectMessageWriter through a
> fixed-size buffer
> --------------------------------------------------------------------------------------
>
> Key: IGNITE-29206
> URL: https://issues.apache.org/jira/browse/IGNITE-29206
> Project: Ignite
> Issue Type: Improvement
> Reporter: Dmitry Werner
> Assignee: Dmitry Werner
> Priority: Major
> Labels: ise
> Time Spent: 20m
> Remaining Estimate: 0h
>
> *Motivation*
> DirectMessageWriter serializes @Compress and pending raw fields by
> accumulating the whole field in a reusable heap buffer (tmpBuf), which
> doubles until the field fits and is never shrunk. Writers are long-lived (per
> NIO session for communication, per node for discovery), so each one retains a
> buffer up to ~2x the largest field it has ever serialized. Growing it also
> copies the data and allocates humongous arrays.
> The exchange coordinator is hit hardest: compressed fields of
> GridDhtPartitionsFullMessage can reach tens of megabytes, and after a PME
> every communication session keeps such a buffer, which can add up to hundreds
> of megabytes of heap on large clusters. The retention was introduced in
> IGNITE-28836.
> Also, if serialization throws, tmpWriter keeps stale nested state that a
> later field could continue from. Current callers drop the writer after an
> error, so this is hardening.
> *Changes*
> The temporary buffer has a fixed size of 10 KB. The field is serialized
> into it piece by piece, like into a socket buffer, and each piece is handed
> to a consumer.
> @Compress fields are deflated on the fly by the new
> CompressedMessage.Compressor, so the raw data is never held in one buffer.
> The wire format is unchanged.
> Raw fields are collected piece by piece and joined into a single byte
> array.
> If serialization fails, tmpWriter is dropped, as DirectMessageReader
> already does with tmpReader.
> Effect: each writer retains a constant 10 KB instead of the largest field
> ever sent. Small fields (the vast majority) fit in a single piece, as before.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
