This is an automated email from the ASF dual-hosted git repository.
petrov-mg 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 2c14ece08e4 IGNITE-29004 Refactored ClientImpl to use single instance
of TcpDiscoveryIoSession per connection (#13507)
2c14ece08e4 is described below
commit 2c14ece08e4e65db248ad09bd2d02970e5102946
Author: Mikhail Petrov <[email protected]>
AuthorDate: Tue Sep 1 00:23:33 2026 +0300
IGNITE-29004 Refactored ClientImpl to use single instance of
TcpDiscoveryIoSession per connection (#13507)
---
.../ignite/spi/discovery/tcp/ClientImpl.java | 223 ++++++++-------------
.../spi/discovery/tcp/TcpDiscoveryIoSession.java | 35 ++--
2 files changed, 104 insertions(+), 154 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
index 53b1dbfc935..3cfacdd43b9 100644
---
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
@@ -535,7 +535,7 @@ class ClientImpl extends TcpDiscoveryImpl {
* @throws IgniteSpiException If failed.
* @see TcpDiscoverySpi#joinTimeout
*/
- @Nullable private SocketStream joinTopology(
+ @Nullable private TcpDiscoveryIoSession joinTopology(
InetSocketAddress prevAddr,
long timeout,
@Nullable Runnable beforeEachSleep,
@@ -581,12 +581,12 @@ class ClientImpl extends TcpDiscoveryImpl {
Collection<InetSocketAddress> addrs0 = new ArrayList<>(addrs);
- T2<Boolean, T2<SocketStream, Integer>> waitAndRes =
sendJoinRequests(prevAddr != null, addrs);
+ T2<Boolean, T2<TcpDiscoveryIoSession, Integer>> waitAndRes =
sendJoinRequests(prevAddr != null, addrs);
addrs.clear();
boolean wait = waitAndRes.get1();
- T2<SocketStream, Integer> res = waitAndRes.get2();
+ T2<TcpDiscoveryIoSession, Integer> res = waitAndRes.get2();
if (res != null)
return res.get1();
@@ -611,7 +611,7 @@ class ClientImpl extends TcpDiscoveryImpl {
}
/** */
- private T2<Boolean, T2<SocketStream, Integer>> sendJoinRequests(
+ private T2<Boolean, T2<TcpDiscoveryIoSession, Integer>> sendJoinRequests(
boolean recon,
Collection<InetSocketAddress> addrs
) throws InterruptedException {
@@ -619,21 +619,21 @@ class ClientImpl extends TcpDiscoveryImpl {
if (Thread.currentThread().isInterrupted())
throw new InterruptedException();
- T2<SocketStream, Integer> sockAndRes = sendJoinRequest(recon,
addr);
+ T2<TcpDiscoveryIoSession, Integer> joinRes =
sendJoinRequest(recon, addr);
- if (sockAndRes == null)
+ if (joinRes == null)
continue;
- assert sockAndRes.get1() != null && sockAndRes.get2() != null :
sockAndRes;
+ assert joinRes.get1() != null && joinRes.get2() != null : joinRes;
- Socket sock = sockAndRes.get1().socket();
+ Socket sock = joinRes.get1().socket();
if (log.isDebugEnabled())
- log.debug("Received response to join request [addr=" + addr +
", res=" + sockAndRes.get2() + ']');
+ log.debug("Received response to join request [addr=" + addr +
", res=" + joinRes.get2() + ']');
- switch (sockAndRes.get2()) {
+ switch (joinRes.get2()) {
case RES_OK:
- return new T2<>(false, sockAndRes);
+ return new T2<>(false, joinRes);
case RES_CONTINUE_JOIN:
case RES_WAIT:
@@ -643,7 +643,7 @@ class ClientImpl extends TcpDiscoveryImpl {
default:
if (log.isDebugEnabled())
- log.debug("Received unexpected response to join
request: " + sockAndRes.get2());
+ log.debug("Received unexpected response to join
request: " + joinRes.get2());
U.closeQuiet(sock);
}
@@ -671,7 +671,7 @@ class ClientImpl extends TcpDiscoveryImpl {
* @param addr Address.
* @return Socket, connect response and client acknowledge support flag.
*/
- @Nullable private T2<SocketStream, Integer> sendJoinRequest(boolean recon,
+ @Nullable private T2<TcpDiscoveryIoSession, Integer>
sendJoinRequest(boolean recon,
InetSocketAddress addr) throws InterruptedException {
assert addr != null;
@@ -731,7 +731,7 @@ class ClientImpl extends TcpDiscoveryImpl {
if (log.isInfoEnabled())
log.info("Reconnecting to the addresses of a proper DC
[addrs=" + redirectAddrs + ']');
- T2<Boolean, T2<SocketStream, Integer>> redirectedRes =
sendJoinRequests(recon, redirectAddrs);
+ T2<Boolean, T2<TcpDiscoveryIoSession, Integer>>
redirectedRes = sendJoinRequests(recon, redirectAddrs);
return redirectedRes.get2();
}
@@ -783,7 +783,7 @@ class ClientImpl extends TcpDiscoveryImpl {
log.debug("Message has been sent to address [msg=" + msg +
", addr=" + addr +
", rmtNodeId=" + rmtNodeId + ']');
- return new T2<>(new SocketStream(sock, ses),
spi.readReceipt(sock, timeoutHelper.nextTimeoutChunk(ackTimeout0)));
+ return new T2<>(ses, spi.readReceipt(sock,
timeoutHelper.nextTimeoutChunk(ackTimeout0)));
}
catch (IOException | IgniteCheckedException e) {
U.closeQuiet(sock);
@@ -968,10 +968,10 @@ class ClientImpl extends TcpDiscoveryImpl {
/** {@inheritDoc} */
@Override public void brakeConnection() {
- SocketStream sockStream = msgWorker.currSock;
+ TcpDiscoveryIoSession ses = msgWorker.currSes;
- if (sockStream != null)
- U.closeQuiet(sockStream.socket());
+ if (ses != null)
+ U.closeQuiet(ses.socket());
}
/** {@inheritDoc} */
@@ -1093,7 +1093,7 @@ class ClientImpl extends TcpDiscoveryImpl {
private final Object mux = new Object();
/** */
- private SocketStream sockStream;
+ private TcpDiscoveryIoSession ses;
/** */
private UUID rmtNodeId;
@@ -1108,12 +1108,12 @@ class ClientImpl extends TcpDiscoveryImpl {
}
/**
- * @param sockStream Socket.
+ * @param ses IO session.
* @param rmtNodeId Rmt node id.
*/
- void setSocket(SocketStream sockStream, UUID rmtNodeId) {
+ void onRemoteSessionReady(TcpDiscoveryIoSession ses, UUID rmtNodeId) {
synchronized (mux) {
- this.sockStream = sockStream;
+ this.ses = ses;
this.rmtNodeId = rmtNodeId;
@@ -1128,16 +1128,14 @@ class ClientImpl extends TcpDiscoveryImpl {
CountDownLatch stopReadLatch;
synchronized (mux) {
- SocketStream stream = sockStream;
-
- if (stream == null)
+ if (ses == null)
return;
this.stopReadLatch = stopReadLatch = new CountDownLatch(1);
- U.closeQuiet(stream.socket());
+ U.closeQuiet(ses.socket());
- this.sockStream = null;
+ this.ses = null;
this.rmtNodeId = null;
mux.notifyAll();
@@ -1149,7 +1147,7 @@ class ClientImpl extends TcpDiscoveryImpl {
/** {@inheritDoc} */
@Override protected void body() throws InterruptedException {
while (!isInterrupted()) {
- SocketStream sockStream;
+ TcpDiscoveryIoSession ses;
UUID rmtNodeId;
// Disconnected from router node.
@@ -1162,25 +1160,23 @@ class ClientImpl extends TcpDiscoveryImpl {
stopReadLatch = null;
}
- if (this.sockStream == null) {
+ if (this.ses == null) {
mux.wait();
continue;
}
- sockStream = this.sockStream;
+ ses = this.ses;
rmtNodeId = this.rmtNodeId;
}
- Socket sock = sockStream.socket();
+ Socket sock = ses.socket();
U.enhanceThreadName(U.id8(rmtNodeId)
- + ' ' + sockStream.sock.getInetAddress().getHostAddress()
- + ":" + sockStream.sock.getPort());
+ + ' ' + sock.getInetAddress().getHostAddress()
+ + ":" + sock.getPort());
try {
- TcpDiscoveryIoSession ses = sockStream.session();
-
assert sock.getKeepAlive() && sock.getTcpNoDelay() :
"Socket wasn't configured properly:" +
" KeepAlive " + sock.getKeepAlive() +
" TcpNoDelay " + sock.getTcpNoDelay();
@@ -1237,7 +1233,7 @@ class ClientImpl extends TcpDiscoveryImpl {
}
}
catch (IOException e) {
- msgWorker.addMessage(new SocketClosedMessage(sockStream));
+ msgWorker.addMessage(new SessionClosedMessage(ses));
if (log.isDebugEnabled())
U.error(log, "Connection failed [sock=" + sock + ",
locNodeId=" + getLocalNodeId() + ']', e);
@@ -1246,8 +1242,8 @@ class ClientImpl extends TcpDiscoveryImpl {
U.closeQuiet(sock);
synchronized (mux) {
- if (this.sockStream == sockStream) {
- this.sockStream = null;
+ if (this.ses == ses) {
+ this.ses = null;
this.rmtNodeId = null;
}
}
@@ -1263,9 +1259,6 @@ class ClientImpl extends TcpDiscoveryImpl {
/** */
private final Object mux = new Object();
- /** */
- private Socket sock;
-
/** */
private TcpDiscoveryIoSession ses;
@@ -1314,7 +1307,7 @@ class ClientImpl extends TcpDiscoveryImpl {
synchronized (mux) {
// If writer was stopped.
- if (sock == null)
+ if (ses == null)
return;
this.forceLeaveLatch = forceLeaveLatch = new CountDownLatch(1);
@@ -1327,14 +1320,10 @@ class ClientImpl extends TcpDiscoveryImpl {
forceLeaveLatch.await();
}
- /**
- * @param sock Socket.
- */
- private void setSocket(Socket sock) {
+ /** */
+ private void onRemoteSessionReady(TcpDiscoveryIoSession ses) {
synchronized (mux) {
- this.sock = sock;
-
- ses = createSession(sock);
+ this.ses = ses;
unackedMsg = null;
@@ -1347,7 +1336,7 @@ class ClientImpl extends TcpDiscoveryImpl {
*/
public boolean isOnline() {
synchronized (mux) {
- return sock != null;
+ return ses != null;
}
}
@@ -1371,12 +1360,12 @@ class ClientImpl extends TcpDiscoveryImpl {
TcpDiscoveryAbstractMessage msg = null;
while (!Thread.currentThread().isInterrupted()) {
- Socket sock;
+ TcpDiscoveryIoSession ses;
synchronized (mux) {
- sock = this.sock;
+ ses = this.ses;
- if (sock == null) {
+ if (ses == null) {
mux.wait();
continue;
@@ -1397,9 +1386,9 @@ class ClientImpl extends TcpDiscoveryImpl {
}
}
- U.closeQuiet(sock);
+ U.closeQuiet(ses.socket());
- this.sock = null;
+ this.ses = null;
clear();
@@ -1490,11 +1479,11 @@ class ClientImpl extends TcpDiscoveryImpl {
else
U.error(log, "Failed to send message: " + msg, e);
- U.closeQuiet(sock);
+ U.closeQuiet(ses.socket());
synchronized (mux) {
- if (sock == this.sock)
- this.sock = null; // Connection has dead.
+ if (ses == this.ses)
+ this.ses = null; // Connection has died.
clear();
}
@@ -1526,7 +1515,7 @@ class ClientImpl extends TcpDiscoveryImpl {
*/
private class Reconnector extends IgniteSpiThread {
/** */
- private volatile SocketStream sockStream;
+ private volatile TcpDiscoveryIoSession ses;
/** */
private final boolean join;
@@ -1551,10 +1540,10 @@ class ClientImpl extends TcpDiscoveryImpl {
public void cancel() {
interrupt();
- SocketStream sockStream = this.sockStream;
+ TcpDiscoveryIoSession ses = this.ses;
- if (sockStream != null)
- U.closeQuiet(sockStream.socket());
+ if (ses != null)
+ U.closeQuiet(ses.socket());
}
/** {@inheritDoc} */
@@ -1572,9 +1561,9 @@ class ClientImpl extends TcpDiscoveryImpl {
try {
while (true) {
- SocketStream joinRes = joinTopology(prevAddr, timeout,
null, null);
+ TcpDiscoveryIoSession ses = joinTopology(prevAddr,
timeout, null, null);
- if (joinRes == null) {
+ if (ses == null) {
if (join) {
joinError(new IgniteSpiException("Join process
timed out, connection failed and " +
"failed to reconnect (consider increasing
'joinTimeout' configuration property) " +
@@ -1587,10 +1576,9 @@ class ClientImpl extends TcpDiscoveryImpl {
return;
}
- sockStream = joinRes;
+ this.ses = ses;
- Socket sock = sockStream.socket();
- TcpDiscoveryIoSession ses = sockStream.session();
+ Socket sock = ses.socket();
if (isInterrupted())
throw new InterruptedException();
@@ -1677,10 +1665,10 @@ class ClientImpl extends TcpDiscoveryImpl {
}
finally {
if (!success) {
- SocketStream sockStream = this.sockStream;
+ TcpDiscoveryIoSession ses = this.ses;
- if (sockStream != null)
- U.closeQuiet(sockStream.socket());
+ if (ses != null)
+ U.closeQuiet(ses.socket());
if (join)
joinError(new IgniteSpiException("Failed to connect to
cluster, connection failed and failed " +
@@ -1700,7 +1688,7 @@ class ClientImpl extends TcpDiscoveryImpl {
private final BlockingDeque<Object> queue = new
LinkedBlockingDeque<>();
/** */
- private SocketStream currSock;
+ private TcpDiscoveryIoSession currSes;
/** */
private Reconnector reconnector;
@@ -1751,7 +1739,7 @@ class ClientImpl extends TcpDiscoveryImpl {
if (state == STARTING) {
joinError(new IgniteSpiException("Join process
timed out, did not receive response for " +
"join request (consider increasing
'joinTimeout' configuration property) " +
- "[joinTimeout=" + spi.joinTimeout + ",
sock=" + currSock + ']'));
+ "[joinTimeout=" + spi.joinTimeout + ",
ses=" + currSes + ']'));
break;
}
@@ -1774,7 +1762,7 @@ class ClientImpl extends TcpDiscoveryImpl {
assert spi.getSpiContext().isStopping();
- if (connected && currSock != null) {
+ if (connected && currSes != null) {
TcpDiscoveryNodeLeftMessage leftMsg = new
TcpDiscoveryNodeLeftMessage(getLocalNodeId());
leftMsg.client(true);
@@ -1796,7 +1784,7 @@ class ClientImpl extends TcpDiscoveryImpl {
sockWriter.forceLeave();
sockReader.forceStopRead();
- currSock = null;
+ currSes = null;
queue.clear();
@@ -1824,13 +1812,13 @@ class ClientImpl extends TcpDiscoveryImpl {
forceFailMsg = msg0;
}
- else if (msg instanceof SocketClosedMessage) {
- if (((SocketClosedMessage)msg).sock == currSock) {
- Socket sock = currSock.sock;
+ else if (msg instanceof SessionClosedMessage sesClosedMsg)
{
+ if (sesClosedMsg.ses == currSes) {
+ Socket sock = currSes.socket();
InetSocketAddress prevAddr = new
InetSocketAddress(sock.getInetAddress(), sock.getPort());
- currSock = null;
+ currSes = null;
boolean join = joinLatch.getCount() > 0;
@@ -1990,10 +1978,10 @@ class ClientImpl extends TcpDiscoveryImpl {
((IgniteEx)spi.ignite()).context().failure().process(new
FailureContext(CRITICAL_ERROR, t));
}
finally {
- SocketStream currSock = this.currSock;
+ TcpDiscoveryIoSession ses = this.currSes;
- if (currSock != null)
- U.closeQuiet(currSock.socket());
+ if (ses != null)
+ U.closeQuiet(ses.socket());
if (joinLatch.getCount() > 0)
joinError(new IgniteSpiException("Some error in join
process.")); // This should not occur.
@@ -2072,10 +2060,10 @@ class ClientImpl extends TcpDiscoveryImpl {
joinCnt++;
- SocketStream joinRes;
+ TcpDiscoveryIoSession ses;
try {
- joinRes = joinTopology(null, spi.joinTimeout,
+ ses = joinTopology(null, spi.joinTimeout,
new Runnable() {
@Override public void run() {
blockingSectionBegin();
@@ -2093,7 +2081,7 @@ class ClientImpl extends TcpDiscoveryImpl {
return;
}
- if (joinRes == null) {
+ if (ses == null) {
if (join)
joinError(new IgniteSpiException("Join process timed out
(timeout = " + spi.joinTimeout + ")."));
else {
@@ -2105,9 +2093,9 @@ class ClientImpl extends TcpDiscoveryImpl {
return;
}
- currSock = joinRes;
+ currSes = ses;
- sockWriter.setSocket(joinRes.socket());
+ sockWriter.onRemoteSessionReady(ses);
if (spi.joinTimeout > 0) {
final int joinCnt0 = joinCnt;
@@ -2117,7 +2105,7 @@ class ClientImpl extends TcpDiscoveryImpl {
}, spi.joinTimeout, MILLISECONDS);
}
- sockReader.setSocket(joinRes, locNode.clientRouterNodeId());
+ sockReader.onRemoteSessionReady(ses, locNode.clientRouterNodeId());
}
/** */
@@ -2519,11 +2507,12 @@ class ClientImpl extends TcpDiscoveryImpl {
if (getLocalNodeId().equals(msg.creatorNodeId())) {
if (reconnector != null) {
assert msg.success() : msg;
+ assert reconnector.ses != null : msg;
- currSock = reconnector.sockStream;
+ currSes = reconnector.ses;
- sockWriter.setSocket(currSock.socket());
- sockReader.setSocket(currSock,
locNode.clientRouterNodeId());
+ sockWriter.onRemoteSessionReady(currSes);
+ sockReader.onRemoteSessionReady(currSes,
locNode.clientRouterNodeId());
reconnector = null;
@@ -2707,62 +2696,14 @@ class ClientImpl extends TcpDiscoveryImpl {
/**
*
*/
- private static class SocketClosedMessage {
+ private static class SessionClosedMessage {
/** */
- private final SocketStream sock;
-
- /**
- * @param sock Socket.
- */
- private SocketClosedMessage(SocketStream sock) {
- this.sock = sock;
- }
- }
-
- /**
- *
- */
- private static class SocketStream {
- /** */
- private final Socket sock;
-
- /**
- * The only session ever used to read messages from the socket. Shared
by all socket users
- * ({@link Reconnector}, {@link SocketReader}), otherwise messages
buffered by one session's
- * read-ahead would be lost when another session takes the socket over.
- */
private final TcpDiscoveryIoSession ses;
- /**
- * @param sock Socket.
- * @param ses Session bound to the socket.
- */
- SocketStream(Socket sock, TcpDiscoveryIoSession ses) {
- assert sock != null;
- assert ses != null;
-
- this.sock = sock;
+ /** */
+ private SessionClosedMessage(TcpDiscoveryIoSession ses) {
this.ses = ses;
}
-
- /**
- * @return Socket.
- */
- Socket socket() {
- return sock;
- }
-
- /**
- * @return Session bound to the socket.
- */
- TcpDiscoveryIoSession session() {
- return ses;
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return sock.toString();
- }
}
/**
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 3749f3b4234..8c34284e87f 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
@@ -85,8 +85,11 @@ public class TcpDiscoveryIoSession {
/** Buffered socket input stream. */
private final CompositeInputStream in;
- /** Intermediate buffer for serializing discovery messages. */
- private final ByteBuffer msgBuf;
+ /** */
+ private final ByteBuffer readBuf;
+
+ /** */
+ private final ByteBuffer writeBuf;
/**
* Creates a new discovery I/O session bound to the given socket.
@@ -99,7 +102,8 @@ public class TcpDiscoveryIoSession {
this.sock = sock;
this.spi = spi;
- msgBuf = ByteBuffer.allocate(MSG_BUFFER_SIZE);
+ readBuf = ByteBuffer.allocate(MSG_BUFFER_SIZE);
+ writeBuf = ByteBuffer.allocate(MSG_BUFFER_SIZE);
msgWriter = new DirectMessageWriter(spi.messageFactory());
msgReader = new DirectMessageReader(spi.messageFactory(), null);
@@ -168,19 +172,19 @@ public class TcpDiscoveryIoSession {
}
msgReader.reset();
- msgReader.setBuffer(msgBuf);
+ msgReader.setBuffer(readBuf);
boolean finished;
do {
- msgBuf.clear();
+ readBuf.clear();
- int read = in.read(msgBuf.array(), msgBuf.position(),
msgBuf.remaining());
+ int read = in.read(readBuf.array(), readBuf.position(),
readBuf.remaining());
if (read == -1)
throw new EOFException("Connection closed before message
was fully read.");
- msgBuf.limit(read);
+ readBuf.limit(read);
finished = MessageSerialization.readFrom(spi.messageFactory(),
msg, msgReader);
@@ -188,10 +192,10 @@ public class TcpDiscoveryIoSession {
// This behaviour guarantees that we never read a next message
from the buffer right after the end of
// the previous message. But it is not guaranteed with Client
Discovery where messages aren't acknowledged.
// Thus, we have to keep the uprocessed bytes read from the
socket. It won't return them again.
- if (msgBuf.hasRemaining()) {
- byte[] unprocessedReadTail = new byte[msgBuf.remaining()];
+ if (readBuf.hasRemaining()) {
+ byte[] unprocessedReadTail = new byte[readBuf.remaining()];
- msgBuf.get(unprocessedReadTail, 0, msgBuf.remaining());
+ readBuf.get(unprocessedReadTail, 0, readBuf.remaining());
in.attachByteArray(unprocessedReadTail);
}
@@ -249,17 +253,17 @@ public class TcpDiscoveryIoSession {
DiscoveryMarshalling.marshal(m, kctx, null);
msgWriter.reset();
- msgWriter.setBuffer(msgBuf);
+ msgWriter.setBuffer(writeBuf);
boolean finished;
do {
// Should be cleared before first operation.
- msgBuf.clear();
+ writeBuf.clear();
finished = MessageSerialization.writeTo(spi.messageFactory(), m,
msgWriter);
- out.write(msgBuf.array(), 0, msgBuf.position());
+ out.write(writeBuf.array(), 0, writeBuf.position());
}
while (!finished);
}
@@ -284,6 +288,11 @@ public class TcpDiscoveryIoSession {
throw new StreamCorruptedException("invalid stream header: " +
hex);
}
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "TcpDiscoveryIoSession [sock=" + sock + ']';
+ }
+
/**
* Input stream implementation that combines a byte array and a regular
InputStream allowing to read bytes
* from the array first and then proceed with reading from InputStream.
