ebr.hpp

Go to the documentation of this file.

/*-
 * Copyright (c) 2015-2018 Mindaugas Rasiukevicius <rmind at noxt eu>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */
 
// SPDX-License-Identifier: BSD-3-Clause
/* Copyright 2021, Intel Corporation */
 
#ifndef LIBPMEMOBJ_EBR_HPP
#define LIBPMEMOBJ_EBR_HPP
 
#include <atomic>
#include <cassert>
#include <functional>
#include <mutex>
#include <thread>
#include <unordered_map>
 
#include <libpmemobj++/detail/common.hpp>
 
namespace pmem
{
 
namespace detail
{
 
class ebr {
    using atomic = std::atomic<size_t>;
    using reference = std::reference_wrapper<atomic>;
 
public:
    class worker;
 
    ebr();
 
    worker register_worker();
    bool sync();
    void full_sync();
    size_t staging_epoch();
    size_t gc_epoch();
 
    class worker {
    public:
        worker(const worker &w) = delete;
        worker(worker &&w) = default;
        ~worker();
 
        worker &operator=(worker &w) = delete;
        worker &operator=(worker &&w) = default;
 
        template <typename F>
        void critical(F &&f);
 
    private:
        worker(ebr *e_, reference ref);
 
        reference local_epoch;
        ebr *e;
 
        friend ebr;
    };
 
private:
    static const size_t ACTIVE_FLAG = static_cast<size_t>(1)
        << (sizeof(size_t) * 8 - 1);
    static const size_t EPOCHS_NUMBER = 3;
 
    atomic global_epoch;
 
    std::unordered_map<std::thread::id, atomic> workers;
    std::mutex mtx;
};
 
ebr::ebr() : global_epoch(0)
{
#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
    VALGRIND_HG_DISABLE_CHECKING(&global_epoch, sizeof(global_epoch));
#endif
}
 
ebr::worker
ebr::register_worker()
{
    std::lock_guard<std::mutex> lock(mtx);
    auto res = workers.emplace(std::this_thread::get_id(), 0);
    if (!res.second) {
        throw std::runtime_error(
            "There can be only one worker per thread");
    }
 
    return worker{this, reference{res.first->second}};
}
 
bool
ebr::sync()
{
    auto current_epoch = global_epoch.load();
 
    std::lock_guard<std::mutex> lock(mtx);
    for (auto &w : workers) {
        LIBPMEMOBJ_CPP_ANNOTATE_HAPPENS_BEFORE(
            std::memory_order_seq_cst, &w.second);
        auto local_e = w.second.load();
        bool active = local_e & ACTIVE_FLAG;
        if (active && (local_e != (current_epoch | ACTIVE_FLAG))) {
            return false;
        }
    }
 
    LIBPMEMOBJ_CPP_ANNOTATE_HAPPENS_BEFORE(std::memory_order_seq_cst,
                           &global_epoch);
    global_epoch.store((current_epoch + 1) % EPOCHS_NUMBER);
 
    return true;
}
 
void
ebr::full_sync()
{
    size_t syncs_cnt = 0;
    while (true) {
        if (sync() && ++syncs_cnt == EPOCHS_NUMBER) {
            break;
        }
    }
}
 
size_t
ebr::staging_epoch()
{
    auto res = global_epoch.load();
    LIBPMEMOBJ_CPP_ANNOTATE_HAPPENS_AFTER(std::memory_order_seq_cst,
                          &global_epoch);
    return res;
}
 
size_t
ebr::gc_epoch()
{
    auto res = (global_epoch.load() + 1) % EPOCHS_NUMBER;
    LIBPMEMOBJ_CPP_ANNOTATE_HAPPENS_AFTER(std::memory_order_seq_cst,
                          &global_epoch);
    return res;
}
 
ebr::worker::worker(ebr *e_, reference ref) : local_epoch(ref), e(e_)
{
#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
    VALGRIND_HG_DISABLE_CHECKING(&ref.get(), sizeof(ref.get()));
#endif
}
 
ebr::worker::~worker()
{
    std::lock_guard<std::mutex> lock(e->mtx);
    e->workers.erase(std::this_thread::get_id());
}
 
template <typename F>
void
ebr::worker::critical(F &&f)
{
    auto new_epoch = e->global_epoch.load() | ACTIVE_FLAG;
    LIBPMEMOBJ_CPP_ANNOTATE_HAPPENS_AFTER(std::memory_order_seq_cst,
                          &(e->global_epoch));
 
    local_epoch.get().store(new_epoch);
    LIBPMEMOBJ_CPP_ANNOTATE_HAPPENS_AFTER(std::memory_order_seq_cst,
                          &local_epoch);
 
    f();
 
    local_epoch.get().store(0);
}
 
} /* namespace detail */
 
} /* namespace pmem */
 
#endif /* LIBPMEMOBJ_EBR_HPP */