qrwlock as quick rwlock without starvation prevention but better perfomance for read-heavy workloads

Author: Coldwings

thread/test/test.cpp

@@ -1891,6 +1891,192 @@ TEST(future, test2) {
     EXPECT_EQ(v, PROMISE_VALUE);
 }
 
+TEST(qrwlock, BasicLockUnlock) {
+    qrwlock mtx;
+
+    // Test exclusive lock
+    EXPECT_EQ(0, mtx.lock(WLOCK));
+    mtx.unlock();
+
+    // Test shared lock
+    EXPECT_EQ(0, mtx.lock(RLOCK));
+    mtx.unlock();
+}
+
+TEST(qrwlock, TryLock) {
+    qrwlock mtx;
+
+    // Try lock should succeed when unlocked
+    EXPECT_TRUE(mtx.try_lock(WLOCK) == 0);
+    // Try lock should fail when write-locked
+    EXPECT_FALSE(mtx.try_lock(WLOCK) == 0);
+    EXPECT_FALSE(mtx.try_lock(RLOCK) == 0);
+    mtx.unlock();
+
+    // Try shared lock should succeed when unlocked
+    EXPECT_TRUE(mtx.try_lock(RLOCK) == 0);
+    // Additional shared locks should succeed
+    EXPECT_TRUE(mtx.try_lock(RLOCK) == 0);
+    // Write lock should fail when read-locked
+    EXPECT_FALSE(mtx.try_lock(WLOCK) == 0);
+    mtx.unlock();
+    mtx.unlock();
+}
+
+TEST(qrwlock, MultipleReaders) {
+    qrwlock mtx;
+    constexpr int NUM_READERS = 100;
+
+    // Acquire many shared locks
+    for (int i = 0; i < NUM_READERS; i++) {
+        EXPECT_TRUE(mtx.try_lock(RLOCK) == 0);
+    }
+
+    // Write lock should fail
+    EXPECT_FALSE(mtx.try_lock(WLOCK) == 0);
+
+    // Release all shared locks
+    for (int i = 0; i < NUM_READERS; i++) {
+        mtx.unlock();
+    }
+
+    // Now write lock should succeed
+    EXPECT_TRUE(mtx.try_lock(WLOCK) == 0);
+    mtx.unlock();
+}
+
+TEST(qrwlock, ConcurrentReaders) {
+    qrwlock mtx;
+    std::atomic<int> active_readers{0};
+    std::atomic<int> max_concurrent{0};
+    std::atomic<uint64_t> total_ops{0};
+    constexpr int NUM_THREADS = 32;
+    constexpr int OPS_PER_THREAD = 1000;
+
+    std::vector<photon::join_handle*> handles;
+    for (int i = 0; i < NUM_THREADS; i++) {
+        handles.emplace_back(photon::thread_enable_join(
+            photon::thread_create11([&]() {
+                for (int j = 0; j < OPS_PER_THREAD; j++) {
+                    mtx.lock(RLOCK);
+                    int cur = active_readers.fetch_add(1, std::memory_order_relaxed) + 1;
+                    // Track max concurrent readers
+                    int prev_max = max_concurrent.load(std::memory_order_relaxed);
+                    while (cur > prev_max &&
+                           !max_concurrent.compare_exchange_weak(prev_max, cur)) {}
+                    photon::thread_yield();
+                    active_readers.fetch_sub(1, std::memory_order_relaxed);
+                    mtx.unlock();
+                    total_ops.fetch_add(1, std::memory_order_relaxed);
+                }
+            })));
+    }
+
+    for (auto& h : handles) {
+        photon::thread_join(h);
+    }
+
+    EXPECT_EQ(0, active_readers.load());
+    EXPECT_GT(max_concurrent.load(), 1);  // Should have concurrent readers
+    EXPECT_EQ(NUM_THREADS * OPS_PER_THREAD, total_ops.load());
+    LOG_INFO("Max concurrent readers: `", max_concurrent.load());
+}
+
+TEST(qrwlock, ReaderWriterExclusion) {
+    qrwlock mtx;
+    std::atomic<bool> writing{false};
+    std::atomic<int> active_readers{0};
+    std::atomic<uint64_t> read_ops{0};
+    std::atomic<uint64_t> write_ops{0};
+    constexpr int NUM_READERS = 16;
+    constexpr int NUM_WRITERS = 4;
+    constexpr int OPS_PER_THREAD = 500;
+
+    std::vector<photon::join_handle*> handles;
+
+    // Reader threads
+    for (int i = 0; i < NUM_READERS; i++) {
+        handles.emplace_back(photon::thread_enable_join(
+            photon::thread_create11([&]() {
+                for (int j = 0; j < OPS_PER_THREAD; j++) {
+                    mtx.lock(RLOCK);
+                    active_readers.fetch_add(1, std::memory_order_relaxed);
+                    EXPECT_FALSE(writing.load(std::memory_order_acquire));
+                    photon::thread_yield();
+                    EXPECT_FALSE(writing.load(std::memory_order_acquire));
+                    active_readers.fetch_sub(1, std::memory_order_relaxed);
+                    mtx.unlock();
+                    read_ops.fetch_add(1, std::memory_order_relaxed);
+                }
+            })));
+    }
+
+    // Writer threads
+    for (int i = 0; i < NUM_WRITERS; i++) {
+        handles.emplace_back(photon::thread_enable_join(
+            photon::thread_create11([&]() {
+                for (int j = 0; j < OPS_PER_THREAD; j++) {
+                    mtx.lock(WLOCK);
+                    EXPECT_EQ(0, active_readers.load(std::memory_order_acquire));
+                    writing.store(true, std::memory_order_release);
+                    photon::thread_yield();
+                    EXPECT_EQ(0, active_readers.load(std::memory_order_acquire));
+                    writing.store(false, std::memory_order_release);
+                    mtx.unlock();
+                    write_ops.fetch_add(1, std::memory_order_relaxed);
+                }
+            })));
+    }
+
+    for (auto& h : handles) {
+        photon::thread_join(h);
+    }
+
+    EXPECT_EQ(NUM_READERS * OPS_PER_THREAD, read_ops.load());
+    EXPECT_EQ(NUM_WRITERS * OPS_PER_THREAD, write_ops.load());
+    LOG_INFO("Read ops: `, Write ops: `", read_ops.load(), write_ops.load());
+}
+
+TEST(qrwlock, LockTimeout) {
+    qrwlock mtx;
+
+    // Acquire write lock in main coroutine
+    EXPECT_EQ(0, mtx.lock(WLOCK));
+
+    // Try to acquire read lock with timeout in another coroutine
+    auto jh = photon::thread_enable_join(
+        photon::thread_create11([&]() {
+            EXPECT_EQ(-1, mtx.lock(RLOCK, 100 * 1000));
+            EXPECT_EQ(ETIMEDOUT, errno);
+        }));
+
+    // Must yield to let the waiter run and timeout
+    photon::thread_usleep(200 * 1000);  // Sleep longer than timeout
+
+    photon::thread_join(jh);
+    mtx.unlock();
+}
+
+TEST(qrwlock, WriteLockTimeout) {
+    qrwlock mtx;
+
+    // Acquire read lock in main coroutine
+    EXPECT_EQ(0, mtx.lock(RLOCK));
+
+    // Try to acquire write lock with timeout in another coroutine
+    auto jh = photon::thread_enable_join(
+        photon::thread_create11([&]() {
+            EXPECT_EQ(-1, mtx.lock(WLOCK, 100 * 1000));
+            EXPECT_EQ(ETIMEDOUT, errno);
+        }));
+
+    // Must yield to let the waiter run and timeout
+    photon::thread_usleep(200 * 1000);  // Sleep longer than timeout
+
+    photon::thread_join(jh);
+    mtx.unlock();
+}
+
 int main(int argc, char** arg)
 {
     if (!photon::is_using_default_engine()) return 0;

thread/thread.h

@@ -549,12 +549,16 @@ namespace photon
     class rwlock
     {
     public:
+        int try_lock(int mode);
         int lock(int mode, Timeout timeout = {});
         int unlock();
     protected:
-        int64_t state = 0;
+        std::atomic<int64_t> state{0};
         condition_variable cvar;
-        mutex mtx;
+        condition_variable cvar_uniq;
+        spinlock mtx;
+        constexpr static int64_t MAX_SHARED_LOCK_COUNT = 1 << 16;
+        constexpr static int64_t WRITE_LOCKED = -1;
     };
 
     class scoped_rwlock
@@ -589,6 +593,118 @@ namespace photon
         bool m_locked;
     };
 
