AMQ-1632 Fix mdb stopping problem and improve some thread-safety

git-svn-id: https://svn.apache.org/repos/asf/activemq/branches/activemq-4.1@640336 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java

@@ -37,10 +37,11 @@ import org.apache.activemq.util.JMSExceptionSupport;
  */
 public class ActiveMQSessionExecutor implements Task {
 
-    private ActiveMQSession session;
+    private final ActiveMQSession session;
-    private MessageDispatchChannel messageQueue = new MessageDispatchChannel();
+    private final MessageDispatchChannel messageQueue = new MessageDispatchChannel();
-    private boolean dispatchedBySessionPool;
+    private volatile boolean dispatchedBySessionPool;
-    private TaskRunner taskRunner;
+    //volatile required to avoid double-checked locking problem.
+    private volatile TaskRunner taskRunner;
 
     ActiveMQSessionExecutor(ActiveMQSession session) {
         this.session = session;
@@ -65,9 +66,13 @@ public class ActiveMQSessionExecutor implements Task {
         if (!dispatchedBySessionPool) {
             if (session.isSessionAsyncDispatch()) {
                 try {
+                    if (taskRunner == null) {
+                        synchronized (this) {
                             if (taskRunner == null) {
                                 taskRunner = session.connection.getSessionTaskRunner().createTaskRunner(this, "ActiveMQ Session: " + session.getSessionId());
                             }
+                        }
+                    }
                     taskRunner.wakeup();
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();

activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java

@@ -180,7 +180,8 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
                                     dequeueCounter++;
                                     dispatched.remove(node);
                                     node.getRegionDestination().getDestinationStatistics().getDequeues().increment();
-                                    prefetchExtension--;
+                                    prefetchExtension=Math.max(0,prefetchExtension-1);
+                                    dispatchMatched();
                                 }
                             }
 

activemq-core/src/main/java/org/apache/activemq/thread/PooledTaskRunner.java

@@ -29,10 +29,13 @@ class PooledTaskRunner implements TaskRunner {
     private final Executor executor;
     private final Task task;
     private final Runnable runable;
+    //guarded by runable
     private boolean queued;
     private boolean shutdown;
     private boolean iterating;
-    private Thread runningThread;
+
+    //setting is not guarded by runable; 
+    private volatile Thread runningThread;
     
     public PooledTaskRunner(Executor executor, Task task, int maxIterationsPerRun) {
         this.executor = executor;

activemq-ra/src/main/java/org/apache/activemq/ra/ActiveMQEndpointWorker.java

@@ -61,25 +61,24 @@ public class ActiveMQEndpointWorker {
         }
     }
 
-    protected ActiveMQResourceAdapter adapter;
+    protected final ActiveMQEndpointActivationKey endpointActivationKey;
-    protected ActiveMQEndpointActivationKey endpointActivationKey;
+    protected final MessageEndpointFactory endpointFactory;
-    protected MessageEndpointFactory endpointFactory;
+    protected final WorkManager workManager;
-    protected WorkManager workManager;
+    protected final boolean transacted;
-    protected boolean transacted;
 
+    private final ActiveMQDestination dest;
+    private final Work connectWork;
 
+    //access to all non-final variables guarded by connectWork
     private ConnectionConsumer consumer;
     private ServerSessionPoolImpl serverSessionPool;
-    private ActiveMQDestination dest;
     private boolean running;
-    private Work connectWork;
+    private ActiveMQConnection connection;
-    protected ActiveMQConnection connection;
-    
     private long reconnectDelay=INITIAL_RECONNECT_DELAY;
 
 
     /**
-     * @param s
+     * @param s session to close
      */
     public static void safeClose(Session s) {
         try {
@@ -88,38 +87,40 @@ public class ActiveMQEndpointWorker {
             }
         }
         catch (JMSException e) {
+            //ignore
         }
     }
 
     /**
-     * @param c
+     * @param c connection to close
      */
-    public static void safeClose(Connection c) {
+    private static void safeClose(Connection c) {
         try {
             if (c != null) {
                 c.close();
             }
         }
         catch (JMSException e) {
+            //ignore
         }
     }
 
     /**
-     * @param cc
+     * @param cc ConnectionConsumer to close
      */
-    public static void safeClose(ConnectionConsumer cc) {
+    private static void safeClose(ConnectionConsumer cc) {
         try {
             if (cc != null) {
                 cc.close();
             }
         }
         catch (JMSException e) {
+            //ignore
         }
     }
 
     public ActiveMQEndpointWorker(final ActiveMQResourceAdapter adapter, ActiveMQEndpointActivationKey key) throws ResourceException {
         this.endpointActivationKey = key;
-        this.adapter = adapter;
         this.endpointFactory = endpointActivationKey.getMessageEndpointFactory();
         this.workManager = adapter.getBootstrapContext().getWorkManager();
         try {
@@ -135,7 +136,7 @@ public class ActiveMQEndpointWorker {
             }
 
             synchronized public void run() {
-                if( !isRunning() )
+                if( !running)
                     return;
                 if( connection!=null )
                     return;
@@ -187,7 +188,8 @@ public class ActiveMQEndpointWorker {
 
     }
 
-    synchronized public void start() throws WorkException, ResourceException {
+    public void start() throws ResourceException {
+        synchronized (connectWork) {
             if (running)
                 return;
             running = true;
@@ -195,25 +197,28 @@ public class ActiveMQEndpointWorker {
             log.debug("Starting");
             serverSessionPool = new ServerSessionPoolImpl(this, endpointActivationKey.getActivationSpec().getMaxSessionsIntValue());
             connect();
+        }
         log.debug("Started");
     }
 
     /**
      *
+     * @throws InterruptedException
      */
-    synchronized public void stop() throws InterruptedException {
+    public void stop() throws InterruptedException {
+        synchronized (connectWork) {
             if (!running)
                 return;
             running = false;
             serverSessionPool.close();
             disconnect();
         }
-
-    private boolean isRunning() {
-        return running;
     }
 
-    synchronized private void connect() {
+    /**
+     * Calls must be synchronized on connectWork
+     */
+    private void connect() {
         if (!running)
             return;
 
@@ -226,9 +231,9 @@ public class ActiveMQEndpointWorker {
     }
 
     /**
-     * 
+     *  Calls must be synchronized on connectWork
      */
-    synchronized private void disconnect() {
+    private void disconnect() {
         safeClose(consumer);
         consumer=null;
         safeClose(connection);
@@ -238,26 +243,28 @@ public class ActiveMQEndpointWorker {
     private void reconnect(JMSException error) {
         log.debug("Reconnect cause: ", error);
         long reconnectDelay;
-        synchronized(this) {
+        try {
+            synchronized (connectWork) {
                 reconnectDelay = this.reconnectDelay;
                 // Only log errors if the server is really down.. And not a temp failure.
                 if (reconnectDelay == MAX_RECONNECT_DELAY) {
                     log.error("Endpoint connection to JMS broker failed: " + error.getMessage());
                     log.error("Endpoint will try to reconnect to the JMS broker in " + (MAX_RECONNECT_DELAY / 1000) + " seconds");
                 }
-        }
-        try {
                 disconnect();
+            }
             Thread.sleep(reconnectDelay);
 
-            synchronized(this) {
+            synchronized (connectWork) {
                 // Use exponential rollback.
                 this.reconnectDelay *= 2;
                 if (this.reconnectDelay > MAX_RECONNECT_DELAY)
                     this.reconnectDelay = MAX_RECONNECT_DELAY;
-            }
                 connect();
-        } catch(InterruptedException e) {}
+            }
+        } catch (InterruptedException e) {
+            Thread.interrupted();
+        }
     }
 
     protected void registerThreadSession(Session session) {
@@ -268,6 +275,12 @@ public class ActiveMQEndpointWorker {
         threadLocal.set(null);
     }
 
+    ActiveMQConnection getConnection() {
+        synchronized (connectWork) {
+            return connection;
+        }
+    }
+
     private String emptyToNull(String value) {
         if (value == null || value.length() == 0) {
             return null;

activemq-ra/src/main/java/org/apache/activemq/ra/ServerSessionPoolImpl.java

@@ -60,7 +60,7 @@ public class ServerSessionPoolImpl implements ServerSessionPool {
     private ServerSessionImpl createServerSessionImpl() throws JMSException {
         ActiveMQActivationSpec activationSpec = activeMQAsfEndpointWorker.endpointActivationKey.getActivationSpec();
         int acknowledge = (activeMQAsfEndpointWorker.transacted) ? Session.SESSION_TRANSACTED : activationSpec.getAcknowledgeModeForSession();
-        final ActiveMQSession session = (ActiveMQSession)activeMQAsfEndpointWorker.connection.createSession(activeMQAsfEndpointWorker.transacted, acknowledge);
+        final ActiveMQSession session = (ActiveMQSession)activeMQAsfEndpointWorker.getConnection().createSession(activeMQAsfEndpointWorker.transacted, acknowledge);
         MessageEndpoint endpoint;
         try {
             int batchSize = 0;
@@ -227,7 +227,7 @@ public class ServerSessionPoolImpl implements ServerSessionPool {
         } else if (s instanceof ActiveMQTopicSession) {
             session = (ActiveMQSession) s;
         } else {
-            activeMQAsfEndpointWorker.connection
+            activeMQAsfEndpointWorker.getConnection()
                     .onAsyncException(new JMSException(
                             "Session pool provided an invalid session type: "
                                     + s.getClass()));