Skip to content

IGNITE-29005 Replaced raw socket usage with session for Discovery network operations #13509

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
petrov-mg wants to merge 1 commit into
base: master
Choose a base branch
from
Open

IGNITE-29005 Replaced raw socket usage with session for Discovery network operations #13509

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 39 additions & 56 deletions modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import java.io.InterruptedIOException;
import java.io.StreamCorruptedException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.ArrayDeque;
import java.util.ArrayList;
Expand Down Expand Up @@ -626,7 +625,7 @@ private T2<Boolean, T2<TcpDiscoveryIoSession, Integer>> sendJoinRequests(

assert joinRes.get1() != null && joinRes.get2() != null : joinRes;

Socket sock = joinRes.get1().socket();
TcpDiscoveryIoSession ses = joinRes.get1();

if (log.isDebugEnabled())
log.debug("Received response to join request [addr=" + addr + ", res=" + joinRes.get2() + ']');
Expand All @@ -637,15 +636,15 @@ private T2<Boolean, T2<TcpDiscoveryIoSession, Integer>> sendJoinRequests(

case RES_CONTINUE_JOIN:
case RES_WAIT:
U.closeQuiet(sock);
U.closeQuiet(ses);

return new T2<>(true, null);

default:
if (log.isDebugEnabled())
log.debug("Received unexpected response to join request: " + joinRes.get2());

U.closeQuiet(sock);
U.closeQuiet(ses);
}
}

Expand Down Expand Up @@ -696,18 +695,16 @@ private static void sleepEx(long millis, Runnable before, Runnable after) throws
DiscoveryDataPacket discoveryData = null;

