- Eliminated a race condition on deletion of PollerHandles

* Added DeletionManager class to delete handles
* Used to stop PollerHandles being deleted whilst still in use


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@599390 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 162 insertions, 5 deletions

diff --git a/cpp/src/qpid/sys/DeletionManager.h b/cpp/src/qpid/sys/DeletionManager.h
new file mode 100644
index 0000000000..43154eb98e
--- /dev/null
+++ b/cpp/src/qpid/sys/DeletionManager.h
@@ -0,0 +1,138 @@
+#ifndef _sys_DeletionManager_h
+#define _sys_DeletionManager_h
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <vector>
+#include <algorithm>
+#include <boost/shared_ptr.hpp>
+
+namespace qpid {
+namespace sys {
+
+struct deleter
+{
+  template <typename T>
+  void operator()(T* ptr){ delete ptr;}
+};
+
+/**
+ * DeletionManager keeps track of handles that need to be deleted but may still be
+ * in use by one of the threads concurrently.
+ * 
+ * The mode of operation is like this:
+ * - When we want to delete but we might still be using the handle we
+ *   * Transfer ownership of the handle to this class
+ *   * Mark the handle as (potentially) in use by every thread
+ * - Then subsequently at points where the thread code knows it isn't
+ *   using any handles it declares that it is using no handles
+ * - When the last thread declares no use of a handle it automatically
+ *   gets deleted by the shared_ptr implementation
+ * 
+ * The class only has static members and data and so can only be used once for
+ * any particular handle type
+ */
+template <typename H>
+class DeletionManager 
+{
+public:
+    // Mark every thread as using the handle - it will be deleted
+    // below after every thread marks the handle as unused
+    static void markForDeletion(H* handle) {
+        allThreadsStatuses.addHandle(shared_ptr(handle));
+    }
+    
+    // Mark this thread is not using any handle -
+    // handles get deleted here when no one else
+    // is using them either 
+    static void markAllUnusedInThisThread() {
+        static __thread ThreadStatus* threadStatus = 0;
+
+        // Thread local vars can't be dynamically constructed so we need
+        // to check whether we've made it yet and construct it if not
+        // (no locking necessary for the check as it's thread local!)
+        if (!threadStatus) {
+            threadStatus = new ThreadStatus;
+            allThreadsStatuses.addThreadStatus(threadStatus);
+        }
+
+        ScopedLock<Mutex> l(threadStatus->lock);
+        
+        // The actual deletions will happen here when all the shared_ptr
+        // ref counts hit 0 (that is when every thread marks the handle unused)
+        threadStatus->handles.clear();
+    }
+
+private:
+    typedef boost::shared_ptr<H> shared_ptr;
+
+    // In theory we know that we never need more handles than the number of
+    // threads runnning so we could use a fixed size array. However at this point
+    // in the code we don't have easy access to this information. 
+    struct ThreadStatus
+    {
+        Mutex lock;
+        std::vector<shared_ptr> handles;
+    };
+
+    class AllThreadsStatuses
+    {
+        Mutex lock;
+        std::vector<ThreadStatus*> statuses;
+
+        struct handleAdder
+        {
+            shared_ptr handle;
+
+            handleAdder(shared_ptr h): handle(h) {}
+        
+            void operator()(ThreadStatus* ptr) {
+                ScopedLock<Mutex> l(ptr->lock);
+                ptr->handles.push_back(handle);
+            }
+        };
+
+    public:
+        // Need this to be able to do static initialisation
+        explicit AllThreadsStatuses(int) {}
+
+        ~AllThreadsStatuses() {
+            ScopedLock<Mutex> l(lock);
+            std::for_each(statuses.begin(), statuses.end(), deleter());
+        }
+
+        void addThreadStatus(ThreadStatus* t) {
+            ScopedLock<Mutex> l(lock);
+            statuses.push_back(t);
+        }
+
+        void addHandle(shared_ptr h) {
+            ScopedLock<Mutex> l(lock);
+            std::for_each(statuses.begin(), statuses.end(), handleAdder(h));
+        }
+    };
+    
+    static AllThreadsStatuses allThreadsStatuses;
+};
+
+}}
+#endif // _sys_DeletionManager_h
diff --git a/cpp/src/qpid/sys/Dispatcher.cpp b/cpp/src/qpid/sys/Dispatcher.cpp
index be5d8a530c..c55f808b42 100644
--- a/cpp/src/qpid/sys/Dispatcher.cpp
+++ b/cpp/src/qpid/sys/Dispatcher.cpp
@@ -341,7 +341,7 @@ void DispatchHandle::doDelete() {
     }
     }
     // If we're not then do it right away
