From 195b9a2ea5124d7d887e6addffbe779320dbcaab Mon Sep 17 00:00:00 2001 From: Keith Wall Date: Mon, 22 Aug 2011 15:31:08 +0000 Subject: QPID-2904: Avoid race condition between SubFlushRunner and MinaAcceptor threads. Send lock was not being held during MessageStop processing. Applying patch by Robbie Gemmell and myself. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1160304 13f79535-47bb-0310-9956-ffa450edef68 --- .../server/subscription/Subscription_0_10.java | 25 +++++++++++++++------- 1 file changed, 17 insertions(+), 8 deletions(-) (limited to 'qpid/java') diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java index 1bfc2205f2..8a5f9ffef5 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java @@ -109,15 +109,15 @@ public class Subscription_0_10 implements Subscription, FlowCreditManager.FlowCr private final MessageAcquireMode _acquireMode; private MessageFlowMode _flowMode; private final ServerSession _session; - private AtomicBoolean _stopped = new AtomicBoolean(true); + private final AtomicBoolean _stopped = new AtomicBoolean(true); private static final Struct[] EMPTY_STRUCT_ARRAY = new Struct[0]; private LogActor _logActor; - private Map _properties = new ConcurrentHashMap(); + private final Map _properties = new ConcurrentHashMap(); private UUID _id; private String _traceExclude; private String _trace; - private long _createTime = System.currentTimeMillis(); + private final long _createTime = System.currentTimeMillis(); private final AtomicLong _deliveredCount = new AtomicLong(0); private final Map _arguments; @@ -727,13 +727,22 @@ public class Subscription_0_10 implements Subscription, FlowCreditManager.FlowCr public void stop() { - if(_state.compareAndSet(State.ACTIVE, State.SUSPENDED)) + try { - _stateListener.stateChange(this, State.ACTIVE, State.SUSPENDED); + getSendLock(); + + if(_state.compareAndSet(State.ACTIVE, State.SUSPENDED)) + { + _stateListener.stateChange(this, State.ACTIVE, State.SUSPENDED); + } + _stopped.set(true); + FlowCreditManager_0_10 creditManager = getCreditManager(); + creditManager.clearCredit(); + } + finally + { + releaseSendLock(); } - _stopped.set(true); - FlowCreditManager_0_10 creditManager = getCreditManager(); - creditManager.clearCredit(); } public void addCredit(MessageCreditUnit unit, long value) -- cgit v1.2.1