IGNITE-28473 Concurrent message serialization in NIO workers causes performance degradation

Author: abashev

modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java

@@ -42,6 +42,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Map.Entry;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -93,6 +94,8 @@
 import org.apache.ignite.thread.IgniteThread;
 import org.jetbrains.annotations.Nullable;
 
+import static java.lang.System.arraycopy;
+import static java.util.Optional.empty;
 import static org.apache.ignite.failure.FailureType.CRITICAL_ERROR;
 import static org.apache.ignite.failure.FailureType.SYSTEM_WORKER_TERMINATION;
 import static org.apache.ignite.internal.processors.tracing.SpanTags.SOCKET_WRITE_BYTES;
@@ -647,23 +650,90 @@ IgniteInternalFuture<?> send(GridNioSession ses,
 
         GridSelectorNioSessionImpl impl = (GridSelectorNioSessionImpl)ses;
 
+        byte[] serialized = serializeMessage(ses, msg).orElse(null);
+
         if (createFut) {
             NioOperationFuture<?> fut = new NioOperationFuture<Void>(impl, NioOperation.REQUIRE_WRITE, msg,
                 skipRecoveryPred.apply(msg), ackC);
 
+            fut.serializedMessage(serialized);
+
             send0(impl, fut, false);
 
             return fut;
         }
         else {
-            SessionWriteRequest req = new WriteRequestImpl(ses, msg, skipRecoveryPred.apply(msg), ackC);
+            var req = new WriteRequestImpl(ses, msg, skipRecoveryPred.apply(msg), ackC, serialized);
 
             send0(impl, req, false);
 
             return null;
         }
     }
 
+    /**
+     * Serializes message to byte array eagerly in the calling thread.
+     *
+     * @param ses Session.
+     * @param msg Message to serialize.
+     * @return Serialized message bytes, or empty if eager serialization is not available.
+     */
+    private Optional<byte[]> serializeMessage(GridNioSession ses, Message msg) {
+        if (writerFactory == null || msgFactory == null)
+            return empty();
+
+        MessageWriter writer;
+        MessageSerializer<Message> msgSer;
+
+        try {
+            writer = writerFactory.writer(ses);
+            msgSer = msgFactory.serializer(msg.directType());
+        }
+        catch (Exception e) {
+            if (log.isDebugEnabled())
+                log.debug("Failed to prepare eager serialization, will use lazy path [msg=" + msg + ", err=" + e + ']');
+
+            return empty();
+        }
+
+        try {
+            int capacity = 4096;
+            var buf = ByteBuffer.allocate(capacity);
+
+            while (true) {
+                writer.setBuffer(buf);
+
+                if (msgSer.writeTo(msg, writer)) {
+                    writer.reset();
+
+                    int len = buf.position();
+                    var result = new byte[len];
+
+                    arraycopy(buf.array(), 0, result, 0, len);
+
+                    return Optional.of(result);
+                }
+
+                int pos = buf.position();
+
+                capacity *= 2;
+
+                var newBuf = ByteBuffer.allocate(capacity);
+
+                arraycopy(buf.array(), 0, newBuf.array(), 0, pos);
+
+                newBuf.position(pos);
+
+                buf = newBuf;
+            }
+        }
+        catch (Exception e) {
+            log.warning("Failed to eagerly serialize message, will use lazy path [msg=" + msg + ']', e);
+
+            return empty();
+        }
+    }
+
     /**
      * @param ses Session.
      * @param req Request.
@@ -1614,7 +1684,21 @@ private boolean writeToBuffer(
 
                 int startPos = buf.position();
 
-                if (messageFactory() == null) {
+                byte[] serialized = req.serializedMessage();
+
+                if (serialized != null) {
+                    int off = req.serializedOffset();
+                    int len = Math.min(serialized.length - off, buf.remaining());
+
+                    buf.put(serialized, off, len);
+                    off += len;
+
+                    finished = off >= serialized.length;
+
+                    if (!finished)
+                        req.serializedOffset(off);
+                }
+                else if (messageFactory() == null) {
                     assert msg instanceof ClientMessage;  // TODO: Will refactor in IGNITE-26554.
 
                     finished = ((ClientMessage)msg).writeTo(buf);
@@ -1815,7 +1899,21 @@ private boolean writeToBuffer(GridSelectorNioSessionImpl ses, ByteBuffer buf, Se
 
                 int startPos = buf.position();
 
-                if (msgFactory == null) {
+                byte[] serialized = req.serializedMessage();
+
+                if (serialized != null) {
+                    int off = req.serializedOffset();
+                    int len = Math.min(serialized.length - off, buf.remaining());
+
+                    buf.put(serialized, off, len);
+                    off += len;
+
+                    finished = off >= serialized.length;
+
+                    if (!finished)
+                        req.serializedOffset(off);
+                }
+                else if (msgFactory == null) {
                     assert msg instanceof ClientMessage;  // TODO: Will refactor in IGNITE-26554.
 
                     finished = ((ClientMessage)msg).writeTo(buf);
@@ -3378,6 +3476,12 @@ static final class WriteRequestImpl implements SessionWriteRequest, SessionChang
         /** Span for tracing. */
         private Span span;
 
