add support for rwlock

src/bthread/bthread.h

@@ -30,6 +30,7 @@
 #if defined(__cplusplus)
 #  include <iostream>
 #  include "bthread/mutex.h"        // use bthread_mutex_t in the RAII way
+#  include "bthread/rwlock.h"        // use bthread_rwlock_t in the RAII way
 #endif
 
 #include "bthread/id.h"

src/bthread/rwlock.cpp

@@ -0,0 +1,164 @@
+#include <pthread.h>
+#include <execinfo.h>
+#include <dlfcn.h> // dlsym
+#include <fcntl.h> // O_RDONLY
+#include "butil/atomicops.h"
+#include "bvar/bvar.h"
+#include "bvar/collector.h"
+#include "butil/macros.h" // BAIDU_CASSERT
+#include "butil/containers/flat_map.h"
+#include "butil/iobuf.h"
+#include "butil/fd_guard.h"
+#include "butil/files/file.h"
+#include "butil/files/file_path.h"
+#include "butil/file_util.h"
+#include "butil/unique_ptr.h"
+#include "butil/third_party/murmurhash3/murmurhash3.h"
+#include "butil/logging.h"
+#include "butil/object_pool.h"
+#include "bthread/butex.h"     // butex_*
+#include "bthread/processor.h" // cpu_relax, barrier
+#include "bthread/bthread.h"
+#include "bthread/sys_futex.h"
+#include "bthread/log.h"
+
+
+namespace bthread {
+
+inline int rwlock_unrlock(bthread_rwlock_t* rwlock) {
+    butil::atomic<unsigned>* whole = 
+        (butil::atomic<unsigned>*)rwlock->lock_flag;
+
+    while(1) {
+        unsigned r = whole->load();
+        if(r==0 || (r>>31) != 0) {
+            LOG(ERROR) << "wrong unrlock!";
+            return 0;
+        }
+        if(!(whole->compare_exchange_weak(r, r-1))) {
+            continue;
+        }
+        //wake up write waiter
+        bthread::butex_wake(whole);
+        return 0;
+    }
+
+}
+
+
+inline int rwlock_unwlock(bthread_rwlock_t* rwlock) {
+    butil::atomic<unsigned>* whole = 
+        (butil::atomic<unsigned>*)rwlock->lock_flag;
+
+    while(1) {
+        unsigned r = whole->load();
+        if(r != (unsigned)(1<<31) ) {
+            LOG(ERROR) << "wrong unwlock!";
+            return 0;
+        }
+        if(!whole->compare_exchange_weak(r, 0)) {
+            continue;
+        }
+        //wake up write waiter first
+        bthread::butex_wake(whole);
+        butil::atomic<unsigned>* w_wait_count = (butil::atomic<unsigned>*)rwlock->w_wait_count;
+        //try reduce wait_count for read waiters,and wake up read waiters
+        w_wait_count->fetch_sub(1);
+        bthread::butex_wake_all(w_wait_count);
+        return 0;
+    }
+
+}
+
+
+inline int rwlock_unlock(bthread_rwlock_t* rwlock) {
+    butil::atomic<unsigned>* whole = 
+        (butil::atomic<unsigned>*)rwlock->lock_flag;
+    if ((whole->load(butil::memory_order_relaxed) >> 31) != 0) {
+        return rwlock_unwlock(rwlock);
+    } else {
+        return rwlock_unrlock(rwlock);
+    }
+}
+
+inline int rwlock_rlock(bthread_rwlock_t* rwlock) {
+    butil::atomic<unsigned>* whole = 
+        (butil::atomic<unsigned>*)rwlock->lock_flag;
+
+    butil::atomic<unsigned>* w_wait_count = (butil::atomic<unsigned>*)rwlock->w_wait_count;
+    while (1) {
+        unsigned w = w_wait_count->load();
+        if( w > 0) {
+            if(bthread::butex_wait(w_wait_count, w, NULL) < 0 && 
+                errno != EWOULDBLOCK && errno != EINTR) {
+                return errno;
+            }
+            continue;
+        }
+        //FIXME!! we don't consider read_wait_count overflow yet,2^31 should be enough here
+        unsigned r = whole->load();
+        if((r >> 31) == 0) {
+            if(whole->compare_exchange_weak(r, r + 1)) {
+                return 0;
+            }
+        }
+    }
+
+}
+
+inline int rwlock_wlock(bthread_rwlock_t* rwlock) {
+    butil::atomic<unsigned>* w_wait_count = (butil::atomic<unsigned>*)rwlock->w_wait_count;
+    butil::atomic<unsigned>* whole = (butil::atomic<unsigned>*)rwlock->lock_flag;
+    //we don't consider w_wait_count overflow yet,2^32 should be enough here
+    w_wait_count->fetch_add(1);
+    while(1) {
+        unsigned r = whole->load();
+        if(r != 0) {
+            if(bthread::butex_wait(whole, r, NULL) < 0 && 
+                errno != EWOULDBLOCK && errno != EINTR) {
+                w_wait_count->fetch_sub(1);
+                bthread::butex_wake_all(w_wait_count);
+                return errno;
+            }
+            continue;
+        }
+        if(whole->compare_exchange_weak(r, (unsigned)(1<<31) )) {
+            return 0;
+        }
+    }
+}
+
+} // namespace bthread
+
+extern "C" {
+
+int bthread_rwlock_init(bthread_rwlock_t* __restrict rwlock,
+                        const bthread_rwlockattr_t* __restrict attr) {
+    rwlock->w_wait_count = bthread::butex_create_checked<unsigned>();
+    rwlock->lock_flag = 
+        bthread::butex_create_checked<unsigned>();
+    if (!rwlock->w_wait_count || !rwlock->lock_flag) {
+        LOG(ERROR) << "no memory";
+        return ENOMEM;
+    }
+    *rwlock->w_wait_count = 0;
+    *rwlock->lock_flag = 0;
+    return 0;
+}
+
+int bthread_rwlock_destroy(bthread_rwlock_t* rwlock) {
+    bthread::butex_destroy(rwlock->w_wait_count);
+    bthread::butex_destroy(rwlock->lock_flag);
+    return 0;
+}
+
+int bthread_rwlock_rdlock(bthread_rwlock_t* rwlock) { return bthread::rwlock_rlock(rwlock); }
+
+int bthread_rwlock_wrlock(bthread_rwlock_t* rwlock) { return bthread::rwlock_wlock(rwlock); }
+
+int bthread_rwlock_unrlock(bthread_rwlock_t* rwlock) { return bthread::rwlock_unrlock(rwlock); }
+
+int bthread_rwlock_unwlock(bthread_rwlock_t* rwlock) { return bthread::rwlock_unwlock(rwlock); }
+
+int bthread_rwlock_unlock(bthread_rwlock_t* rwlock) { return bthread::rwlock_unlock(rwlock); }
+}

