rcu.cpp

Go to the documentation of this file.

// Copyright (c) 2018-2019 The Bitcoin developers
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.

#include <rcu.h>

#include <sync.h>

#include <algorithm>
#include <chrono>
#include <condition_variable>

std::atomic<uint64_t> RCUInfos::revision{0};
thread_local RCUInfos RCUInfos::infos{};

static constexpr int RCU_ACTIVE_LOOP_COUNT = 10;

static std::atomic<RCUInfos *> threadInfos{nullptr};
static RecursiveMutex csThreadInfosDelete;

RCUInfos::RCUInfos() : state(0), next(nullptr) {
    RCUInfos *head = threadInfos.load();
    do {
        next.store(head);
    } while (!threadInfos.compare_exchange_weak(head, this));

    // Release the lock.
    readFree();
}

RCUInfos::~RCUInfos() {
    runCleanups();
    while (cleanups.size() > 0) {
        synchronize();
    }

    while (true) {
        LOCK(csThreadInfosDelete);

        std::atomic<RCUInfos *> *ptr;

        {
            RCULock lock(this);
            ptr = &threadInfos;
            while (true) {
                RCUInfos *current = ptr->load();
                if (current == this) {
                    break;
                }

                assert(current != nullptr);
                ptr = &current->next;
            }
        }

        RCUInfos *current = this;
        if (!ptr->compare_exchange_strong(current, next.load())) {
            continue;
        }

        synchronize();
        break;
    }
}

void RCUInfos::synchronize() {
    uint64_t syncRev = ++revision;

    // Loop a few time lock free.
    for (int i = 0; i < RCU_ACTIVE_LOOP_COUNT; i++) {
        runCleanups();
        if (cleanups.empty() && hasSyncedTo(syncRev)) {
            return;
        }
    }

    // It seems like we have some contention. Let's try to not starve the
    // system. Let's make sure threads that land here proceed one by one.
    // XXX: The best option long term is most likely to use a futex on one of
    // the thread causing synchronization delay so this thread can be waked up
    // at an appropriate time.
    static std::condition_variable cond;
    static Mutex cs;
    WAIT_LOCK(cs, lock);

    do {
        runCleanups();
        cond.notify_one();
    } while (!cond.wait_for(lock, std::chrono::microseconds(1), [&] {
        return cleanups.empty() && hasSyncedTo(syncRev);
    }));
}

void RCUInfos::runCleanups() {
    // By the time we run a set of cleanups, we may have more cleanups
    // available so we loop until there is nothing available for cleanup.
    while (true) {
        if (cleanups.empty()) {
            // There is nothing to cleanup.
            return;
        }

        auto it = cleanups.begin();
        uint64_t syncedTo = hasSyncedTo(it->first);
        while (it != cleanups.end() && it->first <= syncedTo) {
            // Run the cleanup and remove it from the map.
            it->second();
            cleanups.erase(it++);
        }
    }
}

uint64_t RCUInfos::hasSyncedTo(uint64_t cutoff) {
    uint64_t syncedTo = revision.load();

    // Go over the list and check all threads are past the synchronization
    // point.
    RCULock lock(this);
    RCUInfos *current = threadInfos.load();
    while (current != nullptr) {
        syncedTo = std::min(syncedTo, current->state.load());
        if (syncedTo < cutoff) {
            return 0;
        }

        current = current->next.load();
    }

    return syncedTo;
}