From 2c10159e28ff85e52840d5c6964123e4c410458d Mon Sep 17 00:00:00 2001 From: Robert Godfrey Date: Mon, 2 Jan 2012 10:01:21 +0000 Subject: QPID-3713 : Implement producer side flow control for 0-10 in Java Broker git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1226382 13f79535-47bb-0310-9956-ffa450edef68 --- .../main/java/org/apache/qpid/transport/Range.java | 8 +++++++- .../java/org/apache/qpid/transport/Session.java | 22 ++++++++++++++++++---- 2 files changed, 25 insertions(+), 5 deletions(-) (limited to 'java/common') diff --git a/java/common/src/main/java/org/apache/qpid/transport/Range.java b/java/common/src/main/java/org/apache/qpid/transport/Range.java index f976337788..c47171dc4b 100644 --- a/java/common/src/main/java/org/apache/qpid/transport/Range.java +++ b/java/common/src/main/java/org/apache/qpid/transport/Range.java @@ -185,6 +185,12 @@ public abstract class Range implements RangeSet } } + public String toString() + { + return "[" + point + ", " + point + "]"; + } + + } private static class RangeImpl extends Range @@ -283,7 +289,7 @@ public abstract class Range implements RangeSet return range; } - @Override + public void remove() { throw new UnsupportedOperationException(); diff --git a/java/common/src/main/java/org/apache/qpid/transport/Session.java b/java/common/src/main/java/org/apache/qpid/transport/Session.java index 3e823ba6fe..d391181217 100644 --- a/java/common/src/main/java/org/apache/qpid/transport/Session.java +++ b/java/common/src/main/java/org/apache/qpid/transport/Session.java @@ -61,7 +61,7 @@ import java.util.concurrent.atomic.AtomicBoolean; public class Session extends SessionInvoker { private static final Logger log = Logger.get(Session.class); - + public enum State { NEW, DETACHED, RESUMING, OPEN, CLOSING, CLOSED } static class DefaultSessionListener implements SessionListener @@ -96,6 +96,9 @@ public class Session extends SessionInvoker private final long timeout = Long.getLong(ClientProperties.QPID_SYNC_OP_TIMEOUT, Long.getLong(ClientProperties.AMQJ_DEFAULT_SYNCWRITE_TIMEOUT, ClientProperties.DEFAULT_SYNC_OPERATION_TIMEOUT)); + private final long blockedSendTimeout = Long.getLong("qpid.flow_control_wait_failure", timeout); + private long blockedSendReportingPeriod = Long.getLong("qpid.flow_control_wait_notify_period",5000L); + private boolean autoSync = false; private boolean incomingInit; @@ -228,10 +231,21 @@ public class Session extends SessionInvoker { try { - if (!credit.tryAcquire(timeout, TimeUnit.MILLISECONDS)) + long wait = blockedSendTimeout > blockedSendReportingPeriod ? blockedSendReportingPeriod : + blockedSendTimeout; + long totalWait = 1L; + while(totalWait <= blockedSendTimeout && !credit.tryAcquire(wait, TimeUnit.MILLISECONDS)) + { + totalWait+=wait; + log.warn("Message send delayed by " + (totalWait)/1000 + "s due to broker enforced flow control"); + + + } + if(totalWait > blockedSendTimeout) { + log.error("Message send failed due to timeout waiting on broker enforced flow control"); throw new SessionException - ("timed out waiting for message credit"); + ("timed out waiting for message credit"); } } catch (InterruptedException e) @@ -815,7 +829,7 @@ public class Session extends SessionInvoker while (w.hasTime() && state != CLOSED && lt(maxComplete, point)) { checkFailoverRequired("Session sync was interrupted by failover."); - log.debug("%s waiting for[%d]: %d, %s", this, point, maxComplete, commands); + log.debug("%s waiting for[%d]: %d, %s", this, point, maxComplete, Arrays.asList(commands)); w.await(); } -- cgit v1.2.1