3
\$\begingroup\$

The locking mechanism is implemented in the class RW_Lock. The saved_async_completion_handler structure is used in RW_Lock to enqueue completion handlers that can't be called right away.

The rest of the code completes a test-scenario program for the lock. A group of threads run the main io_context where the lock and test logic reside. An independent thread runs another context used for printing to the console. Readers and writers (that do nothing but wait a certain amount of time) are posted periodically.

Motivation: Well, I had a multi threaded server with some data structures that were heavily read, but written every once in a while and I needed to lock them while writing. All I found was the usual mutex-based approach, which blocks threads waiting. I thought that would be bad for performance and implemented this version using Asio's strand. (Btw, if there already is a solution for that, could you point me at it?)

I think it works just fine. But synchronization is usually tricky to get right. Maybe I missed something? Is there anything considered bad practice? What can be improved?

The code and updates can be found in GitHub

#include <thread>
#include <vector>
#include <queue>
#include <asio.hpp>

// only for debug
#include <iostream>
#include <iomanip>
#include <string>



// The io_context used for printing
asio::io_context cout_io_ctx;





// Holds a completion handler, it's associated executor and a work_guard
// Provides constructors that automatically create the work guard
// The executor is obtained by calling `asio::get_associated_executor`
template <class Signature,
          class Executor = asio::any_completion_executor,
          class CompletionHandler = asio::any_completion_handler<Signature>>
struct saved_async_completion_handler
{
    using signature = Signature;
    using executor_type = Executor;
    using work_guard_type = asio::executor_work_guard<executor_type>;
    using completion_handler_type = CompletionHandler;


    saved_async_completion_handler(executor_type ex, completion_handler_type && handler)
        : m_ex(asio::get_associated_executor(handler, ex))
        , m_handler(std::forward<completion_handler_type>(handler))
        , m_work_guard(asio::make_work_guard(m_ex))
    {}

    template <class Handler>
    saved_async_completion_handler(Handler && handler)
        : m_ex(asio::get_associated_executor(handler))
        , m_handler(std::forward<Handler>(handler))
        , m_work_guard(asio::make_work_guard(m_ex))
    {}

    executor_type m_ex;
    completion_handler_type m_handler;
    work_guard_type m_work_guard;
};



template <asio::execution::executor Executor = asio::any_io_executor>
class RW_Lock : asio::noncopyable
{

public: // Type definitions
    using executor_type = Executor;

    // RAII
    struct Read_Lock : asio::noncopyable
    {
        Read_Lock(RW_Lock & rw_lock) : m_rw_lock(&rw_lock) { m_rw_lock->begin_read(); }
        Read_Lock(Read_Lock && other) : m_rw_lock(other.m_rw_lock) { other.m_rw_lock = nullptr; }
        ~Read_Lock(){ if (m_rw_lock) m_rw_lock->end_read(); }

        RW_Lock * m_rw_lock;
    };

    // RAII
    struct Write_Lock : asio::noncopyable
    {
        Write_Lock(RW_Lock & rw_lock) : m_rw_lock(&rw_lock) { m_rw_lock->begin_write(); }
        Write_Lock(Write_Lock && other) : m_rw_lock(other.m_rw_lock) { other.m_rw_lock = nullptr; }
        ~Write_Lock(){ if (m_rw_lock) m_rw_lock->end_write(); }

        RW_Lock * m_rw_lock;
    };

    // Completion signatures for the interface async functions
    using Read_Signature = void(Read_Lock);
    using Write_Signature = void(Write_Lock);


public: // Constructor
    RW_Lock(executor_type executor)
        : m_lock_strand(executor)
    { }


public: // Interface
    template <asio::completion_token_for<Read_Signature> CompletionToken>
    auto
    async_get_read_lock(CompletionToken && completion_token)
    {
        auto initiation = [&](asio::completion_handler_for<Read_Signature> auto && completion_handler)
        {
            auto f = [
                &,
                completion_handler = std::forward<decltype(completion_handler)>(completion_handler)
            ]() mutable
            {
                auto completion_executor = asio::get_associated_executor(completion_handler);
                static_assert(!std::same_as<decltype(completion_executor), asio::system_executor>, "system executor not allowed. Bind executor to your completion handler");            
            
                if (readers_allowed())
                {
                    asio::post(completion_executor,
                    [
                        completion_handler = std::forward<decltype(completion_handler)>(completion_handler),
                        rl = Read_Lock{*this}
                    ] () mutable
                    { completion_handler(std::move(rl)); });
                }
                else
                {
                    m_pending_readers.push( { completion_executor, std::forward<decltype(completion_handler)>(completion_handler) } );

                    print_debug_info("enqueue_reader");
                }
            };


            asio::dispatch(m_lock_strand, std::move(f));
        };


        return asio::async_initiate<CompletionToken, Read_Signature>
            (std::move(initiation), completion_token);
    }


