Fix race condition between concurrent updates of remaining, singleMsgs collections in DistributedProcess.
@@ -217,21 +217,16 @@ public class DistributedProcess<I extends Serializable, R extends Serializable>
p.resFut.listen(f -> sendSingleMessage(p));
}
else if (F.eq(ctx.localNodeId(), p.crdId)) {
- boolean rmvd, isEmpty;
+ boolean isEmpty = false;
synchronized (mux) {
- rmvd = p.remaining.remove(leftNodeId);
-
+ if (p.remaining.remove(leftNodeId))
isEmpty = p.remaining.isEmpty();
}
- if (rmvd) {
- assert !p.singleMsgs.containsKey(leftNodeId);
-
if (isEmpty)
finishProcess(p);
}
- }
});
}
}, EVT_NODE_FAILED, EVT_NODE_LEFT);
@@ -316,20 +311,17 @@ public class DistributedProcess<I extends Serializable, R extends Serializable>
Process p = processes.computeIfAbsent(msg.processId(), id -> new Process(msg.processId()));
p.initCrdFut.listen(f -> {
- boolean rmvd, isEmpty;
+ boolean isEmpty;
synchronized (mux) {
- rmvd = p.remaining.remove(nodeId);
+ if (p.remaining.remove(nodeId))
+ p.singleMsgs.put(nodeId, msg);
isEmpty = p.remaining.isEmpty();
}
- if (rmvd) {
- p.singleMsgs.put(nodeId, msg);
-
if (isEmpty)
finishProcess(p);
- }
});
}
