Copilot commented on code in PR #13791:
URL: https://github.com/apache/ignite/pull/13791#discussion_r4501187336
##########
modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ConnectionClientPool.java:
##########
@@ -355,33 +401,67 @@ public GridCommunicationClient reserveClient(ClusterNode
node, int connIdx) thro
else
fut = oldFut;
- long clientReserveWaitTimeout = registry != null ?
registry.getSystemWorkerBlockedTimeout() / 3
- : cfg.connectionTimeout() / 3;
+ if (ownClient != null)
+ client = ownClient;
+
+ if (client == null) {
+ long perWaitTimeout = registry != null ?
+ registry.getSystemWorkerBlockedTimeout() / 3 :
cfg.connectionTimeout() / 3;
- long currTimeout = System.currentTimeMillis();
+ long currTimeout = System.currentTimeMillis();
- // This cycle will eventually quit when future is
completed by concurrent thread reserving client.
- while (true) {
- try {
- client = fut.get(clientReserveWaitTimeout,
TimeUnit.MILLISECONDS);
+ // This cycle will eventually quit when future is
completed by concurrent thread reserving client.
+ while (true) {
+ // A concurrent thread may have established the
client while this thread was
+ // waiting: return it instead of waiting further
or giving up, even if the
+ // reserve budget has already elapsed — the budget
bounds waiting for a
+ // connection, not the availability of an existing
client.
+ GridCommunicationClient[] curClients0 =
clients.get(nodeId);
- break;
- }
- catch (IgniteFutureTimeoutCheckedException ignored) {
- currTimeout += clientReserveWaitTimeout;
-
- if (log.isDebugEnabled()) {
- log.debug(
- "Still waiting for reestablishing
connection to node " +
- "[nodeId=" + node.id() + ",
waitingTime=" + currTimeout + "ms]"
- );
+ GridCommunicationClient curClient = curClients0 !=
null && connIdx < curClients0.length ?
+ curClients0[connIdx] : null;
+
+ if (curClient != null) {
+ client = curClient;
+
+ break;
Review Comment:
This shortcut can return a client after `stop()` has completed the shared
future exceptionally. The creator can publish through `addNodeClient` after
shutdown, letting a waiter take the client here before the creator rolls back
publication and closes it. Coordinate publication and future completion with
shutdown so a client whose future completed with a stop error is never exposed,
and cover a concurrent waiter in the shutdown regression test.
##########
modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ConnectionClientPool.java:
##########
@@ -355,33 +401,67 @@ public GridCommunicationClient reserveClient(ClusterNode
node, int connIdx) thro
else
fut = oldFut;
- long clientReserveWaitTimeout = registry != null ?
registry.getSystemWorkerBlockedTimeout() / 3
- : cfg.connectionTimeout() / 3;
+ if (ownClient != null)
+ client = ownClient;
+
+ if (client == null) {
+ long perWaitTimeout = registry != null ?
+ registry.getSystemWorkerBlockedTimeout() / 3 :
cfg.connectionTimeout() / 3;
- long currTimeout = System.currentTimeMillis();
+ long currTimeout = System.currentTimeMillis();
- // This cycle will eventually quit when future is
completed by concurrent thread reserving client.
- while (true) {
- try {
- client = fut.get(clientReserveWaitTimeout,
TimeUnit.MILLISECONDS);
+ // This cycle will eventually quit when future is
completed by concurrent thread reserving client.
+ while (true) {
+ // A concurrent thread may have established the
client while this thread was
+ // waiting: return it instead of waiting further
or giving up, even if the
+ // reserve budget has already elapsed — the budget
bounds waiting for a
+ // connection, not the availability of an existing
client.
+ GridCommunicationClient[] curClients0 =
clients.get(nodeId);
- break;
- }
- catch (IgniteFutureTimeoutCheckedException ignored) {
- currTimeout += clientReserveWaitTimeout;
-
- if (log.isDebugEnabled()) {
- log.debug(
- "Still waiting for reestablishing
connection to node " +
- "[nodeId=" + node.id() + ",
waitingTime=" + currTimeout + "ms]"
- );
+ GridCommunicationClient curClient = curClients0 !=
null && connIdx < curClients0.length ?
+ curClients0[connIdx] : null;
+
+ if (curClient != null) {
+ client = curClient;
+
+ break;
+ }
+
+ // Recompute the remaining budget on every
iteration and cap the individual
+ // wait by it, so that the total reserve budget is
enforced even when a single
+ // wait would overshoot it. When the per-wait
timeout is not configured (0),
+ // e.g. because the system worker blocked timeout
is set to a non-positive
+ // value, wait the full remaining budget instead
of aborting before it has
+ // elapsed.
+ long remainingReserveBudget = totalReserveTimeout
- reserveElapsedMillis(reserveStartNanos);
+
+ long clientReserveWaitTimeout = perWaitTimeout > 0
?
+ Math.min(perWaitTimeout,
remainingReserveBudget) : remainingReserveBudget;
+
+ if (clientReserveWaitTimeout <= 0)
+ throw reserveTimeoutException(node,
totalReserveTimeout);
Review Comment:
When client creation fails after consuming the reservation budget, this
check throws a generic reservation timeout before reading the failure already
stored in `fut`, losing the original exception and its causes. It can likewise
hide a stop error when a slow creator returns after shutdown. Resolve an
already-completed future before rejecting a further wait for lack of budget.
##########
modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/GridNioServerWrapper.java:
##########
@@ -489,50 +484,82 @@ public GridNioSession createNioSession(ClusterNode node,
int connIdx) throws Ign
throw new IgniteCheckedException("Local node has
not been started or " +
"fully initialized [isStopping=" +
stateProvider.isStopping() + ']');
+ // The handshake window, re-verified against the
strategy's total timeout (the call
+ // does not advance the backoff state, so the
per-attempt backoff progression is
+ // unchanged).
timeout = connTimeoutStgy.nextTimeout(timeout);
- rcvCnt = safeTcpHandshake(ch,
- node.id(),
- timeout,
- sslMeta,
- new HandshakeMessage(locNode.id(),
- recoveryDesc.incrementConnectCount(),
- recoveryDesc.received(),
- connIdx));
-
- if (rcvCnt == ALREADY_CONNECTED)
- return null;
- else if (rcvCnt == NODE_STOPPING) {
- // Safe to remap on remote node stopping.
- throw new ClusterTopologyCheckedException("Remote
node started stop procedure: " + node.id());
+ long handshakeTimeout = timeout;
+
+ // The deadline covers the handshake and the NIO
session registration: when it
+ // elapses, the channel is closed, which aborts the
in-progress handshake and makes
+ // the registration fail with a ClosedChannelException
- the latter only once the
+ // responsible NIO worker processes the queued
registration operation, so the
+ // registration wait is bounded only under that
precondition. It is armed after the
+ // socket connect, which is bounded by its own connect
timeout, so the connect does
+ // not consume the deadline and the handshake keeps
the full window the strategy
+ // allocated to it.
+ HandshakeTimeoutObject timeoutObj = new
HandshakeTimeoutObject(ch);
+
+ handshakeTimeoutExecutorService.schedule(timeoutObj,
handshakeTimeout, TimeUnit.MILLISECONDS);
+
+ boolean deadlineCancelled = false;
+
+ try {
+ rcvCnt = safeTcpHandshake(ch,
+ node.id(),
+ sslMeta,
+ new HandshakeMessage(locNode.id(),
+ recoveryDesc.incrementConnectCount(),
+ recoveryDesc.received(),
+ connIdx));
+
+ if (rcvCnt == ALREADY_CONNECTED)
+ return null;
+ else if (rcvCnt == NODE_STOPPING) {
+ // Safe to remap on remote node stopping.
+ throw new
ClusterTopologyCheckedException("Remote node started stop procedure: " +
node.id());
+ }
+ else if (rcvCnt == UNKNOWN_NODE)
+ throw new IgniteCheckedException("Remote node
does not observe current node " +
+ "in topology : " + node.id());
+ else if (rcvCnt == NEED_WAIT) {
+ // Should wait for discovery lag without
applying connTimeoutStgy, otherwise cache operations
+ // may hang on waiting inexistent topology,
see IgniteClientConnectTest for test
+ // scenarios with delayed client node join.
+ if (log.isDebugEnabled())
+ log.debug("NEED_WAIT received, handshake
after delay [node = "
+ + node + ", outOfTopologyDelay = " +
DFLT_NEED_WAIT_DELAY + "ms]");
+
+ // The peer-requested discovery lag delay is
not a step of the connection
+ // establishment: end the deadline before the
wait, the next attempt arms a
+ // fresh one.
+ deadlineCancelled = timeoutObj.cancel();
+
+ U.sleep(DFLT_NEED_WAIT_DELAY);
+
+ continue;
+ }
+ else if (rcvCnt < 0)
+ throw new IgniteCheckedException("Unsupported
negative receivedCount [rcvCnt=" + rcvCnt +
+ ", senderNode=" + node + ']');
+
+ recoveryDesc.onHandshake(rcvCnt);
+
+ meta.put(CONSISTENT_ID_META, node.consistentId());
+ meta.put(CONN_IDX_META, connKey);
+ meta.put(GridNioServer.RECOVERY_DESC_META_KEY,
recoveryDesc);
+ meta.put(MSG_SER_CTX.ordinal(),
buildForPeers(stateProvider.ignite(), node));
+
+ ses = nioSrv.createSession(ch, meta, false,
null).get();
Review Comment:
This unbounded `get()` can keep the creating thread blocked after the
establishment deadline expires: closing the channel does not complete the
future until the NIO worker processes registration. A live worker stalled in a
synchronous callback is enough to trigger this; the reservation budget does not
protect the creator. Bound the future wait by the remaining establishment
deadline and safely clean up any session that completes after the wait expires.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