+    // High-performance rwlock without starvation prevention.
+    // Optimized for read-heavy workloads using lock-free atomic operations
+    // for the read lock fast path.
+    class qrwlock {
+    protected:
+        constexpr static int64_t MAX_SHARED_LOCK_COUNT = 1 << 16;
+        constexpr static int64_t WRITE_LOCKED = -1;
+
+        // State encoding:
+        //   state > 0  : number of active readers
+        //   state == 0 : unlocked
+        //   state == -1: write-locked
+
+        std::atomic<int64_t> lock_state{0};
+        photon::condition_variable cv_shared;
+        photon::condition_variable cv_unique;
+        photon::spinlock spin;
+
+        // Must be called with spinlock held.
+        void try_wake() {
+            if (!cv_unique.notify_one()) {
+                cv_shared.notify_all();
+            }
+        }
+
+        // Common blocking lock pattern: fast-path try, then slow-path wait
+        // loop. try_fn must be safe to call both with and without spinlock
+        // held.
+        template <typename TryFunc>
+        int do_lock(TryFunc&& try_fn, photon::condition_variable& cv,
+                    Timeout timeout) {
+            if (try_fn()) return 0;
+            SCOPED_LOCK(spin);
+            while (true) {
+                if (try_fn()) return 0;
+                int ret = cv.wait(spin, timeout);
+                if (ret < 0) {
+                    return -1;
+                }
+            }
+        }
+
+        void __unlock_unique() {
+            SCOPED_LOCK(spin);
+            assert(lock_state.load(std::memory_order_relaxed) == WRITE_LOCKED);
+            lock_state.store(0, std::memory_order_release);
+            try_wake();
+        }
+
+        void __unlock_shared() {
+            auto prev = lock_state.fetch_sub(1, std::memory_order_acq_rel);
+            assert(prev > 0);  // prev<=0 indicates misuse
+            if (prev == 1) {
+                SCOPED_LOCK(spin);
+                try_wake();
+            }
+        }
+
+        bool __trylock() {
+            int64_t expected = 0;
+            return lock_state.compare_exchange_strong(
+                expected, WRITE_LOCKED, std::memory_order_acq_rel,
+                std::memory_order_relaxed);
+        }
+
+        bool __trylock_shared() {
+            auto state = lock_state.load(std::memory_order_acquire);
+            while (state >= 0 && state < MAX_SHARED_LOCK_COUNT) {
+                if (lock_state.compare_exchange_weak(state, state + 1,
+                                                     std::memory_order_acq_rel,
+                                                     std::memory_order_acquire))
+                    return true;
+            }
+            return false;
+        }
+
+    public:
+        qrwlock() = default;
+        qrwlock(const qrwlock&) = delete;
+        qrwlock& operator=(const qrwlock&) = delete;
+
+        int try_lock(int mode) {
+            if (mode == WLOCK)
+                return __trylock() ? 0 : -1;
+            else
+                return __trylock_shared() ? 0 : -1;
+        }
+
+        int lock(int mode, Timeout timeout = {}) {
+            if (mode == WLOCK)
+                return do_lock([this] { return __trylock(); }, cv_unique,
+                               timeout);
+            else
+                return do_lock([this] { return __trylock_shared(); }, cv_shared,
+                               timeout);
+        }
+
+        int unlock() {
+            auto cur_state = lock_state.load(std::memory_order_acquire);
+            if (cur_state == 0) {
+                errno = ENOLCK;
+                return -1;
+            }
+            if (cur_state == WRITE_LOCKED) {
+                __unlock_unique();
+            } else {
+                __unlock_shared();
+            }
+            return 0;
+        }
+    };
+
     // create `n` threads to run `start(arg)`, then get joined
     void threads_create_join(uint64_t n, thread_entry start, void* arg,
                           uint64_t stack_size = DEFAULT_STACK_SIZE);