while (true) {
boolean openSock = false;
boolean openSes = false;

Socket sock = null;
TcpDiscoveryIoSession ses = null;

try {
long tsNanos = System.nanoTime();

sock = spi.openSocket(addr, timeoutHelper);
ses = spi.openSession(addr, timeoutHelper);

openSock = true;

TcpDiscoveryIoSession ses = createSession(sock);
openSes = true;

TcpDiscoveryHandshakeRequest req = new TcpDiscoveryHandshakeRequest(locNodeId, locNode.features());

Expand All @@ -724,9 +721,9 @@ private static void sleepEx(long millis, Runnable before, Runnable after) throws
Collection<InetSocketAddress> redirectAddrs = res.redirectAddresses();

if (redirectAddrs != null) {
U.closeQuiet(sock);
U.closeQuiet(ses);

openSock = false;
openSes = false;

if (log.isInfoEnabled())
log.info("Reconnecting to the addresses of a proper DC [addrs=" + redirectAddrs + ']');
Expand Down Expand Up @@ -783,10 +780,10 @@ private static void sleepEx(long millis, Runnable before, Runnable after) throws
log.debug("Message has been sent to address [msg=" + msg + ", addr=" + addr +
", rmtNodeId=" + rmtNodeId + ']');

return new T2<>(ses, spi.readReceipt(sock, timeoutHelper.nextTimeoutChunk(ackTimeout0)));
return new T2<>(ses, spi.readReceipt(ses, timeoutHelper.nextTimeoutChunk(ackTimeout0)));
}
catch (IOException | IgniteCheckedException e) {
U.closeQuiet(sock);
U.closeQuiet(ses);

if (log.isDebugEnabled())
log.error("Exception on joining: " + e.getMessage(), e);
Expand Down Expand Up @@ -836,7 +833,7 @@ private static void sleepEx(long millis, Runnable before, Runnable after) throws
if (!spi.failureDetectionTimeoutEnabled() && ++reconCnt == spi.getReconnectCount())
break;

if (!openSock) {
if (!openSes) {
// Reconnect for the second time, if connection is not established.
if (connectAttempts < 2) {
connectAttempts++;
Expand Down Expand Up @@ -971,7 +968,7 @@ private NavigableSet<ClusterNode> allVisibleNodes() {
TcpDiscoveryIoSession ses = msgWorker.currSes;

if (ses != null)
U.closeQuiet(ses.socket());
U.closeQuiet(ses);
}

/** {@inheritDoc} */
Expand Down Expand Up @@ -1133,7 +1130,7 @@ private void forceStopRead() throws InterruptedException {

this.stopReadLatch = stopReadLatch = new CountDownLatch(1);

U.closeQuiet(ses.socket());
U.closeQuiet(ses);

this.ses = null;
this.rmtNodeId = null;
Expand Down Expand Up @@ -1170,26 +1167,25 @@ private void forceStopRead() throws InterruptedException {
rmtNodeId = this.rmtNodeId;
}

Socket sock = ses.socket();

U.enhanceThreadName(U.id8(rmtNodeId)
+ ' ' + sock.getInetAddress().getHostAddress()
+ ":" + sock.getPort());
+ ' ' + ses.socket().getInetAddress().getHostAddress()
+ ":" + ses.socket().getPort());

try {
assert sock.getKeepAlive() && sock.getTcpNoDelay() : "Socket wasn't configured properly:" +
" KeepAlive " + sock.getKeepAlive() +
" TcpNoDelay " + sock.getTcpNoDelay();
assert ses.socket().getKeepAlive() && ses.socket().getTcpNoDelay() :
"Socket wasn't configured properly:" +
" KeepAlive " + ses.socket().getKeepAlive() +
" TcpNoDelay " + ses.socket().getTcpNoDelay();

while (!isInterrupted()) {
TcpDiscoveryAbstractMessage msg;

try {
msg = spi.readMessage(ses, sock.getSoTimeout());
msg = spi.readMessage(ses, ses.socket().getSoTimeout());
}
catch (IgniteCheckedException e) {
if (log.isDebugEnabled())
U.error(log, "Failed to read message [sock=" + sock + ", " +
U.error(log, "Failed to read message [ses=" + ses + ", " +
"locNodeId=" + getLocalNodeId() + ", rmtNodeId=" + rmtNodeId + ']', e);

// Exists possibility that exception raised on interruption.
Expand All @@ -1209,7 +1205,7 @@ private void forceStopRead() throws InterruptedException {
"(make sure same versions of all classes are available on all nodes) " +
"[rmtNodeId=" + rmtNodeId + ", err=" + clsNotFoundEx.getMessage() + ']');
else
LT.error(log, e, "Failed to read message [sock=" + sock + ", locNodeId=" +
LT.error(log, e, "Failed to read message [ses=" + ses + ", locNodeId=" +
getLocalNodeId() + ", rmtNodeId=" + rmtNodeId + ']');

continue;
Expand All @@ -1236,10 +1232,10 @@ private void forceStopRead() throws InterruptedException {
msgWorker.addMessage(new SessionClosedMessage(ses));

if (log.isDebugEnabled())
U.error(log, "Connection failed [sock=" + sock + ", locNodeId=" + getLocalNodeId() + ']', e);
U.error(log, "Connection failed [ses=" + ses + ", locNodeId=" + getLocalNodeId() + ']', e);
}
finally {
U.closeQuiet(sock);
U.closeQuiet(ses);

synchronized (mux) {
if (this.ses == ses) {
Expand Down Expand Up @@ -1386,7 +1382,7 @@ void ackReceived(TcpDiscoveryClientAckResponse res) {
}
}

U.closeQuiet(ses.socket());
U.closeQuiet(ses);

this.ses = null;

Expand Down Expand Up @@ -1479,7 +1475,7 @@ void ackReceived(TcpDiscoveryClientAckResponse res) {
else
U.error(log, "Failed to send message: " + msg, e);

U.closeQuiet(ses.socket());
U.closeQuiet(ses);

synchronized (mux) {
if (ses == this.ses)
Expand Down Expand Up @@ -1542,8 +1538,7 @@ public void cancel() {

TcpDiscoveryIoSession ses = this.ses;

if (ses != null)
U.closeQuiet(ses.socket());
U.closeQuiet(ses);
}

/** {@inheritDoc} */
Expand Down Expand Up @@ -1578,26 +1573,19 @@ public void cancel() {

this.ses = ses;

Socket sock = ses.socket();

if (isInterrupted())
throw new InterruptedException();

int oldTimeout = 0;

try {
oldTimeout = sock.getSoTimeout();

sock.setSoTimeout((int)spi.netTimeout);

assert sock.getKeepAlive() && sock.getTcpNoDelay() : "Socket wasn't configured properly:" +
" KeepAlive " + sock.getKeepAlive() +
" TcpNoDelay " + sock.getTcpNoDelay();
assert ses.socket().getKeepAlive() && ses.socket().getTcpNoDelay() :
"Socket wasn't configured properly:" +
" KeepAlive " + ses.socket().getKeepAlive() +
" TcpNoDelay " + ses.socket().getTcpNoDelay();

List<TcpDiscoveryAbstractMessage> msgs = null;

while (!isInterrupted()) {
TcpDiscoveryAbstractMessage msg = spi.readMessage(ses, sock.getSoTimeout());
TcpDiscoveryAbstractMessage msg = spi.readMessage(ses, spi.netTimeout);

if (msg instanceof TcpDiscoveryClientReconnectMessage) {
TcpDiscoveryClientReconnectMessage res = (TcpDiscoveryClientReconnectMessage)msg;
Expand Down Expand Up @@ -1632,7 +1620,7 @@ else if (spi.ensured(msg)) {
}
}
catch (IOException | IgniteCheckedException e) {
U.closeQuiet(sock);
U.closeQuiet(ses);

if (log.isDebugEnabled())
log.error("Reconnect error [join=" + join + ", timeout=" + timeout + ']', e);
Expand All @@ -1650,10 +1638,6 @@ else if (spi.ensured(msg)) {
else
U.warn(log, "Failed to reconnect to cluster (will retry): " + e);
}
finally {
if (success)
sock.setSoTimeout(oldTimeout);
}
}
}
catch (IOException | IgniteCheckedException e) {
Expand All @@ -1667,8 +1651,7 @@ else if (spi.ensured(msg)) {
if (!success) {
TcpDiscoveryIoSession ses = this.ses;

if (ses != null)
U.closeQuiet(ses.socket());
U.closeQuiet(ses);

if (join)
joinError(new IgniteSpiException("Failed to connect to cluster, connection failed and failed " +
Expand Down Expand Up @@ -1814,9 +1797,9 @@ else if (msg instanceof TcpDiscoveryNodeFailedMessage &&
}
else if (msg instanceof SessionClosedMessage sesClosedMsg) {
if (sesClosedMsg.ses == currSes) {
Socket sock = currSes.socket();

InetSocketAddress prevAddr = new InetSocketAddress(sock.getInetAddress(), sock.getPort());
InetSocketAddress prevAddr = new InetSocketAddress(
currSes.socket().getInetAddress(),
currSes.socket().getPort());

currSes = null;

Expand Down Expand Up @@ -1981,7 +1964,7 @@ else if (discoMsg instanceof TcpDiscoveryCheckFailedMessage)
TcpDiscoveryIoSession ses = this.currSes;

if (ses != null)
U.closeQuiet(ses.socket());
U.closeQuiet(ses);

if (joinLatch.getCount() > 0)
joinError(new IgniteSpiException("Some error in join process.")); // This should not occur.
Expand Down
Loading
Loading