+        /** Pre-serialized message bytes. */
+        private final byte[] serializedMsg;
+
+        /** Current offset in pre-serialized bytes for partial writes. */
+        private int serializedOff;
+
         /**
          * @param ses Session.
          * @param msg Message.
@@ -3388,11 +3492,27 @@ static final class WriteRequestImpl implements SessionWriteRequest, SessionChang
             Object msg,
             boolean skipRecovery,
             IgniteInClosure<IgniteException> ackC) {
+            this(ses, msg, skipRecovery, ackC, null);
+        }
+
+        /**
+         * @param ses Session.
+         * @param msg Message.
+         * @param skipRecovery Skip recovery flag.
+         * @param ackC Closure invoked when message ACK is received.
+         * @param serializedMsg Pre-serialized message bytes.
+         */
+        WriteRequestImpl(GridNioSession ses,
+            Object msg,
+            boolean skipRecovery,
+            IgniteInClosure<IgniteException> ackC,
+            byte[] serializedMsg) {
             this.ses = ses;
             this.msg = msg;
             this.skipRecovery = skipRecovery;
             this.ackC = ackC;
             this.span = MTC.span();
+            this.serializedMsg = serializedMsg;
         }
 
         /** {@inheritDoc} */
@@ -3450,6 +3570,21 @@ static final class WriteRequestImpl implements SessionWriteRequest, SessionChang
             return span;
         }
 
+        /** {@inheritDoc} */
+        @Override public byte[] serializedMessage() {
+            return serializedMsg;
+        }
+
+        /** {@inheritDoc} */
+        @Override public int serializedOffset() {
+            return serializedOff;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void serializedOffset(int off) {
+            serializedOff = off;
+        }
+
         /** {@inheritDoc} */
         @Override public String toString() {
             return S.toString(WriteRequestImpl.class, this);
@@ -3496,6 +3631,12 @@ private static class NioOperationFuture<R> extends GridFutureAdapter<R> implemen
         /** */
         private IgniteInClosure<IgniteException> ackC;
 
+        /** Pre-serialized message bytes. */
+        private byte[] serializedMsg;
+
+        /** Current offset in pre-serialized bytes for partial writes. */
+        private int serializedOff;
+
         /**
          * @param sockCh Socket channel.
          * @param accepted {@code True} if socket has been accepted.
@@ -3664,6 +3805,26 @@ boolean accepted() {
             return skipRecovery;
         }
 
+        /** {@inheritDoc} */
+        @Override public byte[] serializedMessage() {
+            return serializedMsg;
+        }
+
+        /** {@inheritDoc} */
+        @Override public int serializedOffset() {
+            return serializedOff;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void serializedOffset(int off) {
+            serializedOff = off;
+        }
+
+        /** Sets pre-serialized message bytes. */
+        void serializedMessage(byte[] serializedMsg) {
+            this.serializedMsg = serializedMsg;
+        }
+
         /** {@inheritDoc} */
         @Override public String toString() {
             return S.toString(NioOperationFuture.class, this);

modules/core/src/main/java/org/apache/ignite/internal/util/nio/SessionWriteRequest.java

@@ -78,4 +78,19 @@ public interface SessionWriteRequest {
      * @return Span.
      */
     Span span();
+
+    /** @return Pre-serialized message bytes, or {@code null} if not pre-serialized. */
+    default byte[] serializedMessage() {
+        return null;
+    }
+
+    /** @return Offset in serialized bytes from which to continue writing. */
+    default int serializedOffset() {
+        return 0;
+    }
+
+    /** @param off New offset after partial write. */
+    default void serializedOffset(int off) {
+        // No-op.
+    }
 }