    template <asio::completion_token_for<Write_Signature> CompletionToken>
    auto
    async_get_write_lock(CompletionToken && completion_token)
    {

        auto initiation = [&](asio::completion_handler_for<Write_Signature> auto && completion_handler)
        {

            auto f = [
                &,
                completion_handler = std::forward<decltype(completion_handler)>(completion_handler)
            ]() mutable
            {
                auto completion_executor = asio::get_associated_executor(completion_handler);
                static_assert(!std::same_as<decltype(completion_executor), asio::system_executor>, "system executor not allowed. Bind executor to your completion handler");
            
                if (writers_allowed())
                {
                    asio::post(completion_executor, 
                    [
                        completion_handler = std::forward<decltype(completion_handler)>(completion_handler),
                        wlock = Write_Lock{*this}
                    ] () mutable
                    { completion_handler(std::move(wlock)); });
                }
                else
                {
                    m_pending_writers.push( { completion_executor, std::forward<decltype(completion_handler)>(completion_handler) } );

                    print_debug_info("enqueue_writer");
                }
            };


            asio::dispatch(m_lock_strand, std::move(f));

        };


        return asio::async_initiate<CompletionToken, Write_Signature>
            (std::move(initiation), completion_token);
    }


public: // for debugging
    auto get_active_reader_count() const { return m_reader_count; }
    auto get_active_writer_count() const { return m_writer_count; }
    auto get_queued_reader_count() const { return m_pending_readers.size(); }
    auto get_queued_writer_count() const { return m_pending_writers.size(); }

    void print_debug_info(std::string tag)
    {
        // This function only gets called from the lock strand
        // but it dispatches the printing to a different context
        // with a dedicated thread.
        // 
        // This is done to so that printing does not affect (much)
        // the lock strand performance
        // 
        // It also means we must take copies of changing values
        // before dispatching to the printing thread


        // Get elapsed time since the first time this function is called
        static const auto start_time = std::chrono::steady_clock::now();
        auto current_time = std::chrono::steady_clock::now();
        auto elapsed = std::chrono::duration_cast<std::chrono::duration<double>>(current_time - start_time);

        // Copy lock counters
        auto ar = get_active_reader_count();
        auto aw = get_active_writer_count();
        auto qr = get_queued_reader_count();
        auto qw = get_queued_writer_count();

        // NOTE: lock counters are copied again here, but this being a debug feature,
        // the nicer looks and readability are preferred over inserting all function calls
        // into the lambda capture or moving all copies there.
        // Also, the compiler should be able to take care of that optimization by itself.
        asio::dispatch(cout_io_ctx, [=, tag = std::move(tag)](){
            std::cout << std::setw(16) << std::setprecision(10) << elapsed << "    "
                    << std::setw(16) << ("<" + tag + ">") << "\t"
                    << "active r/w: " << ar << " / " << aw << "\t"
                    << "queued r/w: " << qr << " / " << qw << std::endl;
        });
    }


private: // auxiliary functions
    // Determine if readers can run now or should be enqueued
    // Returns false if there are writers running or queued
    bool readers_allowed() const
    {
        return m_writer_count == 0 && m_pending_writers.empty();
    }

    // Determine if writers can run now or should be enqueued
    // Returns false if there are readers running
    bool writers_allowed() const
    {
        return m_reader_count == 0 && m_writer_count == 0;
    }


    // Dispatches pending writers and readers, in that order
    // Only dispatches readers/writers when allowed (functions above)
    void dispatch_pending()
    {
        while (writers_allowed() && !m_pending_writers.empty())
        {
            auto [ex, handler, guard] = std::move(m_pending_writers.front());
            m_pending_writers.pop();

            print_debug_info("dequeue_writer");

            asio::dispatch(ex, [h = std::move(handler), wl = Write_Lock{*this}]() mutable { h(std::move(wl)); });
            guard.reset();
        }

        while (readers_allowed() && !m_pending_readers.empty())
        {
            auto [ex, handler, guard] = std::move(m_pending_readers.front());
            m_pending_readers.pop();

            print_debug_info("dequeue_reader");

            asio::dispatch(ex, [h = std::move(handler), rl = Read_Lock{*this}]() mutable { h(std::move(rl)); });
            guard.reset();
        }
    }


    void begin_read()
    {
        // always called from the lock strand
        ++m_reader_count;
        
        print_debug_info("begin_read");
    }

    void begin_write()
    {
        // always called from the lock strand
        ++m_writer_count;

        print_debug_info("begin_write");
    }


    void end_read()
    {
        // this will be called without synchronization
        // dispatch to lock strand
        asio::dispatch(m_lock_strand, [&](){
            --m_reader_count;

            print_debug_info("end_read");

            dispatch_pending();
        });
    }

