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);
    }));
}
 
class RCUInfos::RCUCleanupGuard {
    RCUInfos *infos;
 
public:
    explicit RCUCleanupGuard(RCUInfos *infosIn) : infos(infosIn) {
        infos->isCleaningUp = true;
    }
 
    ~RCUCleanupGuard() { infos->isCleaningUp = false; }
};
 
void RCUInfos::runCleanups() {
    if (isCleaningUp || cleanups.empty()) {
        // We don't want to run cleanups within cleanups.
        return;
    }
 
    RCUCleanupGuard guard(this);
 
    // 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 (!cleanups.empty()) {
        auto it = cleanups.begin();
        uint64_t syncedTo = hasSyncedTo(it->first);
        if (it->first > syncedTo) {
            // We have nothing more ready to be cleaned up.
            return;
        }
 
        while (it != cleanups.end() && it->first <= syncedTo) {
            // Run the cleanup and remove it from the map.
            auto fun = std::move(it->second);
            cleanups.erase(it++);
            fun();
        }
    }
}
 
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;
}