reader_writer_manager.hpp

Go to the documentation of this file.

// -----------------------------------------------------------------------------------------------------
// Copyright (c) 2006-2022, Knut Reinert & Freie Universität Berlin
// Copyright (c) 2016-2022, Knut Reinert & MPI für molekulare Genetik
// This file may be used, modified and/or redistributed under the terms of the 3-clause BSD-License
// shipped with this file and also available at: https://github.com/seqan/seqan3/blob/master/LICENSE.md
// -----------------------------------------------------------------------------------------------------
 
#pragma once
 
#include <cassert>
#include <functional>
#include <mutex>
#include <seqan3/std/new>
 
#include <seqan3/core/detail/strong_type.hpp>
#include <seqan3/utility/parallel/detail/latch.hpp>
#include <seqan3/utility/parallel/detail/spin_delay.hpp>
 
namespace seqan3::detail
{
 
struct writer_count : public detail::strong_type<size_t, writer_count>
{
    using base_t = detail::strong_type<size_t, writer_count>;
 
    using base_t::base_t;
};
 
struct reader_count : public detail::strong_type<size_t, reader_count>
{
    using base_t = detail::strong_type<size_t, reader_count>;
 
    using base_t::base_t;
};
 
class reader_writer_manager
{
private:
    class [[nodiscard]] scoped_writer_type
    {
    public:
        scoped_writer_type() = delete;                                        
        scoped_writer_type(scoped_writer_type const &) = default;             
        scoped_writer_type(scoped_writer_type &&) = default;                  
        scoped_writer_type & operator=(scoped_writer_type const &) = default; 
        scoped_writer_type & operator=(scoped_writer_type &&) = default;      
 
        explicit scoped_writer_type(reader_writer_manager & _manager) : manager{_manager}
        {}
 
        ~scoped_writer_type()
        {
            manager.writer_arrive();
        }
 
        reader_writer_manager & manager;
    };
 
    class [[nodiscard]] scoped_reader_type
    {
    public:
        scoped_reader_type() = delete;                                        
        scoped_reader_type(scoped_reader_type const &) = default;             
        scoped_reader_type(scoped_reader_type &&) = default;                  
        scoped_reader_type & operator=(scoped_reader_type const &) = default; 
        scoped_reader_type & operator=(scoped_reader_type &&) = default;      
 
        explicit scoped_reader_type(reader_writer_manager & _manager) : manager{_manager}
        {}
 
        ~scoped_reader_type()
        {
            manager.reader_arrive();
        }
 
        reader_writer_manager & manager;
    };
 
public:
    reader_writer_manager() = delete;                                          
    reader_writer_manager(reader_writer_manager const &) = delete;             
    reader_writer_manager(reader_writer_manager &&) = delete;                  
    reader_writer_manager & operator=(reader_writer_manager const &) = delete; 
    reader_writer_manager & operator=(reader_writer_manager &&) = delete;      
    ~reader_writer_manager() = default;                                        
 
    template <typename concurrent_t>
        requires requires { std::declval<concurrent_t>().close(); } // requires closable concurrent data structure.
    reader_writer_manager(reader_count const rcount, writer_count const wcount, concurrent_t & ds) :
        reader_latch{static_cast<ptrdiff_t>(rcount.get())},
        writer_latch{static_cast<ptrdiff_t>(wcount.get())},
        completion_fn{[&ds]()
                      {
                          ds.close();
                      }}
    {
        if (rcount.get() < 1 || wcount.get() < 1)
            throw std::invalid_argument{"Both, reader count and writer count must be at least 1."};
    }
 
    void writer_arrive_and_wait() noexcept
    {
        writer_latch.arrive_and_wait();
 
        std::call_once(flag, completion_fn);
    }
 
    void writer_arrive() noexcept
    {
        writer_latch.arrive();
 
        if (writer_latch.try_wait())
            std::call_once(flag, completion_fn);
    }
 
    void reader_arrive_and_wait() noexcept
    {
        reader_latch.arrive_and_wait();
    }
 
    void reader_arrive() noexcept
    {
        reader_latch.arrive();
    }
 
    scoped_writer_type register_writer() noexcept
    {
        return scoped_writer_type{*this};
    }
 
    scoped_reader_type register_reader() noexcept
    {
        return scoped_reader_type{*this};
    }
 
private:
    alignas(std::hardware_destructive_interference_size) latch reader_latch;
    alignas(std::hardware_destructive_interference_size) latch writer_latch;
    alignas(std::hardware_destructive_interference_size) std::once_flag flag;
    std::function<void()> completion_fn;
};
 
} // namespace seqan3::detail