    void end_write()
    {
        // this will be called without synchronization
        // dispatch to lock strand
        asio::dispatch(m_lock_strand, [&](){
            --m_writer_count;

            print_debug_info("end_write");

            dispatch_pending();
        });
    }


private: // members
    asio::strand<executor_type> m_lock_strand;
    int m_reader_count{};
    int m_writer_count{};
    std::queue<saved_async_completion_handler<Read_Signature>> m_pending_readers;
    std::queue<saved_async_completion_handler<Write_Signature>> m_pending_writers;
};





// a fake reader that just waits
asio::awaitable<void> reader(asio::any_io_executor executor, RW_Lock<> & lock)
{
    auto rl = co_await lock.async_get_read_lock(asio::use_awaitable);

    // simulate busy reader that blocks the thread
    // std::this_thread::sleep_for(std::chrono::milliseconds{200});

    // simulate lightweight reader that doesn't hog the CPU
    asio::steady_timer t{executor};
    t.expires_after(std::chrono::milliseconds{200});
    co_await t.async_wait(asio::use_awaitable);
}

// a fake writer that just waits
asio::awaitable<void> writer(asio::any_io_executor executor, RW_Lock<> & lock)
{
    auto wl = co_await lock.async_get_write_lock(asio::use_awaitable);

    // simulate busy writer that blocks the thread
    // std::this_thread::sleep_for(std::chrono::milliseconds{750});

    // simulate lightweight writer that doesn't hog the CPU
    asio::steady_timer t{executor};
    t.expires_after(std::chrono::milliseconds{750});
    co_await t.async_wait(asio::use_awaitable);
}



// a coroutine that posts new readers periodically
asio::awaitable<void> post_readers(asio::any_io_executor executor, RW_Lock<> & lock)
{
    asio::steady_timer t{executor};

    while (true)
    {
        t.expires_after(std::chrono::milliseconds{50});
        co_await t.async_wait(asio::use_awaitable);
        asio::co_spawn(executor, reader(executor, lock), asio::detached);
    }
}

// a coroutine that posts new writers periodically
asio::awaitable<void> post_writers(asio::any_io_executor executor, RW_Lock<> & lock)
{
    asio::steady_timer t{executor};

    while (true)
    {
        t.cancel();
        t.expires_after(std::chrono::milliseconds{1500});
        co_await t.async_wait(asio::use_awaitable);
        asio::co_spawn(executor, writer(executor, lock), asio::detached);
    }
}



int main()
{
    asio::io_context io_ctx;

    RW_Lock<> lock{io_ctx.get_executor()};

    // spawn coroutines that periodically spawn readers and writers
    asio::co_spawn(io_ctx, post_readers(io_ctx.get_executor(), lock), asio::detached);
    asio::co_spawn(io_ctx, post_writers(io_ctx.get_executor(), lock), asio::detached);

    // prevent the printing thread from terminating when there is nothing to print
    auto cout_work_guard = asio::make_work_guard(cout_io_ctx);
    // start the printing thread
    std::jthread cout_thread{ [&](){ cout_io_ctx.run(); } };

    // run the main io_context in multiple threads
    std::vector<std::jthread> threads;
    threads.resize(3);
    for (auto & x : threads)
        x = std::jthread{ [&io_ctx](){ io_ctx.run(); } };
    io_ctx.run();

    cout_work_guard.reset();
}
\$\endgroup\$
2
  • \$\begingroup\$ As @G. Sliepen suggested, I tried using a simple mutex instead of a strand to synchronize access to the internal lock state (counters, queues). Benchmark showed that: 1) under contention the strand approach is slightly faster (7.5us vs 7.9us mean time, but >50% stddev) but I don't think the result is very valid. 2) When there's no contention, the mutex approach is 3 - 4 times faster. Tested on a laptop with processor i5-7200 (4 cores) \$\endgroup\$
    – DeltA
    Sep 18, 2023 at 5:11
  • \$\begingroup\$ I will upload the code and benchmark results to the linked GitHub page \$\endgroup\$
    – DeltA
    Sep 18, 2023 at 5:17

1 Answer 1

1
\$\begingroup\$

Thread safety issues

The classes Read_Lock and Write_Lock are public, and have public constructors. Someone could mistake this for being the public API, and think these classes can be used like std::shared_lock and std::unique_lock. If you do that, and if you have multiple threads calling io_ctx.run(), then you can have multiple threads calling member functions of your class RW_Lock. Then multiple threads could end up calling begin_read() at the same time. Even this line:

++m_reader_count;

