svn commit: r920679 - in /incubator/thrift/trunk/lib/cpp/src/concurrency: Mutex.cpp Mutex.h ThreadManager.cpp

[dr...@apache.org]

Author: dreiss
Date: Tue Mar  9 05:19:59 2010
New Revision: 920679

URL: http://svn.apache.org/viewvc?rev=920679&view=rev
Log:
cpp: non-blocking add for ThreadManager

It's rare for the ThreadManager mutex to be contended, but it is
possible.  For nonblocking applications, it is necessary to have a
strict timeout for the lock acquisition.  With this change, that timeout
is enforced.  Also add timeout parameters to Mutex::lock and
Guard::Guard to support this feature.

Modified:
    incubator/thrift/trunk/lib/cpp/src/concurrency/Mutex.cpp
    incubator/thrift/trunk/lib/cpp/src/concurrency/Mutex.h
    incubator/thrift/trunk/lib/cpp/src/concurrency/ThreadManager.cpp

Modified: incubator/thrift/trunk/lib/cpp/src/concurrency/Mutex.cpp
URL: http://svn.apache.org/viewvc/incubator/thrift/trunk/lib/cpp/src/concurrency/Mutex.cpp?rev=920679&r1=920678&r2=920679&view=diff
==============================================================================
--- incubator/thrift/trunk/lib/cpp/src/concurrency/Mutex.cpp (original)
+++ incubator/thrift/trunk/lib/cpp/src/concurrency/Mutex.cpp Tue Mar  9 05:19:59 2010
@@ -18,6 +18,7 @@
  */
 
 #include "Mutex.h"
+#include "Util.h"
 
 #include <assert.h>
 #include <pthread.h>
@@ -50,6 +51,12 @@ class Mutex::impl {
 
   bool trylock() const { return (0 == pthread_mutex_trylock(&pthread_mutex_)); }
 
+  bool timedlock(int64_t milliseconds) const {
+    struct timespec ts;
+    Util::toTimespec(ts, milliseconds);
+    return (0 == pthread_mutex_timedlock(&pthread_mutex_, &ts));
+  }
+
   void unlock() const { pthread_mutex_unlock(&pthread_mutex_); }
 
   void* getUnderlyingImpl() const { return (void*) &pthread_mutex_; }
@@ -67,6 +74,8 @@ void Mutex::lock() const { impl_->lock()
 
 bool Mutex::trylock() const { return impl_->trylock(); }
 
+bool Mutex::timedlock(int64_t ms) const { return impl_->timedlock(ms); }
+
 void Mutex::unlock() const { impl_->unlock(); }
 
 void Mutex::DEFAULT_INITIALIZER(void* arg) {

Modified: incubator/thrift/trunk/lib/cpp/src/concurrency/Mutex.h
URL: http://svn.apache.org/viewvc/incubator/thrift/trunk/lib/cpp/src/concurrency/Mutex.h?rev=920679&r1=920678&r2=920679&view=diff
==============================================================================
--- incubator/thrift/trunk/lib/cpp/src/concurrency/Mutex.h (original)
+++ incubator/thrift/trunk/lib/cpp/src/concurrency/Mutex.h Tue Mar  9 05:19:59 2010
@@ -37,6 +37,7 @@ class Mutex {
   virtual ~Mutex() {}
   virtual void lock() const;
   virtual bool trylock() const;
+  virtual bool timedlock(int64_t milliseconds) const;
   virtual void unlock() const;
 
   void* getUnderlyingImpl() const;
@@ -75,18 +76,33 @@ private:
 
 class Guard {
  public:
-  Guard(const Mutex& value) : mutex_(value) {
-    mutex_.lock();
+  Guard(const Mutex& value, int64_t timeout = 0) : mutex_(&value) {
+    if (timeout == 0) {
+      value.lock();
+    } else if (timeout < 0) {
+      if (!value.trylock()) {
+        mutex_ = NULL;
+      }
+    } else {
+      if (!value.timedlock(timeout)) {
+        mutex_ = NULL;
+      }
+    }
   }
   ~Guard() {
-    mutex_.unlock();
+    if (mutex_) {
+      mutex_->unlock();
+    }
+  }
+
+  operator bool() const {
+    return (mutex_ != NULL);
   }
 
  private:
-  const Mutex& mutex_;
+  const Mutex* mutex_;
 };
 
-
 // Can be used as second argument to RWGuard to make code more readable
 // as to whether we're doing acquireRead() or acquireWrite().
 enum RWGuardType {

Modified: incubator/thrift/trunk/lib/cpp/src/concurrency/ThreadManager.cpp
URL: http://svn.apache.org/viewvc/incubator/thrift/trunk/lib/cpp/src/concurrency/ThreadManager.cpp?rev=920679&r1=920678&r2=920679&view=diff
==============================================================================
--- incubator/thrift/trunk/lib/cpp/src/concurrency/ThreadManager.cpp (original)
+++ incubator/thrift/trunk/lib/cpp/src/concurrency/ThreadManager.cpp Tue Mar  9 05:19:59 2010
@@ -461,7 +461,11 @@ void ThreadManager::Impl::removeWorker(s
   void ThreadManager::Impl::add(shared_ptr<Runnable> value,
                                 int64_t timeout,
                                 int64_t expiration) {
-    Guard g(mutex_);
+    Guard g(mutex_, timeout);
+
+    if (!g) {
+      throw TimedOutException();
+    }
 
     if (state_ != ThreadManager::STARTED) {
       throw IllegalStateException();