reader_writer_manager.hpp

Go to the documentation of this file.

// -----------------------------------------------------------------------------------------------------
// Copyright (c) 2006-2021, Knut Reinert & Freie Universität Berlin
// Copyright (c) 2016-2021, 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