Avoid the excessive ping-pong when many threads are blocking.
If less then 100 give all a chance to send, if more only select 1.
@@ -26,6 +26,7 @@ public final class SourceSession implements ReplyHandler {
private final ThrottlePolicy throttlePolicy;
private volatile double timeout;
private volatile int pendingCount = 0;
+ private int blockedCount = 0;
private boolean closed = false;
/**
@@ -172,9 +173,14 @@ public final class SourceSession implements ReplyHandler {
return res;
}
synchronized (lock) {
+ try {
+ blockedCount++;
while (!closed && !throttlePolicy.canSend(msg, pendingCount)) {
lock.wait(100);
}
+ } finally {
+ blockedCount--;
+ }
}
}
}
@@ -192,7 +198,11 @@ public final class SourceSession implements ReplyHandler {
throttlePolicy.processReply(reply);
}
done = (closed && pendingCount == 0);
+ if (blockedCount < 100) {
lock.notifyAll();
+ } else {
+ lock.notify();
+ }
}
if (reply.getTrace().shouldTrace(TraceLevel.COMPONENT)) {
reply.getTrace().trace(TraceLevel.COMPONENT,
