This is an automated email from the ASF dual-hosted git repository.
anton-vinogradov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 090bc163559 IGNITE-28836 DirectMessageWriter: reduce per-field
overhead and per-message allocations on the message serialization hot path
(#13296)
090bc163559 is described below
commit 090bc163559688361c1d1ab70e542a9611984bad
Author: Anton Vinogradov <[email protected]>
AuthorDate: Tue Jul 14 20:38:08 2026 +0300
IGNITE-28836 DirectMessageWriter: reduce per-field overhead and per-message
allocations on the message serialization hot path (#13296)
---
.../internal/direct/DirectMessageWriter.java | 118 +++++++++++++--------
1 file changed, 73 insertions(+), 45 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
index 134e11f53cd..319aae7b7e9 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
@@ -67,6 +67,18 @@ public class DirectMessageWriter implements MessageWriter {
/** Buffer for writing. */
private ByteBuffer buf;
+ /**
+ * Cached stream of the current state item. Avoids re-resolving {@code
state.item().stream} on every primitive
+ * write; updated only when the current item changes (buffer set / nested
write enter or exit).
+ */
+ private DirectByteBufferStream curStream;
+
+ /** Reusable writer for compressed payload serialization. Lazily created;
thread-confined like the writer itself. */
+ private DirectMessageWriter tmpWriter;
+
+ /** Reusable scratch buffer for compressed payload serialization. Retained
at the largest size seen so far. */
+ private ByteBuffer tmpBuf;
+
/** @param msgFactory Message factory. */
public DirectMessageWriter(final MessageFactory msgFactory) {
this(msgFactory, DFLT_NETWORK_COMPRESSION);
@@ -91,7 +103,9 @@ public class DirectMessageWriter implements MessageWriter {
@Override public void setBuffer(ByteBuffer buf) {
this.buf = buf;
- state.item().stream.setBuffer(buf);
+ curStream = state.item().stream;
+
+ curStream.setBuffer(buf);
}
/**
@@ -105,7 +119,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeHeader(short type) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeShort(type);
@@ -114,7 +128,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeByte(byte val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeByte(val);
@@ -123,7 +137,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeShort(short val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeShort(val);
@@ -132,7 +146,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeInt(int val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeInt(val);
@@ -141,7 +155,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeLong(long val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeLong(val);
@@ -150,7 +164,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeFloat(float val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeFloat(val);
@@ -159,7 +173,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeDouble(double val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeDouble(val);
@@ -168,7 +182,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeChar(char val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeChar(val);
@@ -177,7 +191,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeBoolean(boolean val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeBoolean(val);
@@ -186,7 +200,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeByteArray(@Nullable byte[] val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeByteArray(val);
@@ -195,7 +209,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeByteArray(byte[] val, long off, int len) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeByteArray(val, off, len);
@@ -204,7 +218,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeShortArray(@Nullable short[] val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeShortArray(val);
@@ -213,7 +227,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeIntArray(@Nullable int[] val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeIntArray(val);
@@ -222,7 +236,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeLongArray(@Nullable long[] val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeLongArray(val);
@@ -231,7 +245,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeLongArray(long[] val, int len) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeLongArray(val, len);
@@ -240,7 +254,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeFloatArray(@Nullable float[] val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeFloatArray(val);
@@ -249,7 +263,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeDoubleArray(@Nullable double[] val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeDoubleArray(val);
@@ -258,7 +272,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeCharArray(@Nullable char[] val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeCharArray(val);
@@ -267,7 +281,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeBooleanArray(@Nullable boolean[] val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeBooleanArray(val);
@@ -276,7 +290,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeString(String val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeString(val);
@@ -285,7 +299,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeBitSet(BitSet val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeBitSet(val);
@@ -294,7 +308,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeUuid(UUID val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeUuid(val);
@@ -303,7 +317,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeIgniteUuid(IgniteUuid val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeIgniteUuid(val);
@@ -312,7 +326,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean
writeAffinityTopologyVersion(AffinityTopologyVersion val) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeAffinityTopologyVersion(val);
@@ -321,11 +335,11 @@ public class DirectMessageWriter implements MessageWriter
{
/** {@inheritDoc} */
@Override public boolean writeMessage(@Nullable Message msg, boolean
compress) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
if (compress)
writeCompressedMessage(
- tmpWriter -> tmpWriter.state.item().stream.writeMessage(msg,
tmpWriter),
+ w -> w.state.item().stream.writeMessage(msg, w),
msg == null,
stream
);
@@ -337,7 +351,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeCacheObject(@Nullable CacheObject obj) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeCacheObject(obj);
@@ -346,7 +360,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeKeyCacheObject(KeyCacheObject obj) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeKeyCacheObject(obj);
@@ -355,7 +369,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeGridLongList(@Nullable GridLongList ll) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeGridLongList(ll);
@@ -364,7 +378,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public <T> boolean writeObjectArray(T[] arr, MessageArrayType
type) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeObjectArray(arr, type, this);
@@ -373,7 +387,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public <T> boolean writeCollection(Collection<T> col,
MessageCollectionType type) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeCollection(col, type, this);
@@ -382,11 +396,11 @@ public class DirectMessageWriter implements MessageWriter
{
/** {@inheritDoc} */
@Override public <K, V> boolean writeMap(Map<K, V> map, MessageMapType
type, boolean compress) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
if (compress)
writeCompressedMessage(
- tmpWriter -> tmpWriter.state.item().stream.writeMap(map, type,
tmpWriter),
+ w -> w.state.item().stream.writeMap(map, type, w),
map == null,
stream
);
@@ -398,7 +412,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeIgniteProductVersion(IgniteProductVersion
ver) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeIgniteProductVersion(ver);
@@ -407,7 +421,7 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean writeGridCacheVersion(GridCacheVersion ver) {
- DirectByteBufferStream stream = state.item().stream;
+ DirectByteBufferStream stream = curStream;
stream.writeGridCacheVersion(ver);
@@ -443,12 +457,18 @@ public class DirectMessageWriter implements MessageWriter
{
@Override public void beforeNestedWrite() {
state.forward();
- state.item().stream.setBuffer(buf);
+ curStream = state.item().stream;
+
+ curStream.setBuffer(buf);
}
/** {@inheritDoc} */
@Override public void afterNestedWrite(boolean finished) {
state.backward(finished);
+
+ curStream = state.item().stream;
+
+ curStream.setBuffer(buf);
}
/** {@inheritDoc} */
@@ -474,9 +494,19 @@ public class DirectMessageWriter implements MessageWriter {
}
if (!stream.serializeFinished()) {
- ByteBuffer tmpBuf = ByteBuffer.allocateDirect(TMP_BUF_CAPACITY);
-
- DirectMessageWriter tmpWriter = new
DirectMessageWriter(msgFactory, compressionLvl);
+ // CompressedMessage consumes the scratch buffer in its
constructor (deflates into its own byte array),
+ // so the buffer never escapes this method and can be reused
across fields; heap is cheaper than direct
+ // here (no native alloc / Cleaner churn).
+ if (tmpBuf == null)
+ tmpBuf = ByteBuffer.allocate(TMP_BUF_CAPACITY);
+ else
+ tmpBuf.clear();
+
+ // Reuse the temp writer across fields/messages instead of
allocating a fresh state stack each time.
+ if (tmpWriter == null)
+ tmpWriter = new DirectMessageWriter(msgFactory,
compressionLvl);
+ else
+ tmpWriter.reset();
tmpWriter.setBuffer(tmpBuf);
@@ -484,14 +514,12 @@ public class DirectMessageWriter implements MessageWriter
{
do {
if (tmpBuf.remaining() <= tmpBuf.capacity() / 10) {
- byte[] bytes = new byte[tmpBuf.position()];
+ ByteBuffer newBuf = ByteBuffer.allocate(tmpBuf.capacity()
* 2);
tmpBuf.flip();
- tmpBuf.get(bytes);
-
- tmpBuf = ByteBuffer.allocateDirect(tmpBuf.capacity() * 2);
+ newBuf.put(tmpBuf);
- tmpBuf.put(bytes);
+ tmpBuf = newBuf;
tmpWriter.setBuffer(tmpBuf);
}
