6 files changed, 178 insertions, 76 deletions

diff --git a/cpp/src/qpid/client/Connector.cpp b/cpp/src/qpid/client/Connector.cpp
index e6355601df..1d1e39cd10 100644
--- a/cpp/src/qpid/client/Connector.cpp
+++ b/cpp/src/qpid/client/Connector.cpp
@@ -220,6 +220,7 @@ bool TCPConnector::closeInternal() {
     bool ret = !closed;
     if (!closed) {
         closed = true;
+        aio->queueForDeletion();
         poller->shutdown();
     }
     if (!joined && receiver.id() != Thread::current().id()) {
@@ -384,14 +385,13 @@ void TCPConnector::run() {
     assert(protect);
     try {
         Dispatcher d(poller);
-	
+
         for (int i = 0; i < 32; i++) {
             aio->queueReadBuffer(new Buff(maxFrameSize));
         }
-	
+
         aio->start(poller);
         d.run();
-        aio->queueForDeletion();
         socket.close();
     } catch (const std::exception& e) {
         QPID_LOG(error, QPID_MSG("FAIL " << identifier << ": " << e.what()));
diff --git a/cpp/src/qpid/client/SslConnector.cpp b/cpp/src/qpid/client/SslConnector.cpp
index 75c3f5677e..a4298dd4ca 100644
--- a/cpp/src/qpid/client/SslConnector.cpp
+++ b/cpp/src/qpid/client/SslConnector.cpp
@@ -221,6 +221,7 @@ bool SslConnector::closeInternal() {
     bool ret = !closed;
     if (!closed) {
         closed = true;
+        aio->queueForDeletion();
         poller->shutdown();
     }
     if (!joined && receiver.id() != Thread::current().id()) {
@@ -386,7 +387,6 @@ void SslConnector::run(){
 	
         aio->start(poller);
         d.run();
-        aio->queueForDeletion();
         socket.close();
     } catch (const std::exception& e) {
         QPID_LOG(error, e.what());
diff --git a/cpp/src/qpid/sys/DispatchHandle.cpp b/cpp/src/qpid/sys/DispatchHandle.cpp
index cbdee7eda6..cd7dec7fa6 100644
--- a/cpp/src/qpid/sys/DispatchHandle.cpp
+++ b/cpp/src/qpid/sys/DispatchHandle.cpp
@@ -21,6 +21,8 @@
 
 #include "DispatchHandle.h"
 
+#include <algorithm>
+
 #include <boost/cast.hpp>
 
 #include <assert.h>
@@ -29,7 +31,6 @@ namespace qpid {
 namespace sys {
 
 DispatchHandle::~DispatchHandle() {
-    stopWatch();
 }
 
 void DispatchHandle::startWatch(Poller::shared_ptr poller0) {
@@ -37,13 +38,21 @@ void DispatchHandle::startWatch(Poller::shared_ptr poller0) {
     bool w = writableCallback;
 
     ScopedLock<Mutex> lock(stateLock);
-    assert(state == IDLE);
+    assert(state == IDLE || state == DELAYED_IDLE);
 
     // If no callbacks set then do nothing (that is what we were asked to do!)
     // TODO: Maybe this should be an assert instead
     if (!r && !w) {
-        state = INACTIVE;
-        return;
+        switch (state) {
+        case IDLE:
+            state = INACTIVE;
+            return;
+        case DELAYED_IDLE:
+            state = DELAYED_INACTIVE;
+            return;
+        default:
+            assert(state == IDLE || state == DELAYED_IDLE);    
+        }        
     }
 
     Poller::Direction d = r ?
@@ -53,9 +62,20 @@ void DispatchHandle::startWatch(Poller::shared_ptr poller0) {
     poller = poller0;
     poller->addFd(*this, d);
     
-    state = r ?
-        (w ? ACTIVE_RW : ACTIVE_R) :
-        ACTIVE_W;
+    switch (state) {
+    case IDLE:
+        state = r ?
+            (w ? ACTIVE_RW : ACTIVE_R) :
+            ACTIVE_W;
+        return;
+    case DELAYED_IDLE:
+        state = r ?
+            (w ? DELAYED_RW : DELAYED_R) :
+            DELAYED_W;
+        return;
+        default:
+            assert(state == IDLE || state == DELAYED_IDLE);            
+    }   
 }
 
 void DispatchHandle::rewatch() {
@@ -93,6 +113,8 @@ void DispatchHandle::rewatch() {
     case ACTIVE_RW:
         // Don't need to do anything already waiting for readable/writable
         break;
+    case ACTIVE_DELETE:
+        assert(state != ACTIVE_DELETE);
     }
 }
 
@@ -130,6 +152,8 @@ void DispatchHandle::rewatchRead() {
         poller->modFd(*this, Poller::INOUT);
         state = ACTIVE_RW;
         break;
+    case ACTIVE_DELETE:
+        assert(state != ACTIVE_DELETE);
     }
 }
 
@@ -167,6 +191,8 @@ void DispatchHandle::rewatchWrite() {
     case ACTIVE_RW:
         // Nothing to do: already waiting for writable
         break;
+    case ACTIVE_DELETE:
+        assert(state != ACTIVE_DELETE);
    }
 }
 
@@ -203,6 +229,8 @@ void DispatchHandle::unwatchRead() {
     case ACTIVE_W:
     case INACTIVE:
         break;
+    case ACTIVE_DELETE:
+        assert(state != ACTIVE_DELETE);
     }
 }
 
@@ -239,6 +267,8 @@ void DispatchHandle::unwatchWrite() {
     case ACTIVE_R:
     case INACTIVE:
         break;
+    case ACTIVE_DELETE:
+        assert(state != ACTIVE_DELETE);
    }
 }
 
@@ -261,6 +291,8 @@ void DispatchHandle::unwatch() {
         poller->modFd(*this, Poller::NONE);
         state = INACTIVE;
         break;
+    case ACTIVE_DELETE:
+        assert(state != ACTIVE_DELETE);
     }            
 }
 
@@ -280,47 +312,72 @@ void DispatchHandle::stopWatch() {
     default:
         state = IDLE;
         break;
+    case ACTIVE_DELETE:
+        assert(state != ACTIVE_DELETE);
     }
     assert(poller);
     poller->delFd(*this);
     poller.reset();
 }
 
+// If we are already in the IDLE state we can't do the callback as we might
+// race to delete and callback at the same time
+// TODO: might be able to fix this by adding a new state, but would make
+// the state machine even more complex
 void DispatchHandle::call(Callback iCb) {
     assert(iCb);
     ScopedLock<Mutex> lock(stateLock);
-    interruptedCallbacks.push(iCb);
-    
-    (void) poller->interrupt(*this);
+    switch (state) {
+    case IDLE:
+    case ACTIVE_DELETE:
+        assert(false);
+        return;
+    default:
+        interruptedCallbacks.push(iCb);
+        assert(poller);
+        (void) poller->interrupt(*this);
+    }
 }
 
 // The slightly strange switch structure
 // is to ensure that the lock is released before
 // we do the delete
 void DispatchHandle::doDelete() {
-    // Ensure that we're no longer watching anything
-    stopWatch();
-
-    // If we're in the middle of a callback defer the delete
     {
     ScopedLock<Mutex> lock(stateLock);
+    // Ensure that we're no longer watching anything
     switch (state) {
+    case DELAYED_R:
+    case DELAYED_W:
+    case DELAYED_RW:
+    case DELAYED_INACTIVE:
+        assert(poller);
+        poller->delFd(*this);
+        poller.reset();
+        // Fallthrough
     case DELAYED_IDLE:
-    case DELAYED_DELETE:
         state = DELAYED_DELETE;
+        // Fallthrough
+    case DELAYED_DELETE:
+    case ACTIVE_DELETE:
         return;
     case IDLE:
         break;
     default:
-        // Can only get out of stopWatch() in DELAYED_IDLE/DELAYED_DELETE/IDLE states
-        assert(false);
+        state = ACTIVE_DELETE;
+        assert(poller);
+        (void) poller->interrupt(*this);
+        poller->delFd(*this);
+        return;
     }
     }
-    // If we're not then do it right away
+    // If we're IDLE we can do this right away
     delete this;
 }
 
 void DispatchHandle::processEvent(Poller::EventType type) {
+    CallbackQueue callbacks;
+
     // Note that we are now doing the callbacks
     {
     ScopedLock<Mutex> lock(stateLock);
@@ -336,6 +393,16 @@ void DispatchHandle::processEvent(Poller::EventType type) {
     case ACTIVE_RW:
         state = DELAYED_RW;
         break;
+    case ACTIVE_DELETE:
+        // Need to make sure we clean up any pending callbacks in this case
+        std::swap(callbacks, interruptedCallbacks);
+        goto saybyebye;
+    // Can get here in idle if we are stopped in a different thread
+    // just after we return with this handle in Poller::wait
+    case IDLE:
+    // Can get here in INACTIVE if a non connection thread unwatches
+    // whilst we were stuck in the above lock 
+    case INACTIVE:
     // Can only get here in a DELAYED_* state in the rare case
     // that we're already here for reading and we get activated for
     // writing and we can write (it might be possible the other way
@@ -348,9 +415,9 @@ void DispatchHandle::processEvent(Poller::EventType type) {
     case DELAYED_IDLE:
     case DELAYED_DELETE:
         return;
-    default:
-        assert(false);
     }
+    
+    std::swap(callbacks, interruptedCallbacks);
     }
 
     // Do callbacks - whilst we are doing the callbacks we are prevented from processing
@@ -378,8 +445,8 @@ void DispatchHandle::processEvent(Poller::EventType type) {
         break;
     case Poller::INTERRUPTED:
         {
-        ScopedLock<Mutex> lock(stateLock);
-        assert(interruptedCallbacks.size() > 0);
+        // We could only be interrupted if we also had a callback to do
+        assert(callbacks.size() > 0);
         // We'll actually do the interrupt below
         }
         break;
@@ -387,16 +454,18 @@ void DispatchHandle::processEvent(Poller::EventType type) {
         assert(false);
     }
 
-    {
-    ScopedLock<Mutex> lock(stateLock);
-    // If we've got a pending interrupt do it now
-    while (interruptedCallbacks.size() > 0) {
-        Callback cb = interruptedCallbacks.front();
+    // If we have any callbacks do them now -
+    // (because we use a copy from before the previous callbacks we won't
+    //  do anything yet that was just added) 
+    while (callbacks.size() > 0) {
+        Callback cb = callbacks.front();
         assert(cb);
         cb(*this);
-        interruptedCallbacks.pop();
+        callbacks.pop();
     }
 
+    {
+    ScopedLock<Mutex> lock(stateLock);
     // If any of the callbacks re-enabled reading/writing then actually
     // do it now
     switch (state) {
@@ -425,7 +494,9 @@ void DispatchHandle::processEvent(Poller::EventType type) {
     case DELAYED_DELETE:
         break;
     }
-    }      
+    }
+
+saybyebye:
     delete this;
 }
 
diff --git a/cpp/src/qpid/sys/DispatchHandle.h b/cpp/src/qpid/sys/DispatchHandle.h
index ffcbd80f7e..fb114ce5be 100644
--- a/cpp/src/qpid/sys/DispatchHandle.h
+++ b/cpp/src/qpid/sys/DispatchHandle.h
@@ -65,6 +65,7 @@ private:
     Mutex stateLock;
     enum {
         IDLE, INACTIVE, ACTIVE_R, ACTIVE_W, ACTIVE_RW,
+        ACTIVE_DELETE,
         DELAYED_IDLE, DELAYED_INACTIVE, DELAYED_R, DELAYED_W, DELAYED_RW,
         DELAYED_DELETE
     } state;
diff --git a/cpp/src/qpid/sys/Poller.h b/cpp/src/qpid/sys/Poller.h
index 8e9f67fefd..96b4b9c361 100644
--- a/cpp/src/qpid/sys/Poller.h
+++ b/cpp/src/qpid/sys/Poller.h
@@ -86,8 +86,9 @@ public:
     //   with the handle and the INTERRUPTED event type
     // if it returns false then the handle is not being monitored by the poller
     // - This can either be because it has just received an event which has been
-    //   reported and has not been reenabled since. Or because it was removed
-    //   from the monitoring set
+    //   reported and has not been reenabled since.
+    // - Because it was removed from the monitoring set
+    // - Or because it is already being interrupted 
     bool interrupt(PollerHandle& handle);
     
     // Poller run loop
diff --git a/cpp/src/qpid/sys/epoll/EpollPoller.cpp b/cpp/src/qpid/sys/epoll/EpollPoller.cpp
index 10705e12da..42b5d8b1aa 100644
--- a/cpp/src/qpid/sys/epoll/EpollPoller.cpp
+++ b/cpp/src/qpid/sys/epoll/EpollPoller.cpp
@@ -54,17 +54,20 @@ class PollerHandlePrivate {
         INACTIVE,
         HUNGUP,
         MONITORED_HUNGUP,
+        INTERRUPTED,
         DELETED
     };
 
     int fd;
     ::__uint32_t events;
+    PollerHandle* pollerHandle;
     FDStat stat;
     Mutex lock;
 
-    PollerHandlePrivate(int f) :
+    PollerHandlePrivate(int f, PollerHandle* p) :
       fd(f),
       events(0),
+      pollerHandle(p),
       stat(ABSENT) {
     }
 
@@ -101,6 +104,14 @@ class PollerHandlePrivate {
         stat = HUNGUP;
     }
 
+    bool isInterrupted() const {
+        return stat == INTERRUPTED;
+    }
+
+    void setInterrupted() {
+        stat = INTERRUPTED;
+    }
+
     bool isDeleted() const {
         return stat == DELETED;
     }
@@ -111,7 +122,7 @@ class PollerHandlePrivate {
 };
 
 PollerHandle::PollerHandle(const IOHandle& h) :
-    impl(new PollerHandlePrivate(toFd(h.impl)))
+    impl(new PollerHandlePrivate(toFd(h.impl), this))
 {}
 
 PollerHandle::~PollerHandle() {
@@ -120,6 +131,10 @@ PollerHandle::~PollerHandle() {
     if (impl->isDeleted()) {
     	return;
     }
+    if (impl->isInterrupted()) {
+        impl->setDeleted();
+        return;
+    }
     if (impl->isActive()) {
         impl->setDeleted();
     }
@@ -243,23 +258,21 @@ class PollerPrivate {
         ::close(epollFd);
     }
     
-    void interrupt(bool all=false) {
+    void interrupt() {
 	    ::epoll_event epe;
-	    if (all) {
-	        // Not EPOLLONESHOT, so we eventually get all threads
-		    epe.events = ::EPOLLIN;
-		    epe.data.u64 = 0; // Keep valgrind happy
-	    } else {
-	    	// Use EPOLLONESHOT so we only wake a single thread
-		    epe.events = ::EPOLLIN | ::EPOLLONESHOT;
-		    epe.data.u64 = 0; // Keep valgrind happy
-		    epe.data.ptr = &static_cast<PollerHandle&>(interruptHandle);	    
-	    }
+    	// Use EPOLLONESHOT so we only wake a single thread
+	    epe.events = ::EPOLLIN | ::EPOLLONESHOT;
+	    epe.data.u64 = 0; // Keep valgrind happy
+	    epe.data.ptr = &static_cast<PollerHandle&>(interruptHandle);
 	    QPID_POSIX_CHECK(::epoll_ctl(epollFd, EPOLL_CTL_MOD, alwaysReadableFd, &epe));	
     }
     
     void interruptAll() {
-    	interrupt(true);
+        ::epoll_event epe;
+        // Not EPOLLONESHOT, so we eventually get all threads
+        epe.events = ::EPOLLIN;
+        epe.data.u64 = 0; // Keep valgrind happy
+        QPID_POSIX_CHECK(::epoll_ctl(epollFd, EPOLL_CTL_MOD, alwaysReadableFd, &epe));  
     }
 };
 
@@ -281,7 +294,7 @@ void Poller::addFd(PollerHandle& handle, Direction dir) {
         epe.events = eh.events | PollerPrivate::directionToEpollEvent(dir);
     }
     epe.data.u64 = 0; // Keep valgrind happy
-    epe.data.ptr = &handle;
+    epe.data.ptr = &eh;
 
     QPID_POSIX_CHECK(::epoll_ctl(impl->epollFd, op, eh.fd, &epe));
 
@@ -312,7 +325,7 @@ void Poller::modFd(PollerHandle& handle, Direction dir) {
     ::epoll_event epe;
     epe.events = PollerPrivate::directionToEpollEvent(dir) | ::EPOLLONESHOT;
     epe.data.u64 = 0; // Keep valgrind happy
-    epe.data.ptr = &handle;
+    epe.data.ptr = &eh;
 
     QPID_POSIX_CHECK(::epoll_ctl(impl->epollFd, EPOLL_CTL_MOD, eh.fd, &epe));
 
@@ -329,7 +342,7 @@ void Poller::rearmFd(PollerHandle& handle) {
     ::epoll_event epe;
     epe.events = eh.events;
     epe.data.u64 = 0; // Keep valgrind happy
-    epe.data.ptr = &handle;
+    epe.data.ptr = &eh;
 
     QPID_POSIX_CHECK(::epoll_ctl(impl->epollFd, EPOLL_CTL_MOD, eh.fd, &epe));
 
@@ -355,15 +368,14 @@ bool Poller::interrupt(PollerHandle& handle) {
 	{
 	    PollerHandlePrivate& eh = *handle.impl;
 	    ScopedLock<Mutex> l(eh.lock);
-	    if (eh.isInactive()) {
+	    if (!eh.isActive()) {
 	    	return false;
 	    }
 	    ::epoll_event epe;
 	    epe.events = 0;
 	    epe.data.u64 = 0; // Keep valgrind happy
-	    epe.data.ptr = &eh;
 	    QPID_POSIX_CHECK(::epoll_ctl(impl->epollFd, EPOLL_CTL_MOD, eh.fd, &epe));
-	    eh.setInactive();
+	    eh.setInterrupted();
 	}
 
 	PollerPrivate::InterruptHandle& ih = impl->interruptHandle;
@@ -422,37 +434,54 @@ Poller::Event Poller::wait(Duration timeout) {
 #else
         int rc = ::epoll_pwait(impl->epollFd, &epe, 1, timeoutMs, &impl->sigMask);
 #endif
-        // Check for shutdown
-        if (impl->isShutdown) {
-            PollerHandleDeletionManager.markAllUnusedInThisThread();
-            return Event(0, SHUTDOWN);
-        }
 
         if (rc ==-1 && errno != EINTR) {
             QPID_POSIX_CHECK(rc);
         } else if (rc > 0) {
             assert(rc == 1);
-            PollerHandle* handle = static_cast<PollerHandle*>(epe.data.ptr);
+            void* dataPtr = epe.data.ptr;
+
+            // Check if this is an interrupt
+            PollerPrivate::InterruptHandle& interruptHandle = impl->interruptHandle;
+            if (dataPtr == &interruptHandle) {
+                PollerHandle* wrappedHandle = 0;
+                {
+                ScopedLock<Mutex> l(interruptHandle.impl->lock);
+                if (interruptHandle.impl->isActive()) {
+                    wrappedHandle = interruptHandle.getHandle();
+                    // If there is an interrupt queued behind this one we need to arm it
+                    // We do it this way so that another thread can pick it up
+                    if (interruptHandle.queuedHandles()) {
+                        impl->interrupt();
+                        interruptHandle.impl->setActive();
+                    } else {
+                        interruptHandle.impl->setInactive();
+                    }
+                }
+                }
+                if (wrappedHandle) {
+                    ScopedLock<Mutex> l(wrappedHandle->impl->lock);
+                    if (!wrappedHandle->impl->isDeleted()) {
+                        wrappedHandle->impl->setInactive();
+                        return Event(wrappedHandle, INTERRUPTED);
+                    }
+                    PollerHandleDeletionManager.markForDeletion(wrappedHandle->impl);
+                }
+                continue;
+            }
+
+            // Check for shutdown
+            if (impl->isShutdown) {
+                PollerHandleDeletionManager.markAllUnusedInThisThread();
+                return Event(0, SHUTDOWN);
+            }
 
-            PollerHandlePrivate& eh = *handle->impl;
+            PollerHandlePrivate& eh = *static_cast<PollerHandlePrivate*>(dataPtr);
             ScopedLock<Mutex> l(eh.lock);
             
             // the handle could have gone inactive since we left the epoll_wait
             if (eh.isActive()) {
-
-                // Check if this is an interrupt
-                if (handle == &impl->interruptHandle) {
-                	PollerHandle* wrappedHandle = impl->interruptHandle.getHandle();
-                	// If there is an interrupt queued behind this one we need to arm it
-                	// We do it this way so that another thread can pick it up
-                	if (impl->interruptHandle.queuedHandles()) {
-                		impl->interrupt();
-                		eh.setActive();
-                	} else {
-                		eh.setInactive();
-                	}
-                	return Event(wrappedHandle, INTERRUPTED);
-                }
+                PollerHandle* handle = eh.pollerHandle;
 
                 // If the connection has been hungup we could still be readable
                 // (just not writable), allow us to readable until we get here again