-    delete this;
+    deferDelete();
 }
 
 void DispatchHandle::dispatchCallbacks(Poller::EventType type) {
@@ -435,7 +435,7 @@ void DispatchHandle::dispatchCallbacks(Poller::EventType type) {
         break;
     }
     }      
-    delete this;
+    deferDelete();
 }
 
 }}
diff --git a/cpp/src/qpid/sys/Poller.h b/cpp/src/qpid/sys/Poller.h
index 843295e268..a7a12fdbe8 100644
--- a/cpp/src/qpid/sys/Poller.h
+++ b/cpp/src/qpid/sys/Poller.h
@@ -45,6 +45,12 @@ class PollerHandle {
 
 public:
     PollerHandle(const Socket& s);
+    
+    // Usual way to delete (will defer deletion until we
+    // can't be returned from a Poller::wait any more)
+    void deferDelete();
+    
+    // Class clients shouldn't ever use this
     virtual ~PollerHandle();
     
     const Socket& getSocket() const {return socket;}
diff --git a/cpp/src/qpid/sys/epoll/EpollPoller.cpp b/cpp/src/qpid/sys/epoll/EpollPoller.cpp
index 57acafd928..0e984d49ee 100644
--- a/cpp/src/qpid/sys/epoll/EpollPoller.cpp
+++ b/cpp/src/qpid/sys/epoll/EpollPoller.cpp
@@ -21,6 +21,7 @@
 
 #include "qpid/sys/Poller.h"
 #include "qpid/sys/Mutex.h"
+#include "qpid/sys/DeletionManager.h"
 #include "qpid/sys/posix/check.h"
 #include "qpid/sys/posix/PrivatePosix.h"
 
@@ -34,6 +35,13 @@
 namespace qpid {
 namespace sys {
 
+// Deletion manager to handle deferring deletion of PollerHandles to when they definitely aren't being used 
+DeletionManager<PollerHandle> PollerHandleDeletionManager;
+
+//  Instantiate (and define) class static for DeletionManager
+template <>
+DeletionManager<PollerHandle>::AllThreadsStatuses DeletionManager<PollerHandle>::allThreadsStatuses(0);
+
 class PollerHandlePrivate {
     friend class Poller;
     friend class PollerHandle;
@@ -98,6 +106,10 @@ PollerHandle::~PollerHandle() {
     delete impl;
 }
 
+void PollerHandle::deferDelete() {
+    PollerHandleDeletionManager.markForDeletion(this);
+}
+
 /**
  * Concrete implementation of Poller to use the Linux specific epoll
  * interface
@@ -239,9 +251,9 @@ void Poller::rearmFd(PollerHandle& handle) {
 }
 
 void Poller::shutdown() {
-	// Allow sloppy code to shut us down more than once
-	if (impl->isShutdown)
-		return;
+    // Allow sloppy code to shut us down more than once
+    if (impl->isShutdown)
+        return;
 
     // Don't use any locking here - isshutdown will be visible to all
     // after the epoll_ctl() anyway (it's a memory barrier)
@@ -261,6 +273,7 @@ Poller::Event Poller::wait(Duration timeout) {
 
     // Repeat until we weren't interupted
     do {
+        PollerHandleDeletionManager.markAllUnusedInThisThread();
         int rc = ::epoll_wait(impl->epollFd, &epe, 1, timeoutMs);
         
         if (impl->isShutdown) {