Is not thread-safe: internally, it has to load the value of m_reader_count in a register, increment it, and then write it back to memory. Thus it is not done in an atomic way. To make operations on an int atomic, just use std::atomic<int>. However, m_pending_readers is much harder to make atomic. You can use a std::mutex to protect against multiple threads modifying the state of a RW_Lock object.

As you mentioned, this was not intentional, and the user should use async_get_read_lock() and async_get_write_lock() instead. You can enforce that by making the constructors of Read_Lock and Write_Lock private, and only give RW_Lock access to them.

Do you really need this?

One of the advantages of coroutines or an io_context that is only running on one thread is that you don't have thread safety issues, as only one function is executing at any given time. So you don't need any locks in those situations.

If you do have multiple threads running, then you can use a regular std::shared_mutex to get a read-write lock. The only drawback is that you shouldn't co_await while holding a lock, because depending on how the io_context schedules things, this might cause a deadlock. If you do need to await something while holding a lock, I see two options:

  1. Unlock the mutex while co_awaiting something. You of course have to assume something else can modify the shared resource you wanted to lock.
  2. Spawn a task on a separate, single-threaded context to do things that modify the shared resource.

Simplify the interface

Many of Boost Asio's functions that need an executor to work allow you to pass objects that have a get_executor() function, so you don't have to manually call that yourself. Consider making it so you can write:

asio::io_context io_ctx;

RW_Lock<> lock{io_ctx};

asio::co_spawn(io_ctx, post_readers(io_ctx, lock), asio::detached);
asio::co_spawn(io_ctx, post_writers(io_ctx, lock), asio::detached);

You might also want to add a get_executor() member function to class RW_Lock.

Prior art

There are non-Boost libraries that build on C++20's coroutines, some of them provide async reader-writer locks. For example, Josh Baldwin's libcoro has a coro::shared_mutex. The interface seems similar. You could look at its source code to see how the implementation is done.

Performance

One big issue with your implementation is that whenever you lock the RW_Lock, even if there is no contention, it will post a task to the m_lock_strand. That is not a trivial operation. If you have multiple threads running the strands, then I think this might also involve cross-thread synchronization to happen behind the scenes. Ideally, you only use m_lock_strand if there is contention.

Consider how std::mutex works: in the uncontended case, it is just a single atomic compare-exchange operation to lock the mutex. Only if this detects that the mutex was already locked, then a futex call is made to suspend the thread until some other thread unlocks the mutex. Using atomic operations in coroutines is safe, so you could take the same approach: have an atomic variable to keep track of whether the RW_Lock is locked, and use atomics to update this variable, and only in case of contention post a task to m_lock_strand to suspend the coroutine until another strand releases the lock.

\$\endgroup\$
6
  • \$\begingroup\$ About the thread safety issues you mention. begin_read() is a private function. It is only called from the constructor of the Read_Lock RAII wrapper which is only intended to run from the lock strand (only 1 thread can run it at a time). Anyway, thanks to your comment, I see there is a way for a user to manually construct a RAII wrapper and end-up calling begin_read() without synchronization (and bypassing the whole locking mechanism). I should make the RAII structures' constructors private. \$\endgroup\$
    – DeltA
    Aug 24, 2023 at 20:17
  • \$\begingroup\$ Yes, I thought Read_Lock and Write_Lock were the equivalent of std::shared_lock and std::unique_lock, and just assumed they were meant to be constructed directly. If you make it so they can only run on a single strand, then it's fine. \$\endgroup\$
    – G. Sliepen
    Aug 24, 2023 at 20:28
  • \$\begingroup\$ Do I really need this? Well, it depends. Most programs, I think, will work the same single-threaded as they do with multiple threads, just slower. As I stated on the question, the code is part of a server, and is expected to service several sessions concurrently. Using only one thread would keep most of the processor idle, especially with a high number of cores. It would be a waste of time and money. Also, while the extra performance may not be needed now, redesigning the whole server later to support multi-threading, I believe would be harder and error-prone. \$\endgroup\$
    – DeltA
    Aug 24, 2023 at 20:33
  • \$\begingroup\$ Performance: that's actually one of the main aspects I wanted some feedback about. You're right, there shouldn't even need to be a context switch if there is no contention. The only reason there is a m_lock_strand is to provide synchronization for the internal lock state (counters, etc); normally a mutex would be used. The idea behind is that waiting on a mutex halts the thread, while a strand does not, allowing other tasks to run while multiple co-routines are waiting for the lock. Perhaps adding an atomic check would be worth it? \$\endgroup\$
    – DeltA
    Sep 18, 2023 at 0:59
  • \$\begingroup\$ On the other hand, when there's no contention, everything should be faster anyway. Perhaps there's not too much to gain from optimizations there. And under contention it's best to do the posting anyway (?) ... I will try to benchmark it. \$\endgroup\$
    – DeltA
    Sep 18, 2023 at 1:04

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.