src/bthread/rwlock.h

@@ -0,0 +1,130 @@
+#ifndef BTHREAD_RW_MUTEX_H
+#define BTHREAD_RW_MUTEX_H
+
+#include "bthread/types.h"
+#include "butil/scoped_lock.h"
+#include "bvar/utils/lock_timer.h"
+#include "bthread/bthread.h"
+
+__BEGIN_DECLS
+// -------------------------------------------
+// Functions for handling read-write locks.
+// -------------------------------------------
+
+// Initialize read-write lock `rwlock' using attributes `attr', or use
+// the default values if later is NULL.
+extern int bthread_rwlock_init(bthread_rwlock_t* __restrict rwlock,
+                               const bthread_rwlockattr_t* __restrict attr);
+
+// Destroy read-write lock `rwlock'.
+extern int bthread_rwlock_destroy(bthread_rwlock_t* rwlock);
+
+// Acquire read lock for `rwlock'.
+extern int bthread_rwlock_rdlock(bthread_rwlock_t* rwlock);
+
+// Try to acquire read lock for `rwlock'.
+extern int bthread_rwlock_tryrdlock(bthread_rwlock_t* rwlock);
+
+// Try to acquire read lock for `rwlock' or return after specfied time.
+extern int bthread_rwlock_timedrdlock(bthread_rwlock_t* __restrict rwlock,
+                                      const struct timespec* __restrict abstime);
+
+// Acquire write lock for `rwlock'.
+extern int bthread_rwlock_wrlock(bthread_rwlock_t* rwlock);
+
+// Try to acquire write lock for `rwlock'.
+extern int bthread_rwlock_trywrlock(bthread_rwlock_t* rwlock);
+
+// Try to acquire write lock for `rwlock' or return after specfied time.
+extern int bthread_rwlock_timedwrlock(bthread_rwlock_t* __restrict rwlock,
+                                      const struct timespec* __restrict abstime);
+
+// Unlock `rwlock'.
+extern int bthread_rwlock_unlock(bthread_rwlock_t* rwlock);
+
+
+// ---------------------------------------------------
+// Functions for handling read-write lock attributes.
+// ---------------------------------------------------
+
+// Initialize attribute object `attr' with default values.
+extern int bthread_rwlockattr_init(bthread_rwlockattr_t* attr);
+
+// Destroy attribute object `attr'.
+extern int bthread_rwlockattr_destroy(bthread_rwlockattr_t* attr);
+
+// Return current setting of reader/writer preference.
+extern int bthread_rwlockattr_getkind_np(const bthread_rwlockattr_t* attr, int* pref);
+
+// Set reader/write preference.
+extern int bthread_rwlockattr_setkind_np(bthread_rwlockattr_t* attr, int pref);
+__END_DECLS
+
+
+// Specialize std::lock_guard and std::unique_lock for bthread_rwlock_t
+
+namespace bthread {
+
+class wlock_guard {
+public:
+    explicit wlock_guard(bthread_rwlock_t& mutex) : _pmutex(&mutex) {
+#if !defined(NDEBUG)
+        const int rc = bthread_rwlock_wrlock(_pmutex);
+        if (rc) {
+            LOG(FATAL) << "Fail to lock bthread_rwlock_t=" << _pmutex << ", " << berror(rc);
+            _pmutex = NULL;
+        }
+#else
+        bthread_rwlock_wrlock(_pmutex);
+#endif // NDEBUG
+    }
+
+    ~wlock_guard() {
+#ifndef NDEBUG
+        if (_pmutex) {
+            bthread_rwlock_unlock(_pmutex);
+        }
+#else
+        bthread_rwlock_unlock(_pmutex);
+#endif
+    }
+
+private:
+    DISALLOW_COPY_AND_ASSIGN(wlock_guard);
+    bthread_rwlock_t* _pmutex;
+};
+
+class rlock_guard {
+public:
+    explicit rlock_guard(bthread_rwlock_t& mutex) : _pmutex(&mutex) {
+#if !defined(NDEBUG)
+        const int rc = bthread_rwlock_rdlock(_pmutex);
+        if (rc) {
+            LOG(FATAL) << "Fail to lock bthread_rwlock_t=" << _pmutex << ", " << berror(rc);
+            _pmutex = NULL;
+        }
+#else
+        bthread_rwlock_rdlock(_pmutex);
+#endif // NDEBUG
+    }
+
+    ~rlock_guard() {
+#ifndef NDEBUG
+        if (_pmutex) {
+            bthread_rwlock_unlock(_pmutex);
+        }
+#else
+        bthread_rwlock_unlock(_pmutex);
+#endif
+    }
+
+private:
+    DISALLOW_COPY_AND_ASSIGN(rlock_guard);
+    bthread_rwlock_t* _pmutex;
+};
+
+
+} // namespace bthread
+
+
+#endif

src/bthread/types.h

@@ -173,6 +173,8 @@ typedef struct {
 } bthread_condattr_t;
 
 typedef struct {
+    unsigned* w_wait_count;   //include the bthread who holding wlock yet
+    unsigned* lock_flag;     //highest bit 1 for wlocked, low 31 bit for read lock holding count, 0 for unlocked
 } bthread_rwlock_t;
 
 typedef struct {

test/BUILD

@@ -235,6 +235,7 @@ cc_test(
         "bthread_dispatcher_unittest.cpp",
         "bthread_fd_unittest.cpp",
         "bthread_mutex_unittest.cpp",
+        "bthread_brpc_rwlock_unittest.cpp",
         "bthread_setconcurrency_unittest.cpp",
         # glog CHECK die with a fatal error
         "bthread_key_unittest.cpp"