diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java index 5974e61780743..d722f9ff51381 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java @@ -99,7 +99,6 @@ import org.apache.ignite.internal.util.typedef.C1; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.P1; -import org.apache.ignite.internal.util.typedef.T2; import org.apache.ignite.internal.util.typedef.X; import org.apache.ignite.internal.util.typedef.internal.LT; import org.apache.ignite.internal.util.typedef.internal.S; @@ -122,8 +121,10 @@ import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage; import org.apache.ignite.spi.discovery.DiscoverySpiListener; import org.apache.ignite.spi.discovery.IgniteDiscoveryThread; +import org.apache.ignite.spi.discovery.tcp.internal.ClientMessageHolder; import org.apache.ignite.spi.discovery.tcp.internal.DiscoveryDataPacket; import org.apache.ignite.spi.discovery.tcp.internal.FutureTask; +import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryMessageSerializer; import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode; import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNodesRing; import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoverySpiState; @@ -2859,7 +2860,7 @@ protected class RingMessageWorker extends MessageWorker> { + private class ClientMessageWorker extends MessageWorker { /** Node ID. */ private final UUID clientNodeId; - // The code responsible for sending and receiving messages to and from client nodes represents a special case in ServerImpl, - // as it is split into two separate components. - // One part, ClientMessageWorker, handles only message sending to clients and does not process responses. - // The other part, which reads messages from clients, is implemented in SocketReader. - // Due to this separation, we don't require a full TcpDiscoveryIoSession here - // and can instead extract just the message-writing functionality. - // At the same time, we aim to keep both reading and writing logic encapsulated within TcpDiscoveryIoSession. - // As a result, we need to copy some code from TcpDiscoveryIoSession into the new class, TcpDiscoveryMessageSerializer. - /** */ - private final TcpDiscoveryMessageSerializer clientMsgSer; - /** Session shared with the socket reader serving the same client connection. */ private final TcpDiscoveryIoSession ses; @@ -7518,8 +7510,6 @@ private ClientMessageWorker(TcpDiscoveryIoSession ses, UUID clientNodeId, Ignite this.ses = ses; this.clientNodeId = clientNodeId; - clientMsgSer = new TcpDiscoveryMessageSerializer(ctx); - lastMetricsUpdateMsgTimeNanos = System.nanoTime(); } @@ -7546,24 +7536,19 @@ void metrics(ClusterMetrics metrics) { this.metrics = metrics; } - /** - * @param msg Message. - */ + /** @param msg Discovery Message. */ void addMessage(TcpDiscoveryAbstractMessage msg) { - addMessage(msg, null); + addMessage(new ClientMessageHolder(msg)); } - /** - * @param msg Message. - * @param msgBytes Optional message bytes. - */ - void addMessage(TcpDiscoveryAbstractMessage msg, @Nullable byte[] msgBytes) { - T2 t = new T2<>(msg, msgBytes); + /** @param msgHolder Holder of a Discovery Message to send to the client. */ + void addMessage(ClientMessageHolder msgHolder) { + TcpDiscoveryAbstractMessage msg = msgHolder.message(); if (msg.highPriority()) - queue.addFirst(t); + queue.addFirst(msgHolder); else - queue.add(t); + queue.add(msgHolder); DebugLogger log = messageLogger(msg); @@ -7572,10 +7557,10 @@ void addMessage(TcpDiscoveryAbstractMessage msg, @Nullable byte[] msgBytes) { } /** {@inheritDoc} */ - @Override protected void processMessage(T2 msgT) { + @Override protected void processMessage(ClientMessageHolder msgHolder) { boolean success = false; - TcpDiscoveryAbstractMessage msg = msgT.get1(); + TcpDiscoveryAbstractMessage msg = msgHolder.message(); try { assert msg.verified() : msg; @@ -7601,8 +7586,9 @@ else if (msgLog.isDebugEnabled()) { + getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']'); } - writeToSocket(msgT, spi.failureDetectionTimeoutEnabled() ? spi.clientFailureDetectionTimeout() : - spi.getSocketTimeout()); + long timeout = spi.failureDetectionTimeoutEnabled() ? spi.clientFailureDetectionTimeout() : spi.getSocketTimeout(); + + writeMessage(msgHolder, timeout); } } else { @@ -7613,7 +7599,7 @@ else if (msgLog.isDebugEnabled()) { assert topologyInitialized(msg) : msg; - writeToSocket(msgT, spi.getEffectiveSocketTimeout(false)); + writeMessage(msgHolder, spi.getEffectiveSocketTimeout(false)); } boolean clientFailed = msg instanceof TcpDiscoveryNodeFailedMessage && @@ -7643,14 +7629,16 @@ else if (msgLog.isDebugEnabled()) { } /** - * @param msgT Message tuple. + * @param msgHolder Message holder. * @param timeout Timeout. */ - private void writeToSocket(T2 msgT, long timeout) - throws IgniteCheckedException, IOException { - byte[] msgBytes = msgT.get2() == null ? clientMsgSer.serializeMessage(msgT.get1()) : msgT.get2(); + private void writeMessage(ClientMessageHolder msgHolder, long timeout) throws IgniteCheckedException, IOException { + byte[] msgBytes = msgHolder.messageBytes(); - spi.write(ses, msgBytes, timeout); + if (msgBytes != null) + spi.write(ses, msgBytes, timeout); + else + spi.writeMessage(ses, msgHolder.message(), timeout); } /** diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java index 0251712fc95b7..5fe8ec2bab3f7 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java @@ -29,6 +29,7 @@ import java.net.SocketException; import java.nio.ByteBuffer; import java.security.cert.Certificate; +import java.util.concurrent.locks.ReentrantLock; import javax.net.ssl.SSLPeerUnverifiedException; import javax.net.ssl.SSLSocket; import org.apache.ignite.IgniteCheckedException; @@ -36,7 +37,6 @@ import org.apache.ignite.IgniteLogger; import org.apache.ignite.internal.GridKernalContext; import org.apache.ignite.internal.direct.DirectMessageReader; -import org.apache.ignite.internal.direct.DirectMessageWriter; import org.apache.ignite.internal.managers.communication.DiscoveryMarshalling; import org.apache.ignite.internal.managers.communication.UnknownMessageException; import org.apache.ignite.internal.util.CommonUtils; @@ -47,6 +47,7 @@ import org.apache.ignite.plugin.extensions.communication.Message; import org.apache.ignite.plugin.extensions.communication.MessageFactory; import org.apache.ignite.plugin.extensions.communication.MessageSerializer; +import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryMessageSerializer; import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -66,8 +67,8 @@ public class TcpDiscoveryIoSession implements AutoCloseable { /** Default size of buffer used for buffering socket in/out. */ private static final int DFLT_SOCK_BUFFER_SIZE = 8192; - /** Size for an intermediate buffer for serializing discovery messages. */ - private static final int MSG_BUFFER_SIZE = 100; + /** Size of the intermediate buffer a message is deserialized through. */ + private static final int READ_BUFFER_SIZE = 100; /** */ private final GridKernalContext ctx; @@ -82,11 +83,14 @@ public class TcpDiscoveryIoSession implements AutoCloseable { private final Socket sock; /** */ - private final DirectMessageWriter msgWriter; + private final TcpDiscoveryMessageSerializer msgSer; /** */ private final DirectMessageReader msgReader; + /** */ + private final ByteBuffer readBuf; + /** Buffered socket output stream. */ private final OutputStream out; @@ -94,10 +98,7 @@ public class TcpDiscoveryIoSession implements AutoCloseable { private final CompositeInputStream in; /** */ - private final ByteBuffer readBuf; - - /** */ - private final ByteBuffer writeBuf; + private final ReentrantLock sesWriteLock = new ReentrantLock(); /** * Creates a new discovery I/O session bound to the given socket. @@ -112,12 +113,11 @@ public class TcpDiscoveryIoSession implements AutoCloseable { this.msgFactory = ctx.messageFactory(); this.log = ctx.log(getClass()); - readBuf = ByteBuffer.allocate(MSG_BUFFER_SIZE); - writeBuf = ByteBuffer.allocate(MSG_BUFFER_SIZE); - - msgWriter = new DirectMessageWriter(msgFactory); + readBuf = ByteBuffer.allocate(READ_BUFFER_SIZE); msgReader = new DirectMessageReader(msgFactory, null); + msgSer = new TcpDiscoveryMessageSerializer(ctx); + try { int sendBufSize = sock.getSendBufferSize() > 0 ? sock.getSendBufferSize() : DFLT_SOCK_BUFFER_SIZE; int rcvBufSize = sock.getReceiveBufferSize() > 0 ? sock.getReceiveBufferSize() : DFLT_SOCK_BUFFER_SIZE; @@ -137,8 +137,10 @@ public class TcpDiscoveryIoSession implements AutoCloseable { * @throws IgniteCheckedException If serialization fails. */ void writeMessage(TcpDiscoveryAbstractMessage msg) throws IgniteCheckedException, IOException { + sesWriteLock.lock(); + try { - serializeMessage((Message)msg, out); + msgSer.writeTo(msg, out); out.flush(); } @@ -153,6 +155,9 @@ void writeMessage(TcpDiscoveryAbstractMessage msg) throws IgniteCheckedException throw new IgniteCheckedException(e); } + finally { + sesWriteLock.unlock(); + } } /** @@ -262,32 +267,6 @@ public Socket socket() { return sock; } - /** - * Serializes a discovery message into given output stream. - * - * @param m Discovery message to serialize. - * @param out Output stream to write serialized message. - * @throws IOException If serialization fails. - */ - void serializeMessage(Message m, OutputStream out) throws IOException, IgniteCheckedException { - DiscoveryMarshalling.marshal(m, ctx, null); - - msgWriter.reset(); - msgWriter.setBuffer(writeBuf); - - boolean finished; - - do { - // Should be cleared before first operation. - writeBuf.clear(); - - finished = MessageSerialization.writeTo(msgFactory, m, msgWriter); - - out.write(writeBuf.array(), 0, writeBuf.position()); - } - while (!finished); - } - /** * Writes raw data to the underlying socket output stream. * @@ -295,9 +274,16 @@ void serializeMessage(Message m, OutputStream out) throws IOException, IgniteChe * @throws IOException If failed. */ void write(byte[] data) throws IOException { - out.write(data); + sesWriteLock.lock(); + + try { + out.write(data); - out.flush(); + out.flush(); + } + finally { + sesWriteLock.unlock(); + } } /** @@ -307,9 +293,16 @@ void write(byte[] data) throws IOException { * @throws IOException If failed. */ void write(int b) throws IOException { - out.write(b); + sesWriteLock.lock(); + + try { + out.write(b); - out.flush(); + out.flush(); + } + finally { + sesWriteLock.unlock(); + } } /** diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryMessageSerializer.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryMessageSerializer.java deleted file mode 100644 index ec7cdc569f0c7..0000000000000 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryMessageSerializer.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.ignite.spi.discovery.tcp; - -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.net.Socket; -import org.apache.ignite.IgniteCheckedException; -import org.apache.ignite.internal.GridKernalContext; -import org.apache.ignite.plugin.extensions.communication.Message; -import org.apache.ignite.plugin.extensions.communication.MessageSerializer; -import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage; - -/** - * Class is responsible for serializing discovery messages using RU-ready {@link MessageSerializer} mechanism. - *

- * It is used in a special case: when server wants to send discovery messages to clients, it may not have a {@link TcpDiscoveryIoSession} - * to serialize the messages. - * This class enables server to serialize discovery messages anyway, duplicating serialization code from {@link TcpDiscoveryIoSession}. - */ -class TcpDiscoveryMessageSerializer extends TcpDiscoveryIoSession { - /** - * @param ctx Kernal context. - */ - public TcpDiscoveryMessageSerializer(GridKernalContext ctx) { - super(ctx, new Socket() { - @Override public OutputStream getOutputStream() throws IOException { - return null; - } - - @Override public InputStream getInputStream() throws IOException { - return null; - } - }); - } - - /** - * Serializes a discovery message into a byte array. - * - * @param msg Discovery message to serialize. - * @return Serialized byte array containing the message data. - * @throws IgniteCheckedException If serialization fails. - * @throws IOException If serialization fails. - */ - byte[] serializeMessage(TcpDiscoveryAbstractMessage msg) throws IgniteCheckedException, IOException { - try (ByteArrayOutputStream out = new ByteArrayOutputStream()) { - serializeMessage((Message)msg, out); - - return out.toByteArray(); - } - } -} diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/ClientMessageHolder.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/ClientMessageHolder.java new file mode 100644 index 0000000000000..8e2f50152b57f --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/ClientMessageHolder.java @@ -0,0 +1,59 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.spi.discovery.tcp.internal; + +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage; +import org.jetbrains.annotations.Nullable; + +/** */ +public class ClientMessageHolder { + /** */ + private final TcpDiscoveryAbstractMessage msg; + + /** */ + private byte[] msgBytes; + + /** */ + public ClientMessageHolder(TcpDiscoveryAbstractMessage msg) { + assert msg != null; + + this.msg = msg; + } + + /** */ + public TcpDiscoveryAbstractMessage message() { + return msg; + } + + /** */ + public synchronized byte @Nullable [] messageBytes() { + return msgBytes; + } + + /** */ + public synchronized void serialize(TcpDiscoveryMessageSerializer ser) throws IgniteCheckedException { + if (msgBytes == null) + msgBytes = ser.serialize(msg); + } + + /** {@inheritDoc} */ + @Override public String toString() { + return msg.toString(); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryMessageSerializer.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryMessageSerializer.java new file mode 100644 index 0000000000000..45aa1d9b76743 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryMessageSerializer.java @@ -0,0 +1,96 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.spi.discovery.tcp.internal; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.direct.DirectMessageWriter; +import org.apache.ignite.internal.managers.communication.DiscoveryMarshalling; +import org.apache.ignite.internal.util.io.GridByteArrayOutputStream; +import org.apache.ignite.internal.util.nio.MessageSerialization; +import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage; + +/** */ +public class TcpDiscoveryMessageSerializer { + /** Size of the intermediate buffer a message is serialized through. */ + private static final int BUFFER_SIZE = 100; + + /** */ + private final GridKernalContext ctx; + + /** */ + private final DirectMessageWriter writer; + + /** */ + private final ByteBuffer buf = ByteBuffer.allocate(BUFFER_SIZE); + + /** @param ctx Kernal context. */ + public TcpDiscoveryMessageSerializer(GridKernalContext ctx) { + this.ctx = ctx; + + writer = new DirectMessageWriter(ctx.messageFactory()); + } + + /** + * Serializes a discovery message into given output stream. + * + * @param msg Discovery message to serialize. + * @param out Output stream to write serialized message. + * @throws IgniteCheckedException If serialization fails. + * @throws IOException If serialization fails. + */ + public void writeTo(TcpDiscoveryAbstractMessage msg, OutputStream out) throws IgniteCheckedException, IOException { + DiscoveryMarshalling.marshal(msg, ctx, null); + + writer.reset(); + writer.setBuffer(buf); + + boolean finished; + + do { + // Should be cleared before first operation. + buf.clear(); + + finished = MessageSerialization.writeTo(ctx.messageFactory(), msg, writer); + + out.write(buf.array(), 0, buf.position()); + } + while (!finished); + } + + /** + * Serializes a discovery message into a byte array. + * + * @param msg Discovery message to serialize. + * @return Serialized byte array containing the message data. + * @throws IgniteCheckedException If serialization fails. + */ + public byte[] serialize(TcpDiscoveryAbstractMessage msg) throws IgniteCheckedException { + try (GridByteArrayOutputStream out = new GridByteArrayOutputStream()) { + writeTo(msg, out); + + return out.toByteArray(); + } + catch (IOException e) { + throw new IgniteCheckedException("Failed to serialize a discovery message: " + msg, e); + } + } +}