mroeppis commented on code in PR #13791:
URL: https://github.com/apache/ignite/pull/13791#discussion_r4504648769
##########
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:
My original implementation establishes a communication client and then
publishes it to other threads through a shared future and the node-client map.
The problem is that publishing the client and completing the shared future
are not atomic with respect to `stop()`.
Consider this interleaving:
1. Thread A establishes a new communication client.
2. Thread A publishes the client in the node-client map.
3. Thread B calls `stop()` and completes the shared future exceptionally.
4. Thread C concurrently retrieves the client from the node-client map.
5. Thread A attempts to complete the future with the client, but fails
because the future has already been completed by `stop()`.
6. Thread A removes and closes the client.
Thread C may therefore receive a client that is being closed immediately
afterwards.
This is a genuine publication race: the client becomes visible to other
threads before the implementation knows whether shutdown has won the completion
race.
The important detail is that the shared future itself is also a publication
mechanism. Protecting only the map would not be sufficient: a waiting thread
could obtain the client directly from the future.
**Why the finding is valid**: Shutdown and client publication need a common
synchronization mechanism. Otherwise, a client can escape to another thread
even though the creator subsequently decides that it must be discarded.
My revised implementation addresses this by coordinating the stopping flag,
future completion, and client publication under `stopLock`.
--
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]
