diff --git a/CMakeLists.txt b/CMakeLists.txt index 13ce998a4d7b..b9ba86f6ac6e 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -299,6 +299,12 @@ add_ccf_static_library( ${CCF_DIR}/src/kv/untyped_map_diff.cpp LINK_LIBS ccf_threading ) +# Enforced (see kv/store.h, consensus/aft/raft.h, consensus/aft/impl/state.h, +# node/history.h) so that ccf_kv's compiled objects can safely be linked, +# unmodified, by every test target - including one that recompiles those +# headers with a different ccf::pal::Mutex (see +# src/commit_concurrency/interleaving_lock_override.h) - without an ODR violation. +target_compile_definitions(ccf_kv PRIVATE CCF_STATIC_LIBRARY_BUILD) # CCF endpoints lib add_ccf_static_library( @@ -338,6 +344,8 @@ add_ccf_static_library( ${CCF_DIR}/src/tasks/worker.cpp LINK_LIBS ccf_threading ) +# See the comment on ccf_kv's own CCF_STATIC_LIBRARY_BUILD above. +target_compile_definitions(ccf_tasks PRIVATE CCF_STATIC_LIBRARY_BUILD) find_library(BACKTRACE_LIBRARY backtrace) if(NOT BACKTRACE_LIBRARY) @@ -720,6 +728,91 @@ if(BUILD_TESTS) ) target_link_libraries(raft_test PRIVATE ccfcrypto ccf_tasks) + # Combines a real ccf::kv::Store, a real aft::Aft (raft consensus), and a + # real ccf::MerkleTxHistory under real OS-thread concurrency - the three + # components production code relies on together, but which no other unit + # test suite exercises jointly (kv_test stubs consensus, raft_test stubs + # the store, history_test stubs consensus). DETECT_DEADLOCKS is passed + # because the interleaving primitive itself (src/commit_concurrency/interleaving.h) + # could deadlock if buggy. + add_unit_test( + commit_concurrency_test + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/interleaving_test.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/threaded/main.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/threaded/smoke.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/threaded/deterministic.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/threaded/fuzzer.cpp + DETECT_DEADLOCKS + ) + set_property( + TEST commit_concurrency_test + APPEND + PROPERTY LABELS concurrency + ) + target_link_libraries( + commit_concurrency_test + PRIVATE ccfcrypto http_parser ccf_kv ccf_tasks + ) + + # Explores every legal interleaving of a bounded scenario (rather than + # sampling timing-dependent ones, as commit_concurrency_test does) + # via ccf::kv::test::explore_all_interleavings() in + # src/commit_concurrency/deterministic_scheduler.h. DETECT_DEADLOCKS is passed for + # the same reason as above. + add_unit_test( + commit_concurrency_model_test + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/deterministic_scheduler_test.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/model_checked/main.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/model_checked/rejected_commit_stall.cpp + # ccf::tasks' own sources (normally built once into ccf_tasks and + # shared unmodified - see kv_test's use of ccf_kv, for example) are + # rebuilt here instead of linking ccf_tasks, so that they see the + # same -include below as everything else in this target: ccf::tasks + # keeps a process-wide job board (a real ccf::pal::Mutex user) that + # outlives any single explored schedule, so every thread that can + # reach it - including any of ccf::tasks' own internals - needs the + # same scheduler-aware lock for DriverRegistration (see + # deterministic_scheduler.h) to keep it consistent across schedules. + ${CCF_DIR}/src/tasks/task_system.cpp + ${CCF_DIR}/src/tasks/job_board.cpp + ${CCF_DIR}/src/tasks/ordered_tasks.cpp + ${CCF_DIR}/src/tasks/fan_in_tasks.cpp + ${CCF_DIR}/src/tasks/thread_manager.cpp + ${CCF_DIR}/src/tasks/worker.cpp + DETECT_DEADLOCKS + ) + set_property( + TEST commit_concurrency_model_test + APPEND + PROPERTY LABELS concurrency + ) + # The -include flag makes every source file in this target (and only + # this target) see ccf::pal::Mutex itself resolve to SchedulerMutex - + # see src/commit_concurrency/interleaving_lock_override.h. + target_compile_options( + commit_concurrency_model_test + PRIVATE + -include + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/interleaving_lock_override.h + ) + # ccf_kv and ccfcrypto are safe to share, unmodified, with every other + # test target here despite the -include above: none of their own + # sources include store.h, raft.h, impl/state.h, or history.h, and + # each of those four headers refuses to compile at all into either of + # them (CCF_STATIC_LIBRARY_BUILD, set on both below), so this stops + # being true loudly, at build time, rather than silently. ccf_tasks is + # deliberately not linked here - see the comment on its sources above. + target_link_libraries( + commit_concurrency_model_test + PRIVATE + ccfcrypto + http_parser + ccf_kv + ccf_threading + ${CMAKE_DL_LIBS} + ${BACKTRACE_LIBRARY} + ) + add_unit_test( raft_enclave_test ${CMAKE_CURRENT_SOURCE_DIR}/src/consensus/aft/test/enclave.cpp diff --git a/cmake/crypto.cmake b/cmake/crypto.cmake index 0efcee0742dd..b7407aa41164 100644 --- a/cmake/crypto.cmake +++ b/cmake/crypto.cmake @@ -35,6 +35,8 @@ find_library(TLS_LIBRARY ssl) add_library(ccfcrypto STATIC ${CCFCRYPTO_SRC}) add_warning_checks(ccfcrypto) +# See the comment on ccf_kv's own CCF_STATIC_LIBRARY_BUILD in CMakeLists.txt. +target_compile_definitions(ccfcrypto PRIVATE CCF_STATIC_LIBRARY_BUILD) target_compile_options( ccfcrypto PRIVATE $<$:-Wno-vla-cxx-extension> diff --git a/include/ccf/pal/locking.h b/include/ccf/pal/locking.h index 72cb019b46ea..08691ee775e4 100644 --- a/include/ccf/pal/locking.h +++ b/include/ccf/pal/locking.h @@ -6,6 +6,7 @@ #include #include +#include #include namespace ccf::pal @@ -13,6 +14,20 @@ namespace ccf::pal class ConditionVariable; class MutexGuard; +#if defined(CCF_TEST_INTERLEAVING_LOCK_TYPE) + // A test build may define this (before this header is first included, + // via a -include compiler flag applying to every source file in that + // build) to replace ccf::pal::Mutex itself, everywhere, with a different, + // instrumented lock type - see that type's own declaration for what it + // does instead of real locking. MutexGuard and ConditionVariable below + // are both written against the name Mutex, so they bind to whichever + // type this resolves to; the replacement type must therefore expose the + // same public lock()/try_lock()/unlock() surface, and (for + // ConditionVariable::wait() and friends to keep compiling) a private + // member also named `mutex`, friended to ConditionVariable, of type + // std::mutex. + using Mutex = CCF_TEST_INTERLEAVING_LOCK_TYPE; +#else /** * Virtual enclaves and the host code share the same PAL. */ @@ -50,6 +65,7 @@ namespace ccf::pal return mutex.native_handle(); } }; +#endif class CCF_SCOPED_CAPABILITY MutexGuard { @@ -160,4 +176,92 @@ namespace ccf::pal lock.get(), timeout_time, std::move(predicate)); } }; + + // Called (if non-null) whenever a ccf::pal::unique_lock below actually + // acquires its lock, with a short label describing why - either given + // explicitly at the call site, or (if not) a source-location-derived + // default. Null outside of test code that wants to observe this; see + // src/commit_concurrency/deterministic_scheduler.h's SchedulerThreadContext, + // the one place that currently sets it, forwarding to + // DeterministicScheduler::set_action() so a failing scenario's + // describe() can show real semantic reasons at real lock points, not + // just its own explicit yield_point() labels. Deliberately not + // thread_local: the one place that installs it already reads its own + // thread-local state to decide whether the calling thread has an active + // scheduler, so this only ever needs a single, one-time global install. + using LockLabelSink = void (*)(const char* label); + inline LockLabelSink lock_label_sink = nullptr; + + // A drop-in replacement for std::unique_lock (supporting the same + // deferred-locking constructor and lock()/try_lock()/unlock() surface + // used against ccf::pal::Mutex elsewhere in this codebase), with an + // optional label describing why this lock is being taken - reported to + // lock_label_sink above every time this actually acquires the lock. With + // no label given, the label defaults to the call site's source location. + template + class unique_lock + { + std::unique_lock inner; + const char* label; + std::source_location loc; + + void report_if_locked() + { + if (inner.owns_lock() && lock_label_sink != nullptr) + { + lock_label_sink(label != nullptr ? label : loc.function_name()); + } + } + + public: + explicit unique_lock( + LockType& mtx, + const char* label_ = nullptr, + std::source_location loc_ = std::source_location::current()) : + inner(mtx), + label(label_), + loc(loc_) + { + report_if_locked(); + } + + unique_lock( + LockType& mtx, + std::defer_lock_t defer, + const char* label_ = nullptr, + std::source_location loc_ = std::source_location::current()) : + inner(mtx, defer), + label(label_), + loc(loc_) + {} + + void lock() + { + inner.lock(); + report_if_locked(); + } + + bool try_lock() + { + const bool locked = inner.try_lock(); + if (locked) + { + report_if_locked(); + } + return locked; + } + + void unlock() + { + inner.unlock(); + } + + bool owns_lock() const + { + return inner.owns_lock(); + } + + unique_lock(const unique_lock&) = delete; + unique_lock& operator=(const unique_lock&) = delete; + }; } diff --git a/src/commit_concurrency/deterministic_scheduler.h b/src/commit_concurrency/deterministic_scheduler.h new file mode 100644 index 000000000000..9f8128082a9d --- /dev/null +++ b/src/commit_concurrency/deterministic_scheduler.h @@ -0,0 +1,884 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#pragma once + +// A cooperative scheduler for deterministically exploring thread +// interleavings, plus SchedulerMutex, a lock type that reports its +// lock()/unlock() calls to whichever scheduler is active on the calling +// thread. Each participating actor runs on its own real OS thread, but the +// scheduler only ever lets one actor execute application code at a time; +// SchedulerMutex's lock()/unlock() calls are the points where it may hand +// control to a different actor instead of letting the caller continue. +// +// explore_all_interleavings() repeats a run once for every distinct +// sequence of such handoffs, via depth-first search with replay: each run +// records the choice made at every point where more than one actor was +// ready to proceed, and the next run replays the same choices up to the +// last such point and then tries the next untried alternative there. +// Every actor's work must therefore be reconstructed from scratch for each +// run (a fresh fixture, fresh threads) and depend on nothing outside what +// the scheduler controls, or two runs that replay the same prefix could +// diverge and make the recorded prefix meaningless. +// +// Exhaustive search does not scale to every scenario - estimate_schedule_ +// count() gives a rough, cheap estimate of how many schedules a scenario +// would take to exhaust, before committing to running that many; once it +// is clearly too many, explore_random_interleavings() samples a chosen +// number of schedules at random instead, still fully reproducibly from a +// seed (exactly, unlike a real-thread fuzzer's timing-based randomness). +// +// A SchedulerMutex used with no scheduler active on the calling thread +// behaves like an ordinary mutex. + +#include "ccf/ds/thread_safety.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace ccf::pal +{ + // Forward declared so SchedulerMutex below can friend it - see + // SchedulerMutex's own declaration for why. ccf/pal/locking.h is only + // included (see below) once SchedulerMutex is a complete type - it may + // become the definition of ccf::pal::Mutex itself for a whole build (see + // CCF_TEST_INTERLEAVING_LOCK_TYPE there), which locking.h's own + // MutexGuard and ConditionVariable need to be complete to compile + // against. + class ConditionVariable; +} + +namespace ccf::kv::test +{ + using ActorId = size_t; + + class DeterministicScheduler + { + public: + // One entry per point where the scheduler chose which ready actor + // would run next: every actor that was ready at that point (with + // whatever action label - see set_action() below - it had most + // recently set for itself), and the index within that list of the one + // actually chosen. + struct Decision + { + std::vector ready; + std::vector ready_actions; + size_t chosen_index; + }; + + private: + struct MutexState + { + std::optional owner; + std::vector waiters; + }; + + std::mutex m; + std::condition_variable cv; + size_t num_actors; + size_t parked_count = 0; + std::vector finished; + std::vector blocked_on_lock; + std::optional running; + std::function chooser; + std::vector path; + std::vector actor_names; + std::vector current_action; + + // Falls back to "actor " for any actor with no name given to the + // constructor, or an empty name. + std::string actor_label(ActorId a) const + { + if (a < actor_names.size() && !actor_names[a].empty()) + { + return actor_names[a]; + } + return "actor " + std::to_string(a); + } + + // Must be called with m held. Picks the next actor to run by asking + // `chooser` for an index into the ready set - the set of every actor + // that is neither finished nor currently blocked waiting on a lock - + // see the constructor's comment for what strategies that can be. + // + // TODO: every lock/unlock/yield_point() is currently an unconditional + // decision point (branches the search over every ready actor). A + // useful middle ground: treat each of these as only a *candidate* + // decision point, and let a per-scenario predicate (matched against + // the real semantic label already reported via + // ccf::pal::lock_label_sink/yield_point()'s own label - no further + // production code changes needed) decide whether it actually + // branches, or just fast-passes the current actor through unchanged + // (as the driver "actor" already does unconditionally below). Real + // mutual exclusion is unaffected either way - only whether the search + // explores alternatives there. This lets a scenario dial the search + // space down to exactly the handful of points it cares about (e.g. + // "the unlock of version_lock in Store::commit()"), rather than + // choosing between "every lock branches" (often computationally + // infeasible - see estimate_schedule_count()) and "only explicit + // yield_points branch" (may miss semantic-lock-ordering bugs + // entirely). Suggested workflow once this exists: random search over + // the full, unfiltered space to find violations; turn each found + // violation into a deterministic regression test pinned to its exact + // decision sequence; then fuzz with a narrow allowlist around those + // known points for cheap, targeted, ongoing coverage. + void choose_next(std::unique_lock& lock) + { + (void)lock; + std::vector ready; + for (ActorId a = 0; a < num_actors; ++a) + { + if (!finished[a] && !blocked_on_lock[a]) + { + ready.push_back(a); + } + } + if (ready.empty()) + { + throw std::logic_error( + "DeterministicScheduler: every unfinished actor is blocked on a " + "lock - deadlock"); + } + + const size_t chosen_index = chooser(ready.size()); + if (chosen_index >= ready.size()) + { + throw std::logic_error( + "DeterministicScheduler: chooser returned an out-of-range index " + "- if replaying a recorded path, the scenario is not " + "deterministic given the choices the scheduler controls"); + } + std::vector ready_actions; + ready_actions.reserve(ready.size()); + for (auto a : ready) + { + ready_actions.push_back(current_action[a]); + } + path.push_back(Decision{ready, std::move(ready_actions), chosen_index}); + running = ready[chosen_index]; + cv.notify_all(); + } + + public: + // `chooser_` is asked, at every decision point, to pick an index in + // [0, num_ready) - the strategy that makes it e.g. depth-first search + // with replay, or uniformly random, lives outside this class (see + // explore_all_interleavings() and explore_random_interleavings() + // below); DeterministicScheduler itself is agnostic to how choices are + // made, only to enacting whichever one is made. `actor_names_`, if + // given, is used by describe() below in place of "actor " - it + // need not name every actor, and is otherwise unused. + DeterministicScheduler( + size_t num_actors_, + std::function chooser_, + std::vector actor_names_ = {}) : + num_actors(num_actors_), + finished(num_actors_, false), + blocked_on_lock(num_actors_, false), + chooser(std::move(chooser_)), + actor_names(std::move(actor_names_)), + // One extra slot beyond the real actors, for the reserved driver id + // (see DriverRegistration) - the driver never contends for a lock or + // gets scheduled, but can still call set_action() (transitively, via + // ccf::pal::unique_lock's label reporting) while running application + // code during make_run()/on_schedule(). + current_action(num_actors_ + 1) + {} + + // Called by each actor's thread before it does any real work. Blocks + // until every actor has reached this point, then further blocks until + // this actor is the first one chosen to run. + void wait_for_start(ActorId self) + { + std::unique_lock lock(m); + ++parked_count; + cv.notify_all(); + cv.wait(lock, [&] { return running == self; }); + } + + // Called by the driver thread once every actor has been created, to + // make the first scheduling decision. Blocks until all actors have + // reached wait_for_start(). + void kick_off() + { + std::unique_lock lock(m); + cv.wait(lock, [&] { return parked_count == num_actors; }); + choose_next(lock); + } + + // An explicit, always-branching decision point: every ready actor is a + // candidate, regardless of what any of them are doing. Scenarios use + // this (via the free function yield_point() below) to mark specific + // points as worth exploring every interleaving of, independent of + // whether a lock happens to be involved there - e.g. a gap between two + // unrelated critical sections. Called with no scheduler active, it is + // a no-op (see yield_point()). If `label` is non-empty, it is recorded + // as this actor's current action (as set_action() below would) before + // the decision is made, so it appears in describe()'s output for this + // decision point. + void yield_point(ActorId self, std::string label = {}) + { + std::unique_lock lock(m); + if (!label.empty()) + { + current_action[self] = std::move(label); + } + choose_next(lock); + cv.wait(lock, [&] { return running == self; }); + } + + // Records what this actor is currently doing (or about to do), purely + // for describe() to report later - does not itself create a decision + // point. Overwrites whatever this actor last set, and has no effect + // once set until the next call (in particular, it is not cleared when + // the actor finishes, so the last thing an actor did remains visible + // in describe() for any later decision another actor triggers). + void set_action(ActorId self, std::string label) + { + std::unique_lock lock(m); + current_action[self] = std::move(label); + } + + // Called by SchedulerMutex::lock(). Blocks until this actor actually + // holds the lock. Every acquisition is itself a decision point - once + // this actor takes ownership (whether or not it had to wait for it), + // the scheduler considers every ready actor, including this one + // continuing immediately, before letting it proceed. The one + // exception is the reserved driver "actor" (see DriverRegistration): + // it never actually contends with a real actor for any lock, so its + // own incidental lock use (e.g. real work done while constructing a + // scenario's fixture) only needs to update ownership bookkeeping + // consistently for whichever real actor looks at the same lock next - + // not create a decision point of its own, since no other actor thread + // even exists yet to be a candidate. + void before_lock(ActorId self, void* mutex_key) + { + std::unique_lock lock(m); + auto& mtx = mutex_states[mutex_key]; + if (self >= num_actors) + { + mtx.owner = self; + return; + } + while (mtx.owner.has_value()) + { + mtx.waiters.push_back(self); + blocked_on_lock[self] = true; + choose_next(lock); + cv.wait(lock, [&] { return running == self; }); + // Someone else may have taken it between this actor being woken + // and it running again - the loop condition re-checks that. + } + mtx.owner = self; + choose_next(lock); + cv.wait(lock, [&] { return running == self; }); + } + + // Called by SchedulerMutex::unlock(), after releasing it. Every + // release is itself a decision point, whether or not anything was + // specifically waiting on this lock - any ready actor (including one + // now free to claim this lock) is a candidate to run next. As in + // before_lock() above, the reserved driver "actor" is the one + // exception - it only needs to clear its own ownership bookkeeping. + void after_unlock(ActorId self, void* mutex_key) + { + std::unique_lock lock(m); + auto& mtx = mutex_states[mutex_key]; + mtx.owner.reset(); + if (self >= num_actors) + { + return; + } + if (!mtx.waiters.empty()) + { + const auto woken = mtx.waiters.front(); + mtx.waiters.erase(mtx.waiters.begin()); + blocked_on_lock[woken] = false; + } + choose_next(lock); + cv.wait(lock, [&] { return running == self; }); + } + + // Called by an actor's thread once it has no more work to do. + void finish(ActorId self) + { + std::unique_lock lock(m); + finished[self] = true; + if (std::all_of( + finished.begin(), finished.end(), [](bool f) { return f; })) + { + running.reset(); + cv.notify_all(); + return; + } + choose_next(lock); + } + + const std::vector& decision_path() const + { + return path; + } + + // A human-readable rendering of decision_path(), one line per + // decision: every actor that was ready at that point (name and + // current action, if either was given), with the one chosen marked. + // Intended for a failing test to attach to its own failure output + // (e.g. via DOCTEST_INFO) - this scheduler has no opinion on when + // that should happen. + std::string describe() const + { + std::string out; + for (size_t i = 0; i < path.size(); ++i) + { + const auto& decision = path[i]; + out += std::to_string(i) + ": "; + for (size_t j = 0; j < decision.ready.size(); ++j) + { + if (j > 0) + { + out += ", "; + } + out += (j == decision.chosen_index ? "-> " : " "); + out += actor_label(decision.ready[j]); + if (!decision.ready_actions[j].empty()) + { + out += " (" + decision.ready_actions[j] + ")"; + } + } + out += "\n"; + } + return out; + } + + private: + // Keyed by SchedulerMutex identity (its `this` pointer) rather than + // held inside SchedulerMutex itself, so SchedulerMutex stays a plain, + // cheap, default-constructible value with no dependency on whichever + // scheduler (if any) ends up using it. + std::unordered_map mutex_states; + }; + + // Finds, and points a thread at, whichever DeterministicScheduler (if + // any) is exploring interleavings on the calling thread. + class SchedulerThreadContext + { + static thread_local DeterministicScheduler* current_scheduler; + static thread_local ActorId current_actor; + + public: + // Forwards ccf::pal::unique_lock's label reports (see + // include/ccf/pal/locking.h) to whichever scheduler is active on the + // calling thread (if any - a no-op otherwise), as with set_action() + // below. Installed once, globally, by the static initializer below; + // reads the calling thread's own current_scheduler/current_actor to + // decide what to do, so does not itself need to be installed or + // removed per-thread. Defined out-of-line, after ccf/pal/locking.h is + // included below (see SchedulerMutex's own comment for why that must + // come after this point in the file). + static void forward_lock_label(const char* label); + + static void set(DeterministicScheduler* scheduler, ActorId actor) + { + current_scheduler = scheduler; + current_actor = actor; + } + + static void clear() + { + current_scheduler = nullptr; + } + + static DeterministicScheduler* scheduler() + { + return current_scheduler; + } + + static ActorId actor() + { + return current_actor; + } + }; + + inline thread_local DeterministicScheduler* + SchedulerThreadContext::current_scheduler = nullptr; + inline thread_local ActorId SchedulerThreadContext::current_actor = 0; + + // An explicit point for explore_all_interleavings() to consider every + // ready actor as a candidate to run next, independent of any lock - + // e.g. a gap between two unrelated critical sections that a scenario + // wants every interleaving of, not just the ones lock contention alone + // would produce. A no-op with no scheduler active on the calling + // thread. If `label` is non-empty, it is recorded as with set_action() + // below before the decision is made. + inline void yield_point(std::string label = {}) + { + auto* scheduler = SchedulerThreadContext::scheduler(); + if (scheduler != nullptr) + { + scheduler->yield_point(SchedulerThreadContext::actor(), std::move(label)); + } + } + + // Records what the calling actor is currently doing (or about to do), + // purely so that DeterministicScheduler::describe() can report it + // against whichever decision point comes next - see + // DeterministicScheduler::set_action() for details. A no-op with no + // scheduler active on the calling thread. + inline void set_action(std::string label) + { + auto* scheduler = SchedulerThreadContext::scheduler(); + if (scheduler != nullptr) + { + scheduler->set_action(SchedulerThreadContext::actor(), std::move(label)); + } + } + + // A BasicLockable/Lockable type, suitable everywhere ccf::pal::Mutex is + // (std::lock_guard, std::unique_lock, std::scoped_lock all accept any + // type with these three members). With no DeterministicScheduler active + // on the calling thread, this behaves like an ordinary mutex; the + // scheduler-driven behaviour above only applies inside a run started via + // explore_all_interleavings() (or DeterministicScheduler used directly). + // + // Carries the same Clang thread-safety annotations as ccf::pal::Mutex, + // and the same private member name `mutex` (friended to + // ccf::pal::ConditionVariable, exactly as ccf::pal::Mutex friends it), so + // that this can stand in for ccf::pal::Mutex itself for a whole build + // (see CCF_TEST_INTERLEAVING_LOCK_TYPE in include/ccf/pal/locking.h) - + // including code that only compiles ccf::pal::ConditionVariable::wait() + // and friends without ever actually executing them at runtime. + class CCF_CAPABILITY("mutex") SchedulerMutex + { + friend class ccf::pal::ConditionVariable; + std::mutex mutex; + + public: + using native_handle_type = std::mutex::native_handle_type; + + SchedulerMutex() = default; + SchedulerMutex(const SchedulerMutex&) = delete; + SchedulerMutex& operator=(const SchedulerMutex&) = delete; + + void lock() CCF_ACQUIRE() + { + auto* scheduler = SchedulerThreadContext::scheduler(); + if (scheduler == nullptr) + { + mutex.lock(); + return; + } + scheduler->before_lock(SchedulerThreadContext::actor(), this); + } + + void unlock() CCF_RELEASE() + { + auto* scheduler = SchedulerThreadContext::scheduler(); + if (scheduler == nullptr) + { + mutex.unlock(); + return; + } + scheduler->after_unlock(SchedulerThreadContext::actor(), this); + } + + bool try_lock() CCF_TRY_ACQUIRE(true) + { + auto* scheduler = SchedulerThreadContext::scheduler(); + if (scheduler == nullptr) + { + return mutex.try_lock(); + } + // Not part of any of the scenarios this rig currently drives - + // implement only once a scenario actually needs it, so that its + // scheduling semantics can be designed against a real use rather + // than guessed at. + throw std::logic_error( + "SchedulerMutex::try_lock() is not implemented under an active " + "DeterministicScheduler"); + } + + native_handle_type native_handle() + { + return mutex.native_handle(); + } + }; +} + +// Only included here, rather than at the top of this file, because +// ccf/pal/locking.h may make ccf::pal::Mutex itself an alias for +// SchedulerMutex above (see CCF_TEST_INTERLEAVING_LOCK_TYPE there) - its +// own MutexGuard and ConditionVariable need SchedulerMutex to already be a +// complete type to compile against it. +#include "ccf/pal/locking.h" + +namespace ccf::kv::test +{ + inline void SchedulerThreadContext::forward_lock_label(const char* label) + { + if (current_scheduler != nullptr) + { + current_scheduler->set_action(current_actor, label); + } + } + + // Installs forward_lock_label() as ccf::pal::lock_label_sink exactly + // once, for the lifetime of the process - not per-thread, since + // forward_lock_label() already reads its own calling thread's + // thread-local current_scheduler to no-op when that thread has none. + namespace + { + struct LockLabelSinkInstaller + { + LockLabelSinkInstaller() + { + ccf::pal::lock_label_sink = &SchedulerThreadContext::forward_lock_label; + } + }; + const LockLabelSinkInstaller lock_label_sink_installer; + } + + // Registers/unregisters the calling (driver) thread with `scheduler` as + // a reserved actor id (one beyond the real actors, so it never collides + // with one), so that any SchedulerMutex it locks - during make_run() or + // on_schedule(), the only places the driver thread runs application code + // - goes through the same scheduler bookkeeping a real actor's would, + // rather than falling back to real locking. This driver "actor" never + // actually contends with a real actor for any lock: make_run() runs + // strictly before any actor thread starts, and on_schedule() strictly + // after every actor thread has finished and been joined. + class DriverRegistration + { + DeterministicScheduler& scheduler; + ActorId id; + bool registered = false; + + public: + DriverRegistration(DeterministicScheduler& scheduler_, ActorId id_) : + scheduler(scheduler_), + id(id_) + { + set(); + } + + ~DriverRegistration() + { + clear(); + } + + void set() + { + if (!registered) + { + SchedulerThreadContext::set(&scheduler, id); + registered = true; + } + } + + void clear() + { + if (registered) + { + SchedulerThreadContext::clear(); + registered = false; + } + } + + DriverRegistration(const DriverRegistration&) = delete; + DriverRegistration& operator=(const DriverRegistration&) = delete; + }; + + // Runs `make_run` once per explored schedule. `make_run` must construct + // whatever fresh state the scenario needs (e.g. a fixture) and return + // exactly `num_actors` callables - the body to run, on its own thread, + // for each actor in that particular run. Every callable must call + // ccf::kv::test::SchedulerThreadContext::set() first if it wants that + // thread's SchedulerMutex use to be scheduled (any thread that never + // calls it behaves as if no scheduler were active at all). + // + // If given, `on_schedule` is called after every schedule's actors have + // all finished, before the state made by that schedule's `make_run` call + // is discarded - the place to check per-schedule invariants or tally + // outcomes across schedules. It is passed the scheduler itself, so it + // can call scheduler.describe() (typically attached via DOCTEST_INFO) + // to explain what happened on that schedule if it goes on to report a + // failure. + // + // `actor_names`, if given, labels each actor in scheduler.describe()'s + // output in place of "actor " - see DeterministicScheduler's + // constructor. + // + // Explores schedules via depth-first search with replay (see file + // comment above) until every alternative at every decision point has + // been tried, or `max_schedules` is reached first - a circuit breaker + // against scenarios whose interleaving space is too large to exhaust in + // practice, so a mistakenly-unbounded scenario fails loudly rather than + // running forever. Use estimate_schedule_count() below to get a rough + // idea of how large that space is before committing to an exhaustive + // search. Returns the number of schedules explored. + inline size_t explore_all_interleavings( + size_t num_actors, + const std::function>()>& make_run, + const std::function& on_schedule = {}, + size_t max_schedules = 100000, + std::vector actor_names = {}) + { + // Replays `prefix` (the choices made at each decision point up to and + // including the last one backtracked to), then defaults to the + // left-most alternative for every decision point beyond that - + // exactly depth-first search with replay. + struct PrefixThenLeftmostChooser + { + std::vector prefix; + size_t pos = 0; + + size_t operator()(size_t num_ready) + { + const size_t chosen = pos < prefix.size() ? prefix[pos] : 0; + ++pos; + return chosen < num_ready ? chosen : num_ready; + } + }; + + std::vector prefix; + size_t schedules_explored = 0; + + for (;;) + { + if (schedules_explored >= max_schedules) + { + throw std::logic_error( + "explore_all_interleavings: max_schedules reached without " + "exhausting every interleaving - scope the scenario down, or " + "raise the limit if this many schedules is genuinely expected"); + } + + DeterministicScheduler scheduler( + num_actors, PrefixThenLeftmostChooser{prefix, 0}, actor_names); + + // make_run() (constructing whatever fixture the scenario needs) runs + // here, on this driver thread, before any actor thread exists - so + // it is registered with this schedule's scheduler too (as actor id + // num_actors, never used by any real actor), rather than left + // unregistered. This matters whenever a SchedulerMutex reachable + // from make_run() is shared with something outside this scenario's + // own fixture (e.g. a process-wide singleton) - an unregistered + // thread takes such a lock for real, while a registered one only + // does scheduler bookkeeping; consistently registering every thread + // that can reach such a lock avoids that mismatch. Safe because + // this driver "actor" is never actually contended for by a real + // actor - it only ever touches such locks strictly before any actor + // starts, or strictly after every actor has finished (see below). + DriverRegistration driver_registration(scheduler, num_actors); + auto bodies = make_run(); + if (bodies.size() != num_actors) + { + throw std::logic_error( + "explore_all_interleavings: make_run() did not return one body " + "per actor"); + } + driver_registration.clear(); + + std::vector threads; + threads.reserve(num_actors); + for (ActorId a = 0; a < num_actors; ++a) + { + threads.emplace_back([&scheduler, &bodies, a]() { + SchedulerThreadContext::set(&scheduler, a); + scheduler.wait_for_start(a); + bodies[a](); + scheduler.finish(a); + SchedulerThreadContext::clear(); + }); + } + scheduler.kick_off(); + for (auto& t : threads) + { + t.join(); + } + ++schedules_explored; + if (on_schedule) + { + driver_registration.set(); + on_schedule(scheduler); + driver_registration.clear(); + } + + // Backtrack: find the last decision with an untried alternative, + // and set the prefix to replay everything up to and including it, + // advanced to the next alternative there. + const auto& path = scheduler.decision_path(); + std::optional backtrack_at; + for (size_t i = path.size(); i-- > 0;) + { + if (path[i].chosen_index + 1 < path[i].ready.size()) + { + backtrack_at = i; + break; + } + } + if (!backtrack_at.has_value()) + { + // Every decision, at every depth, chose its last alternative: + // nothing left to explore. + return schedules_explored; + } + + prefix.clear(); + prefix.reserve(*backtrack_at + 1); + for (size_t i = 0; i < *backtrack_at; ++i) + { + prefix.push_back(path[i].chosen_index); + } + prefix.push_back(path[*backtrack_at].chosen_index + 1); + } + } + + // Runs `make_run` once per sample, choosing uniformly at random (seeded + // by `seed`, so the whole sequence of samples is reproducible) at every + // decision point instead of exhaustively searching every alternative. + // Useful once estimate_schedule_count() below shows the full space is + // too large to exhaust in practice, but a scenario is still worth + // sampling for interleavings a purely timing-based fuzzer might miss. + // `make_run`, `on_schedule`, and `actor_names` behave exactly as in + // explore_all_interleavings(). + inline void explore_random_interleavings( + size_t num_actors, + const std::function>()>& make_run, + const std::function& on_schedule, + size_t num_samples, + uint32_t seed, + std::vector actor_names = {}) + { + struct RandomChooser + { + std::mt19937 rng; + + size_t operator()(size_t num_ready) + { + return std::uniform_int_distribution(0, num_ready - 1)(rng); + } + }; + + std::mt19937 seed_rng(seed); + for (size_t sample = 0; sample < num_samples; ++sample) + { + DeterministicScheduler scheduler( + num_actors, RandomChooser{std::mt19937(seed_rng())}, actor_names); + DriverRegistration driver_registration(scheduler, num_actors); + auto bodies = make_run(); + if (bodies.size() != num_actors) + { + throw std::logic_error( + "explore_random_interleavings: make_run() did not return one " + "body per actor"); + } + driver_registration.clear(); + + std::vector threads; + threads.reserve(num_actors); + for (ActorId a = 0; a < num_actors; ++a) + { + threads.emplace_back([&scheduler, &bodies, a]() { + SchedulerThreadContext::set(&scheduler, a); + scheduler.wait_for_start(a); + bodies[a](); + scheduler.finish(a); + SchedulerThreadContext::clear(); + }); + } + scheduler.kick_off(); + for (auto& t : threads) + { + t.join(); + } + if (on_schedule) + { + driver_registration.set(); + on_schedule(scheduler); + driver_registration.clear(); + } + } + } + + // A rough estimate of how many schedules explore_all_interleavings() + // would need to exhaust the full interleaving space of this scenario, + // without actually exhausting it: `num_walks` independent random walks + // down the decision tree, each multiplying together the number of ready + // candidates at every decision point it passes through (an unbiased + // estimator of the tree's total leaf count - the same technique used to + // estimate game tree sizes without expanding them in full). A single + // walk has high variance, so this returns every walk's estimate rather + // than just one number - look at the spread (e.g. min/max, or a + // geometric mean) rather than trusting any individual value, and treat + // the result as an order of magnitude, not a precise count. + inline std::vector estimate_schedule_count( + size_t num_actors, + const std::function>()>& make_run, + size_t num_walks = 30, + uint32_t seed = 1) + { + struct EstimatingRandomChooser + { + std::mt19937 rng; + double* product; + + size_t operator()(size_t num_ready) + { + *product *= static_cast(num_ready); + return std::uniform_int_distribution(0, num_ready - 1)(rng); + } + }; + + std::vector estimates; + estimates.reserve(num_walks); + std::mt19937 seed_rng(seed); + + for (size_t walk = 0; walk < num_walks; ++walk) + { + double product = 1.0; + DeterministicScheduler scheduler( + num_actors, + EstimatingRandomChooser{std::mt19937(seed_rng()), &product}); + DriverRegistration driver_registration(scheduler, num_actors); + auto bodies = make_run(); + if (bodies.size() != num_actors) + { + throw std::logic_error( + "estimate_schedule_count: make_run() did not return one body " + "per actor"); + } + driver_registration.clear(); + + std::vector threads; + threads.reserve(num_actors); + for (ActorId a = 0; a < num_actors; ++a) + { + threads.emplace_back([&scheduler, &bodies, a]() { + SchedulerThreadContext::set(&scheduler, a); + scheduler.wait_for_start(a); + bodies[a](); + scheduler.finish(a); + SchedulerThreadContext::clear(); + }); + } + scheduler.kick_off(); + for (auto& t : threads) + { + t.join(); + } + estimates.push_back(product); + } + return estimates; + } +} diff --git a/src/commit_concurrency/deterministic_scheduler_test.cpp b/src/commit_concurrency/deterministic_scheduler_test.cpp new file mode 100644 index 000000000000..2784f938027f --- /dev/null +++ b/src/commit_concurrency/deterministic_scheduler_test.cpp @@ -0,0 +1,271 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#include "commit_concurrency/deterministic_scheduler.h" + +#define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES +#include +#include +#include + +using namespace ccf::kv::test; + +DOCTEST_TEST_CASE( + "A single actor with no contention explores exactly one schedule" * + doctest::test_suite("deterministic_scheduler")) +{ + size_t counter = 0; + const auto explored = + explore_all_interleavings(1, [&]() -> std::vector> { + counter = 0; + return {[&]() { counter = 1; }}; + }); + DOCTEST_CHECK(explored == 1); + DOCTEST_CHECK(counter == 1); +} + +DOCTEST_TEST_CASE( + "Two actors each incrementing a shared counter under a shared lock reach " + "the same, correct total on every explored interleaving" * + doctest::test_suite("deterministic_scheduler")) +{ + size_t counter = 0; + SchedulerMutex mtx; + + const auto explored = explore_all_interleavings( + 2, + [&]() -> std::vector> { + counter = 0; + return { + [&]() { + std::lock_guard guard(mtx); + counter++; + }, + [&]() { + std::lock_guard guard(mtx); + counter++; + }}; + }, + [&](const DeterministicScheduler&) { DOCTEST_CHECK(counter == 2); }); + + DOCTEST_INFO(fmt::format("Explored {} schedules", explored)); + DOCTEST_CHECK(explored > 1); +} + +namespace +{ + // A deliberately racy "lazy initialisation" pattern: each actor reads + // whether initialisation has already happened, and if not, performs it - + // but the read and the (potential) write are two separate critical + // sections rather than one, leaving a gap in which another actor can + // run. run_actor_with_gap() also marks that gap with yield_point(), on + // top of the decision points already made at each lock/unlock, purely + // to give it an explicit, named label in describe()'s output. + struct LazyInitScenario + { + bool initialised = false; + size_t init_count = 0; + SchedulerMutex mtx; + + void run_actor_with_gap() + { + bool already_done; + { + std::lock_guard guard(mtx); + already_done = initialised; + } + yield_point("checked initialised flag, about to act on it"); + if (!already_done) + { + std::lock_guard guard(mtx); + initialised = true; + init_count++; + } + } + + void run_actor_without_gap() + { + std::lock_guard guard(mtx); + if (!initialised) + { + initialised = true; + init_count++; + } + } + }; +} + +DOCTEST_TEST_CASE( + "A lazy-init race across two separate critical sections is caught on at " + "least one, but not all, explored interleavings, and describe() explains " + "the first such schedule" * + doctest::test_suite("deterministic_scheduler")) +{ + std::unique_ptr scenario; + size_t schedules_with_double_init = 0; + size_t schedules_with_single_init = 0; + std::string first_bad_schedule_description; + + const auto explored = explore_all_interleavings( + 2, + [&]() -> std::vector> { + scenario = std::make_unique(); + return { + [&]() { scenario->run_actor_with_gap(); }, + [&]() { scenario->run_actor_with_gap(); }}; + }, + [&](const DeterministicScheduler& scheduler) { + if (scenario->init_count > 1) + { + schedules_with_double_init++; + if (first_bad_schedule_description.empty()) + { + first_bad_schedule_description = scheduler.describe(); + } + } + else + { + schedules_with_single_init++; + } + }, + 100000, + {"first", "second"}); + + DOCTEST_INFO(fmt::format( + "Explored {} schedules: {} with a double init, {} with a single init", + explored, + schedules_with_double_init, + schedules_with_single_init)); + DOCTEST_CHECK(explored > 1); + DOCTEST_CHECK(schedules_with_double_init > 0); + DOCTEST_CHECK(schedules_with_single_init > 0); + + DOCTEST_INFO( + "describe() names the two actors as given, and shows both taking " + "their post-check action label before either one wins the race"); + DOCTEST_CHECK( + first_bad_schedule_description.find("first") != std::string::npos); + DOCTEST_CHECK( + first_bad_schedule_description.find("second") != std::string::npos); + DOCTEST_CHECK( + first_bad_schedule_description.find( + "checked initialised flag, about to act on it") != std::string::npos); +} + +DOCTEST_TEST_CASE( + "Collapsing the check and the write into one critical section removes " + "the race on every explored interleaving" * + doctest::test_suite("deterministic_scheduler")) +{ + std::unique_ptr scenario; + size_t schedules_with_double_init = 0; + + const auto explored = explore_all_interleavings( + 2, + [&]() -> std::vector> { + scenario = std::make_unique(); + return { + [&]() { scenario->run_actor_without_gap(); }, + [&]() { scenario->run_actor_without_gap(); }}; + }, + [&](const DeterministicScheduler&) { + if (scenario->init_count > 1) + { + schedules_with_double_init++; + } + }); + + DOCTEST_INFO(fmt::format("Explored {} schedules", explored)); + DOCTEST_CHECK(explored > 1); + DOCTEST_CHECK(schedules_with_double_init == 0); +} + +DOCTEST_TEST_CASE( + "Random sampling of the same racy scenario reliably hits the bug too, " + "and is exactly reproducible from its seed" * + doctest::test_suite("deterministic_scheduler")) +{ + std::unique_ptr scenario; + auto make_run = [&]() -> std::vector> { + scenario = std::make_unique(); + return { + [&]() { scenario->run_actor_with_gap(); }, + [&]() { scenario->run_actor_with_gap(); }}; + }; + + size_t schedules_with_double_init = 0; + explore_random_interleavings( + 2, + make_run, + [&](const DeterministicScheduler&) { + if (scenario->init_count > 1) + { + schedules_with_double_init++; + } + }, + 50, + 42); + DOCTEST_INFO(fmt::format( + "{} of 50 randomly sampled schedules hit the double-init bug", + schedules_with_double_init)); + DOCTEST_CHECK(schedules_with_double_init > 0); + + // Same seed, same 50 samples: an exact repeat, not just "close enough". + size_t schedules_with_double_init_repeat = 0; + explore_random_interleavings( + 2, + make_run, + [&](const DeterministicScheduler&) { + if (scenario->init_count > 1) + { + schedules_with_double_init_repeat++; + } + }, + 50, + 42); + DOCTEST_CHECK( + schedules_with_double_init_repeat == schedules_with_double_init); +} + +DOCTEST_TEST_CASE( + "estimate_schedule_count reports exactly one schedule for a scenario " + "with no branching, and a plausible order of magnitude for one that has " + "some" * + doctest::test_suite("deterministic_scheduler")) +{ + { + size_t counter = 0; + const auto estimates = + estimate_schedule_count(1, [&]() -> std::vector> { + counter = 0; + return {[&]() { counter = 1; }}; + }); + for (const auto estimate : estimates) + { + DOCTEST_CHECK(estimate == 1.0); + } + } + + { + // The exhaustive test above finds exactly 736 schedules for this + // scenario now that every lock/unlock (not just contended ones) is a + // decision point - a random-walk estimate is not expected to land on + // that exactly, but should be in the right ballpark rather than off + // by orders of magnitude. + std::unique_ptr scenario; + const auto estimates = + estimate_schedule_count(2, [&]() -> std::vector> { + scenario = std::make_unique(); + return { + [&]() { scenario->run_actor_with_gap(); }, + [&]() { scenario->run_actor_with_gap(); }}; + }); + double min_estimate = *std::min_element(estimates.begin(), estimates.end()); + double max_estimate = *std::max_element(estimates.begin(), estimates.end()); + DOCTEST_INFO(fmt::format( + "Estimates ranged from {} to {} (true count is 736)", + min_estimate, + max_estimate)); + DOCTEST_CHECK(min_estimate >= 1.0); + DOCTEST_CHECK(max_estimate <= 20000.0); + } +} diff --git a/src/commit_concurrency/interleaving.h b/src/commit_concurrency/interleaving.h new file mode 100644 index 000000000000..1053d0df9523 --- /dev/null +++ b/src/commit_concurrency/interleaving.h @@ -0,0 +1,159 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#pragma once + +// Generic building blocks for deterministically interleaving real +// production code paths (e.g. Store::commit()'s batching loop, or a Tx's +// write-set serialisation) with a concurrent action injected from a +// controller thread (e.g. a Store::rollback() triggered by a real raft +// view change). +// +// Two complementary tools are provided: +// - Checkpoint: a named pause/release rendezvous, for pinning an exact +// interleaving (a worker thread pauses at a point of interest; a +// controller thread waits for that, performs some action, then releases +// it). +// - random_delay: an unpinned timing-fuzz helper, for shaking loose races +// whose exact window is not known up front. +// +// Neither of these requires any changes to production code: they attach via +// existing extension points (ccf::kv::CommittableTx::WriteSetObserver, and +// wrapping ccf::kv::PendingTx). + +#include "kv/kv_types.h" + +#include +#include +#include +#include +#include +#include +#include + +namespace ccf::kv::test +{ + // A single pause/release rendezvous point. One thread calls pause() and + // blocks; another thread calls wait_until_paused() to learn that the first + // thread has reached this point, does whatever it needs to do while the + // first thread is parked, then calls release() to let it continue. + // + // A Checkpoint may be reused for multiple pause/release cycles (e.g. one + // per iteration of a batching loop, or one per fuzzer iteration), as long + // as each cycle's pause() is fully released before the next one begins. + class Checkpoint + { + std::mutex lock; + std::condition_variable paused_cv; + std::condition_variable resume_cv; + bool paused = false; + bool resume = false; + + public: + // Optional name, purely for log/assertion messages when a test uses + // several Checkpoints at once. + const std::string name; + + Checkpoint(std::string name_ = "") : name(std::move(name_)) {} + + // Called by the worker thread. Blocks until a controller thread calls + // release(). + void pause() + { + { + std::lock_guard guard(lock); + // Consume any leftover `resume` from a previous pause/release cycle + // on this Checkpoint before waiting on it again, so this can be + // safely reused (see release(), which deliberately does not touch + // this flag itself, to avoid racing with the very wait() below). + resume = false; + paused = true; + } + paused_cv.notify_one(); + + std::unique_lock guard(lock); + resume_cv.wait(guard, [this]() { return resume; }); + } + + // Called by the controller thread. Blocks until a worker thread has + // called pause(). + void wait_until_paused() + { + std::unique_lock guard(lock); + paused_cv.wait(guard, [this]() { return paused; }); + // Consume `paused`, so this Checkpoint can be reused for a later + // pause/release cycle without wait_until_paused() immediately + // (incorrectly) returning for a pause() call that hasn't happened yet. + paused = false; + } + + // Called by the controller thread. Releases a worker thread waiting in + // pause(). + void release() + { + std::lock_guard guard(lock); + resume = true; + resume_cv.notify_one(); + } + + // Convenience for the controller thread: wait for a worker to arrive, + // then immediately release it. Useful when the interleaving only needs a + // happens-before edge (e.g. "let this transaction's local application + // complete before doing anything else") rather than an inspection + // window. + void wait_until_paused_and_release() + { + wait_until_paused(); + release(); + } + }; + + // A ccf::kv::CommittableTx::WriteSetObserver-compatible adaptor which + // pauses at a Checkpoint every time it is invoked, i.e. once the + // transaction's write set has been serialised but before it is handed to + // Store::commit(). + inline auto checkpoint_write_set_observer(Checkpoint& checkpoint) + { + return [&checkpoint](const auto&, const auto&) { checkpoint.pause(); }; + } + + // Wraps another PendingTx, and pauses at a Checkpoint after the inner + // PendingTx has produced its result (i.e. after the entry's local + // application to the KV is complete) but before that result is returned to + // Store::commit()'s batching loop. Use this to pin a rollback so it lands + // strictly between two entries of the same in-flight commit batch. + class PausingPendingTx : public ccf::kv::PendingTx + { + std::unique_ptr inner; + Checkpoint& checkpoint; + + public: + PausingPendingTx( + std::unique_ptr inner_, Checkpoint& checkpoint_) : + inner(std::move(inner_)), + checkpoint(checkpoint_) + {} + + ccf::kv::PendingTxInfo call() override + { + auto info = inner->call(); + checkpoint.pause(); + return info; + } + }; + + // Pure timing-fuzz helper (no pinned interleaving): sleeps the calling + // thread for a pseudo-random duration in [0, max), drawn from the given + // RNG. Used by actors that should jitter relative to one another without + // the test dictating an exact interleaving. + inline void random_delay(std::mt19937& rng, std::chrono::microseconds max) + { + if (max.count() <= 0) + { + return; + } + + const auto delay_us = + std::uniform_int_distribution(0, max.count() - 1)(rng); + std::this_thread::sleep_for(std::chrono::microseconds(delay_us)); + } +} diff --git a/src/commit_concurrency/interleaving_lock_override.h b/src/commit_concurrency/interleaving_lock_override.h new file mode 100644 index 000000000000..883c5e530838 --- /dev/null +++ b/src/commit_concurrency/interleaving_lock_override.h @@ -0,0 +1,28 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#pragma once + +// Force-included (via a -include compiler flag) into every translation +// unit of the model-checked test target, before anything else, so that +// ccf::pal::Mutex itself (see include/ccf/pal/locking.h) resolves to +// SchedulerMutex for the whole of that target - and nowhere else, since no +// other target passes this flag. Every production call site that declares +// a ccf::pal::Mutex is therefore covered automatically, with no +// per-call-site changes anywhere in production code. +// +// Any static/singleton state reachable from this target that itself uses +// ccf::pal::Mutex (e.g. ccf::tasks' job board) is covered by this too, as +// long as every thread that can touch it is registered with the scheduler +// for the currently-running schedule - see DriverRegistration in +// deterministic_scheduler.h for the thread that runs make_run()/ +// on_schedule() itself, outside of any actor thread. +// +// CCF_TEST_INTERLEAVING_LOCK_TYPE must be defined before +// deterministic_scheduler.h is included below - that header now also +// includes ccf/pal/locking.h itself (to install its lock-label sink; see +// SchedulerThreadContext), and ccf/pal/locking.h's own #pragma once means +// whichever definition of Mutex is in scope on its first inclusion in this +// translation unit is the one every subsequent include sees. +#define CCF_TEST_INTERLEAVING_LOCK_TYPE ccf::kv::test::SchedulerMutex + +#include "commit_concurrency/deterministic_scheduler.h" diff --git a/src/commit_concurrency/interleaving_test.cpp b/src/commit_concurrency/interleaving_test.cpp new file mode 100644 index 000000000000..d854fd70c8a9 --- /dev/null +++ b/src/commit_concurrency/interleaving_test.cpp @@ -0,0 +1,121 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#include "ccf/crypto/sha256_hash.h" +#include "commit_concurrency/interleaving.h" + +#define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES +#include +#include +#include + +// These tests exercise the Checkpoint/random_delay primitives entirely in +// isolation, with no Store/Raft/History involved, to validate the mechanism +// itself before it is relied upon elsewhere. + +DOCTEST_TEST_CASE( + "Checkpoint pauses a worker until explicitly released" * + doctest::test_suite("interleaving")) +{ + ccf::kv::test::Checkpoint checkpoint("test"); + std::atomic worker_progressed{false}; + + std::thread worker([&]() { + checkpoint.pause(); + worker_progressed = true; + }); + + checkpoint.wait_until_paused(); + // The worker must still be blocked in pause() at this point - there is no + // way to observe this with perfect certainty without a race, but a short + // delay makes a bug here overwhelmingly likely to be caught. + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + DOCTEST_CHECK_FALSE(worker_progressed.load()); + + checkpoint.release(); + worker.join(); + DOCTEST_CHECK(worker_progressed.load()); +} + +DOCTEST_TEST_CASE( + "Checkpoint can be reused for multiple sequential pause/release cycles" * + doctest::test_suite("interleaving")) +{ + ccf::kv::test::Checkpoint checkpoint; + constexpr size_t cycles = 20; + + for (size_t i = 0; i < cycles; ++i) + { + std::atomic progressed{0}; + std::thread worker([&]() { + checkpoint.pause(); + progressed = i + 1; + }); + + checkpoint.wait_until_paused(); + checkpoint.release(); + worker.join(); + DOCTEST_REQUIRE(progressed.load() == i + 1); + } +} + +DOCTEST_TEST_CASE( + "wait_until_paused_and_release is a one-shot happens-before edge" * + doctest::test_suite("interleaving")) +{ + ccf::kv::test::Checkpoint checkpoint; + std::atomic worker_progressed{false}; + + std::thread worker([&]() { + checkpoint.pause(); + worker_progressed = true; + }); + + checkpoint.wait_until_paused_and_release(); + worker.join(); + DOCTEST_CHECK(worker_progressed.load()); +} + +DOCTEST_TEST_CASE( + "checkpoint_write_set_observer pauses when invoked" * + doctest::test_suite("interleaving")) +{ + ccf::kv::test::Checkpoint checkpoint; + auto observer = ccf::kv::test::checkpoint_write_set_observer(checkpoint); + + std::atomic worker_progressed{false}; + std::thread worker([&]() { + observer(ccf::crypto::Sha256Hash(), std::string("evidence")); + worker_progressed = true; + }); + + checkpoint.wait_until_paused(); + DOCTEST_CHECK_FALSE(worker_progressed.load()); + checkpoint.release(); + worker.join(); + DOCTEST_CHECK(worker_progressed.load()); +} + +DOCTEST_TEST_CASE( + "random_delay respects its upper bound and can be zero" * + doctest::test_suite("interleaving")) +{ + std::mt19937 rng(1234); + + DOCTEST_INFO("A zero bound returns immediately"); + const auto before = std::chrono::steady_clock::now(); + ccf::kv::test::random_delay(rng, std::chrono::microseconds(0)); + const auto after = std::chrono::steady_clock::now(); + DOCTEST_CHECK(after - before < std::chrono::milliseconds(50)); + + DOCTEST_INFO("A non-zero bound is respected, across many draws"); + constexpr auto bound = std::chrono::microseconds(2000); + for (size_t i = 0; i < 100; ++i) + { + const auto start = std::chrono::steady_clock::now(); + ccf::kv::test::random_delay(rng, bound); + const auto elapsed = std::chrono::steady_clock::now() - start; + // Generous upper margin for scheduling jitter - this is checking that + // random_delay is bounded, not that it is precise. + DOCTEST_CHECK(elapsed < bound + std::chrono::milliseconds(50)); + } +} diff --git a/src/commit_concurrency/model_checked/main.cpp b/src/commit_concurrency/model_checked/main.cpp new file mode 100644 index 000000000000..db993926ffbd --- /dev/null +++ b/src/commit_concurrency/model_checked/main.cpp @@ -0,0 +1,20 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. + +// Doctest entry point for the model-checked concurrency suite: unlike +// commit_concurrency_test (real OS-thread timing, seeded but not +// exactly replayable), this suite drives the same real Store + Aft + +// MerkleTxHistory stack through ccf::kv::test::explore_all_interleavings(), +// exhaustively trying every legal interleaving of a bounded scenario rather +// than sampling a subset of them. + +#define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES +#define DOCTEST_CONFIG_IMPLEMENT +#include + +int main(int argc, char** argv) +{ + doctest::Context context; + context.applyCommandLine(argc, argv); + return context.run(); +} diff --git a/src/commit_concurrency/model_checked/rejected_commit_stall.cpp b/src/commit_concurrency/model_checked/rejected_commit_stall.cpp new file mode 100644 index 000000000000..d3b731638013 --- /dev/null +++ b/src/commit_concurrency/model_checked/rejected_commit_stall.cpp @@ -0,0 +1,154 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#include "commit_concurrency/deterministic_scheduler.h" +#include "commit_concurrency/threaded/fixture.h" + +#define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES +#include +#include +#include + +using namespace ccf::kv::test; + +namespace +{ + // Actor 0 (or any writer actor): reads (fixing this transaction's + // commit view), then attempts to commit an ordinary write. + // yield_point() is an explicit point for the scheduler to consider + // interleaving an election here, mirroring how a real thread could be + // preempted at that instant even though nothing here takes a lock. + void run_writer(CommitConcurrencyFixture& fixture, size_t key) + { + auto tx = fixture.store->create_tx(); + tx.rw(fixture.table)->put(key, key); + yield_point( + "read for a write to key " + std::to_string(key) + ", about to commit"); + tx.commit(); + } + + // Checks that replication has not permanently fallen behind the + // Store's own version - if a write landed locally without reaching + // consensus, a further ordinary commit must still let replication + // catch up to it. + bool replication_can_catch_up(CommitConcurrencyFixture& fixture) + { + auto later_tx = fixture.store->create_tx(); + later_tx.rw(fixture.table)->put(1000, 1000); + later_tx.commit(); + return fixture.raft->get_last_idx() == fixture.store->current_txid().seqno; + } +} + +// Randomly samples interleavings of a transaction committing across a +// real election, rather than the one pinned interleaving in +// deterministic.cpp - see that file for the invariant being checked and +// why it currently fails. +// estimate_schedule_count() below puts this scenario's interleaving space +// (every real lock acquisition is now a decision point, not just +// contended ones) far beyond what is practical to exhaust, so this +// samples a fixed, reproducible number of random schedules instead. +DOCTEST_TEST_CASE( + "Randomly sampled: every sampled interleaving of a stale-view commit " + "and a real election leaves replication able to catch up to the " + "Store's own version" * + doctest::test_suite("commit_concurrency_model")) +{ + std::unique_ptr fixture; + ccf::TxID baseline_txid; + + const auto make_run = [&]() -> std::vector> { + fixture = std::make_unique(); + baseline_txid = fixture->commit_signature(); + return {[&]() { run_writer(*fixture, 0); }, [&]() { fixture->reelect(); }}; + }; + const auto on_schedule = [&](const DeterministicScheduler& scheduler) { + if (fixture->store->current_txid().seqno != baseline_txid.seqno) + { + const auto description = "Schedule:\n" + scheduler.describe(); + DOCTEST_INFO(description); + DOCTEST_CHECK( + replication_can_catch_up(*fixture)); // NOTE_REJECTED_COMMIT_STALL + } + }; + + const auto estimates = estimate_schedule_count(2, make_run); + const double min_estimate = + *std::min_element(estimates.begin(), estimates.end()); + const double max_estimate = + *std::max_element(estimates.begin(), estimates.end()); + DOCTEST_MESSAGE(fmt::format( + "Estimated schedule count for the single-writer scenario: {} to {}", + min_estimate, + max_estimate)); + + constexpr size_t num_samples = 500; + constexpr uint32_t seed = 42; + DOCTEST_INFO(fmt::format( + "Sampling {} of an estimated {}-{} schedules (seed {})", + num_samples, + min_estimate, + max_estimate, + seed)); + explore_random_interleavings( + 2, make_run, on_schedule, num_samples, seed, {"writer 0", "elector"}); +} + +// The same invariant as above, but with a second concurrent writer +// added. estimate_schedule_count() below puts this scenario's +// interleaving space even further beyond what is practical to exhaust +// (see the DOCTEST_MESSAGE this prints), so this samples a fixed, +// reproducible number of random schedules instead. +DOCTEST_TEST_CASE( + "Randomly sampled: every sampled interleaving of two concurrent " + "stale-view commits and a real election leaves replication able to " + "catch up to the Store's own version" * + doctest::test_suite("commit_concurrency_model")) +{ + std::unique_ptr fixture; + ccf::TxID baseline_txid; + + const auto make_run = [&]() -> std::vector> { + fixture = std::make_unique(); + baseline_txid = fixture->commit_signature(); + return { + [&]() { run_writer(*fixture, 0); }, + [&]() { run_writer(*fixture, 1); }, + [&]() { fixture->reelect(); }}; + }; + const auto on_schedule = [&](const DeterministicScheduler& scheduler) { + if (fixture->store->current_txid().seqno != baseline_txid.seqno) + { + const auto description = "Schedule:\n" + scheduler.describe(); + DOCTEST_INFO(description); + DOCTEST_CHECK( + replication_can_catch_up(*fixture)); // NOTE_REJECTED_COMMIT_STALL + } + }; + + const auto estimates = estimate_schedule_count(3, make_run); + const double min_estimate = + *std::min_element(estimates.begin(), estimates.end()); + const double max_estimate = + *std::max_element(estimates.begin(), estimates.end()); + DOCTEST_MESSAGE(fmt::format( + "Estimated schedule count for the two-writer scenario: {} to {} " + "(compare with the single-writer scenario's estimate above)", + min_estimate, + max_estimate)); + + constexpr size_t num_samples = 500; + constexpr uint32_t seed = 42; + DOCTEST_INFO(fmt::format( + "Sampling {} of an estimated {}-{} schedules (seed {})", + num_samples, + min_estimate, + max_estimate, + seed)); + explore_random_interleavings( + 3, + make_run, + on_schedule, + num_samples, + seed, + {"writer 0", "writer 1", "elector"}); +} diff --git a/src/commit_concurrency/threaded/deterministic.cpp b/src/commit_concurrency/threaded/deterministic.cpp new file mode 100644 index 000000000000..44ad464e218d --- /dev/null +++ b/src/commit_concurrency/threaded/deterministic.cpp @@ -0,0 +1,424 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#include "commit_concurrency/threaded/fixture.h" + +#define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES +#include +#include +#include +#include +#include + +// Deterministic scenarios driven by CommitConcurrencyFixture, pinned via +// ccf::kv::test::Checkpoint from src/commit_concurrency/interleaving.h. + +using namespace ccf::kv::test; + +namespace +{ + // Directly drives Store::commit()-style application of a write to a + // specific, pre-reserved TxID - mirroring how a signature transaction + // fills a slot reserved earlier via next_txid(). + class ReservedWritePendingTx : public ccf::kv::PendingTx + { + ccf::TxID txid; + ccf::kv::Store& store; + CommitConcurrencyTable& table; + size_t key; + size_t value; + + public: + ReservedWritePendingTx( + ccf::TxID txid_, + ccf::kv::Store& store_, + CommitConcurrencyTable& table_, + size_t key_, + size_t value_) : + txid(txid_), + store(store_), + table(table_), + key(key_), + value(value_) + {} + + ccf::kv::PendingTxInfo call() override + { + auto tx = store.create_reserved_tx(txid); + tx.rw(table)->put(key, value); + return tx.commit_reserved(); + } + }; +} + +DOCTEST_TEST_CASE( + "Long-lived transaction is rolled back after a real leadership loss, and " + "TxHistory follows the Store exactly" * + doctest::test_suite("commit_concurrency_deterministic")) +{ + CommitConcurrencyFixture fixture; + const auto baseline_txid = fixture.commit_signature(); + + DOCTEST_INFO("Start applying a local transaction in the initial view"); + auto stale_tx = fixture.store->create_tx(); + stale_tx.rw(fixture.table)->put(1, 2); + + Checkpoint checkpoint("stale_tx write-set observer"); + std::optional stale_result; + std::thread stale_worker([&]() { + stale_result = stale_tx.commit( + ccf::empty_claims(), nullptr, checkpoint_write_set_observer(checkpoint)); + }); + checkpoint.wait_until_paused(); + // stale_worker is now parked inside checkpoint.pause(), and must be + // released and joined before this scope exits by any path - including a + // failed DOCTEST_REQUIRE below, which throws to unwind the test case. + // Destroying a still-joinable std::thread calls std::terminate(), + // crashing the whole test binary instead of cleanly reporting a single + // test failure, so any exception here is caught, the worker is + // released/joined, and then rethrown. + try + { + DOCTEST_REQUIRE( + stale_tx.get_txid() == + ccf::TxID(fixture.initial_view, baseline_txid.seqno + 1)); + } + catch (...) + { + checkpoint.release(); + stale_worker.join(); + throw; + } + + DOCTEST_INFO("Lose leadership after the transaction has an assigned TxID"); + fixture.step_down(); + + DOCTEST_INFO("Aft rejects the transaction and rolls the Store back"); + checkpoint.release(); + stale_worker.join(); + DOCTEST_REQUIRE(stale_result.has_value()); + DOCTEST_CHECK( + stale_result.value() == ccf::kv::CommitResult::FAIL_NO_REPLICATE); + DOCTEST_CHECK(fixture.store->current_txid() == baseline_txid); + DOCTEST_CHECK(fixture.history_txid() == baseline_txid); + DOCTEST_CHECK_FALSE(read_value(*fixture.store, fixture.table, 1).has_value()); + + DOCTEST_INFO( + "Win a later election and replicate the next transaction normally"); + fixture.raft->force_become_primary(); + const auto fresh_view = fixture.raft->get_view(); + auto fresh_tx = fixture.store->create_tx(); + fresh_tx.rw(fixture.table)->put(2, 3); + DOCTEST_REQUIRE(fresh_tx.commit() == ccf::kv::CommitResult::SUCCESS); + // Note: history_txid().view is not expected to match fresh_view here - + // see the comment on CommitConcurrencyFixture::history_txid() for why an + // ordinary in-term commit does not refresh it. Seqno agreement and + // history_term_of_next_version() are checked instead. + const auto fresh_seqno = baseline_txid.seqno + 1; + DOCTEST_CHECK( + fixture.store->current_txid() == ccf::TxID(fresh_view, fresh_seqno)); + DOCTEST_CHECK(fixture.history_txid().seqno == fresh_seqno); + DOCTEST_CHECK(fixture.history_term_of_next_version() == fresh_view); + DOCTEST_CHECK(read_value(*fixture.store, fixture.table, 2) == 3); + + DOCTEST_INFO( + "Rejecting the stale transaction did not leave anything behind to " + "clean up: every further ordinary commit keeps reaching consensus " + "immediately, with no additional election required (contrast with " + "the test below, where regaining leadership before the stale commit " + "lands currently does leave the Store unable to replicate anything " + "further until another election happens)"); + for (size_t i = 0; i < 3; ++i) + { + auto later_tx = fixture.store->create_tx(); + later_tx.rw(fixture.table)->put(i + 10, i + 10); + DOCTEST_CHECK(later_tx.commit() == ccf::kv::CommitResult::SUCCESS); + DOCTEST_CHECK(fixture.raft->get_last_idx() == fresh_seqno + i + 1); + } +} + +// NOTE_REJECTED_COMMIT_STALL: a transaction rejected by Store::commit() +// for a stale view (FAIL_NO_REPLICATE) can still leave its local write +// applied to the Store, with no corresponding entry ever reaching +// consensus. Once that has happened, every ordinary transaction +// committed afterwards can also keep succeeding locally without +// reaching consensus, until a further election restores agreement. +// Elsewhere in this suite, DOCTEST_CHECKs marked with this same tag are +// the specific assertions currently broken by this. +DOCTEST_TEST_CASE( + "Regaining leadership before a stale-view commit lands must not " + "permanently stall replication" * + doctest::test_suite("commit_concurrency_deterministic")) +{ + CommitConcurrencyFixture fixture; + const auto baseline_txid = fixture.commit_signature(); + + DOCTEST_INFO( + "Read state (fixing this transaction's commit view) in the initial " + "view"); + auto tx = fixture.store->create_tx(); + tx.rw(fixture.table)->put(0, 1); + + DOCTEST_INFO("Win a later election before assigning the transaction a TxID"); + fixture.reelect(); + DOCTEST_CHECK(fixture.store->current_txid() == baseline_txid); + DOCTEST_CHECK(fixture.history_txid() == baseline_txid); + + DOCTEST_INFO( + "A transaction whose commit view was fixed by a read in a now-stale " + "term is rejected when it reaches Store::commit()"); + DOCTEST_CHECK(tx.commit() == ccf::kv::CommitResult::FAIL_NO_REPLICATE); + + DOCTEST_INFO( + "A rejected transaction should not leave a local write behind that " + "never reaches consensus: the Store should read back exactly as it " + "did before this transaction was attempted"); + DOCTEST_CHECK( + fixture.store->current_txid() == + baseline_txid); // NOTE_REJECTED_COMMIT_STALL + DOCTEST_CHECK(fixture.history_txid() == baseline_txid); + + DOCTEST_INFO( + "Whatever the Store's state after the rejection above, every " + "ordinary transaction committed from here on must still reach " + "consensus - the Store's replicated state must never fall " + "permanently behind its own local version"); + for (size_t i = 0; i < 3; ++i) + { + auto later_tx = fixture.store->create_tx(); + later_tx.rw(fixture.table)->put(i + 1, i + 1); + DOCTEST_CHECK(later_tx.commit() == ccf::kv::CommitResult::SUCCESS); + DOCTEST_CHECK( // NOTE_REJECTED_COMMIT_STALL + fixture.raft->get_last_idx() == fixture.store->current_txid().seqno); + } + + DOCTEST_INFO( + "A further election always restores agreement between the Store, " + "TxHistory, and raft's own record of what has been replicated"); + fixture.reelect(); + auto healed_tx = fixture.store->create_tx(); + healed_tx.rw(fixture.table)->put(0, 2); + DOCTEST_CHECK(healed_tx.commit() == ccf::kv::CommitResult::SUCCESS); + DOCTEST_CHECK( + fixture.raft->get_last_idx() == fixture.store->current_txid().seqno); + DOCTEST_CHECK( + fixture.history_txid().seqno == fixture.store->current_txid().seqno); + DOCTEST_CHECK( + fixture.history_term_of_next_version() == fixture.raft->get_view()); +} + +DOCTEST_TEST_CASE( + "A stale-view commit that lands while merely a pre-vote candidate rolls " + "back cleanly too, exactly like the follower case" * + doctest::test_suite("commit_concurrency_deterministic")) +{ + CommitConcurrencyFixture fixture; + const auto baseline_txid = fixture.commit_signature(); + + auto tx = fixture.store->create_tx(); + tx.rw(fixture.table)->put(0, 1); + + DOCTEST_INFO( + "Step down to follower, then add a second (never-responding) node to " + "the configuration and let the election timeout elapse, so this node " + "becomes a pre-vote candidate on its own - still not primary, exactly " + "like the follower case above, rather than having regained " + "leadership"); + fixture.step_down(); + ccf::kv::Configuration::Nodes two_node_config; + two_node_config.try_emplace(fixture.node_id); + two_node_config.try_emplace(ccf::NodeId("NeverRespondingSecondNode")); + fixture.raft->add_configuration( + fixture.raft->get_last_idx(), two_node_config); + fixture.raft->periodic(std::chrono::milliseconds(200)); + DOCTEST_REQUIRE_FALSE(fixture.raft->is_primary()); + + DOCTEST_CHECK(tx.commit() == ccf::kv::CommitResult::FAIL_NO_REPLICATE); + DOCTEST_CHECK(fixture.store->current_txid() == baseline_txid); + DOCTEST_CHECK(fixture.history_txid() == baseline_txid); + + DOCTEST_INFO( + "As with the follower case, nothing was left behind to clean up: " + "winning the next election lets ordinary commits reach consensus " + "immediately, with no further election needed"); + fixture.raft->force_become_primary(); + auto healed_tx = fixture.store->create_tx(); + healed_tx.rw(fixture.table)->put(0, 2); + DOCTEST_CHECK(healed_tx.commit() == ccf::kv::CommitResult::SUCCESS); + DOCTEST_CHECK(healed_tx.get_txid()->seqno == baseline_txid.seqno + 1); + DOCTEST_CHECK(fixture.raft->get_last_idx() == baseline_txid.seqno + 1); +} + +DOCTEST_TEST_CASE( + "An ordinary commit immediately after a real election keeps Store and " + "TxHistory in agreement" * + doctest::test_suite("commit_concurrency_deterministic")) +{ + CommitConcurrencyFixture fixture; + const auto baseline_txid = fixture.commit_signature(); + + DOCTEST_INFO("Win a later election with no prior in-flight transaction"); + const auto reelection_view = fixture.reelect(); + DOCTEST_CHECK(fixture.store->current_txid() == baseline_txid); + DOCTEST_CHECK(fixture.history_txid() == baseline_txid); + + DOCTEST_INFO( + "A transaction reading and writing entirely in the new view commits " + "cleanly"); + auto tx = fixture.store->create_tx(); + tx.rw(fixture.table)->put(0, 1); + DOCTEST_REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + const auto committed_seqno = baseline_txid.seqno + 1; + DOCTEST_CHECK(tx.get_txid() == ccf::TxID(reelection_view, committed_seqno)); + DOCTEST_CHECK(fixture.store->current_txid().seqno == committed_seqno); + DOCTEST_CHECK(fixture.history_txid().seqno == committed_seqno); + DOCTEST_CHECK(fixture.history_term_of_next_version() == reelection_view); + DOCTEST_CHECK(read_value(*fixture.store, fixture.table, 0) == 1); +} + +// Store::commit() can batch several already-applied transactions into a +// single call to consensus, rather than replicating each one individually. +// The next two test cases check what happens when a real election lands +// partway through such a batch: TxHistory must end up exactly where the +// Store does, never ahead of it. + +DOCTEST_TEST_CASE( + "Concurrent rollback triggered by a real election during an in-flight " + "commit batch does not leave TxHistory ahead of the Store's own " + "replicated state" * + doctest::test_suite("commit_concurrency_deterministic")) +{ + // The election lands after the first entry of the batch has been applied, + // but before the second has - so the rollback below runs against a batch + // that is genuinely partway through, not one that never started. + CommitConcurrencyFixture fixture; + const auto baseline_txid = fixture.commit_signature(); + const ccf::TxID first_txid(fixture.initial_view, baseline_txid.seqno + 1); + const ccf::TxID second_txid(fixture.initial_view, baseline_txid.seqno + 2); + + DOCTEST_INFO( + "Reserve the first slot as a hole, and park the second entry behind it " + "(wrapped so it pauses on its own local application) - neither can be " + "replicated while the hole remains"); + DOCTEST_REQUIRE(fixture.store->next_txid() == first_txid); + Checkpoint checkpoint("second entry's local application"); + DOCTEST_REQUIRE( + fixture.store->commit( + second_txid, + std::make_unique( + std::make_unique( + second_txid, *fixture.store, fixture.table, 3, 4), + checkpoint), + false) == ccf::kv::CommitResult::SUCCESS); + // Nothing has been sent to consensus yet - the hole is still missing, so + // history cannot have moved past the baseline, and the pause above was + // never reached (this call returned before its batching loop, since the + // hole made it non-contiguous). + DOCTEST_CHECK(fixture.history_txid() == baseline_txid); + + DOCTEST_INFO( + "Fill the hole. This bundles [first, second] into one Store::commit() " + "batch: the first entry is applied and recorded in history for real, " + "then the second (already-queued) entry pauses on its own local " + "application, before it is recorded"); + std::optional result; + std::thread worker([&]() { + result = fixture.store->commit( + first_txid, + std::make_unique( + first_txid, *fixture.store, fixture.table, 1, 2), + false); + }); + checkpoint.wait_until_paused(); + + DOCTEST_INFO( + "Concurrently win a real election, while the worker above is still " + "paused mid-commit"); + fixture.reelect(); + DOCTEST_CHECK(fixture.store->current_txid() == baseline_txid); + + checkpoint.release(); + worker.join(); + + DOCTEST_REQUIRE(result.has_value()); + DOCTEST_INFO( + "Store::commit() correctly refuses to advance its own replicated state " + "past the rollback"); + DOCTEST_CHECK(result.value() == ccf::kv::CommitResult::FAIL_NO_REPLICATE); + DOCTEST_CHECK(fixture.store->current_txid() == baseline_txid); + + DOCTEST_INFO( + "TxHistory ends up back at the baseline too, discarding anything it " + "recorded before the rollback and never recording anything from " + "after it"); + DOCTEST_CHECK(fixture.history_txid() == baseline_txid); +} + +DOCTEST_TEST_CASE( + "Fuzz: repeated real elections against a busy writer keep TxHistory " + "consistent with the Store" * + doctest::test_suite("commit_concurrency_deterministic")) +{ + // Broader, randomised complement to the pinned test above. One thread + // continually commits new ordinary transactions (so Store::commit()'s + // batching loop is usually short, but with enough of them in flight to + // create many small windows for a race), while another thread repeatedly + // wins a fresh real election - mimicking a raft node that keeps losing and + // regaining leadership, discarding all of its own unreplicated writes + // every time. Because no further signature is emitted during the fuzzing, + // the one committed at the start remains a permanently-safe rollback + // target throughout (Store::commit() can never let last_replicated fall + // below a seqno it has itself successfully replicated), so the final + // state is fully deterministic regardless of how the two threads + // interleaved. + // + // This currently exercises NOTE_IS_PRIMARY_RACE (see + // CommitConcurrencyFixture::reelect() in fixture.h). Expect this test to fail + // occasionally, or to abort the whole process under ThreadSanitizer, + // until that race is fixed. + CommitConcurrencyFixture fixture; + const auto baseline_txid = fixture.commit_signature(); + + constexpr size_t reelection_iterations = 300; + std::atomic stop{false}; + + std::thread writer([&]() { + size_t i = 0; + while (!stop) + { + auto tx = fixture.store->create_tx(); + tx.rw(fixture.table)->put(i, i); + // Any result is acceptable here - conflicts and rollback-induced + // failures are expected and simply retried with a fresh transaction. + tx.commit(); + i++; + } + }); + + std::thread election_churn([&]() { + std::mt19937 rng(42); + for (size_t i = 0; i < reelection_iterations; ++i) + { + random_delay(rng, std::chrono::microseconds(200)); + fixture.reelect(); + } + stop = true; + }); + + writer.join(); + election_churn.join(); + + DOCTEST_INFO( + "After all concurrent activity has stopped, one final, fully " + "deterministic election settles the Store at the permanently-safe " + "baseline used throughout this fuzz run"); + fixture.reelect(); + + const auto final_txid = fixture.store->current_txid(); + DOCTEST_CHECK(final_txid == baseline_txid); + DOCTEST_INFO( + "TxHistory's own record of what has been replicated must exactly match " + "this final, deterministic state - never ahead (which would mean " + "history recorded entries that were actually rolled back or never " + "truly committed) and never behind"); + DOCTEST_CHECK(fixture.history_txid() == final_txid); +} diff --git a/src/commit_concurrency/threaded/fixture.h b/src/commit_concurrency/threaded/fixture.h new file mode 100644 index 000000000000..90d99fce6aae --- /dev/null +++ b/src/commit_concurrency/threaded/fixture.h @@ -0,0 +1,215 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#pragma once + +// A harness combining a real ccf::kv::Store, a real +// aft::Aft (raft consensus), and a real +// ccf::MerkleTxHistory, for tests that exercise how these three components +// interact under real concurrency. Other unit tests exercise each of these +// components in isolation, with lighter-weight stubs standing in for the +// others. +// +// LedgerStubProxy and ChannelStubProxy remain stubs here: they are the +// host-disk and network I/O boundaries, not part of what this suite tests. + +#include "ccf/crypto/ec_key_pair.h" +#include "ccf/ds/unit_strings.h" +#include "ccf/ds/x509_time_fmt.h" +#include "ccf/service/consensus_config.h" +#include "commit_concurrency/interleaving.h" +#include "consensus/aft/raft.h" +#include "consensus/aft/test/logging_stub.h" +#include "crypto/certs.h" +#include "crypto/openssl/ec_key_pair.h" +#include "kv/store.h" +#include "kv/test/null_encryptor.h" +#include "kv/test/stub_consensus.h" +#include "node/encryptor.h" +#include "node/history.h" +#include "node/ledger_secrets.h" + +#include +#include +#include + +namespace ccf::kv::test +{ + using CommitConcurrencyRaft = aft::Aft; + using CommitConcurrencyTable = ccf::kv::Map; + + inline const ccf::consensus::Configuration& commit_concurrency_raft_settings() + { + static const ccf::consensus::Configuration settings{ + ccf::ds::TimeString{"10ms"}, ccf::ds::TimeString{"100ms"}, 0}; + return settings; + } + + inline std::optional read_value( + ccf::kv::Store& store, CommitConcurrencyTable& table, size_t key) + { + auto tx = store.create_read_only_tx(); + return tx.ro(table)->get(key); + } + + // A harness combining the real stack described above, plus helpers for + // driving genuine raft view changes (which in turn trigger genuine + // Store::rollback() calls, exactly as a production election would). + struct CommitConcurrencyFixture + { + const ccf::NodeId node_id = ccf::kv::test::PrimaryNodeId; + // Used only as the notional sender of the fake RequestVote messages + // step_down() constructs below - never actually configured as a real + // peer. + const ccf::NodeId phantom_peer = + ccf::NodeId("CommitConcurrencyFixturePhantomPeer"); + std::shared_ptr node_kp = + ccf::crypto::make_ec_key_pair(); + std::shared_ptr service_kp = + std::dynamic_pointer_cast( + ccf::crypto::make_ec_key_pair()); + std::shared_ptr store = std::make_shared(); + std::shared_ptr history; + std::shared_ptr raft; + CommitConcurrencyTable table{"public:table"}; + ccf::View initial_view = 0; + + // use_real_crypto selects between NullTxEncryptor (default: fast enough + // for a tight fuzzing loop) and a real ccf::NodeEncryptor (slower, but + // exercises real AES-GCM IV/nonce derivation - relevant to catching + // nonce-reuse-across-rollback style bugs that NullTxEncryptor cannot). + explicit CommitConcurrencyFixture(bool use_real_crypto = false) + { + if (use_real_crypto) + { + auto secrets = std::make_shared(); + secrets->init(); + store->set_encryptor(std::make_shared(secrets)); + } + else + { + store->set_encryptor(std::make_shared()); + } + + history = + std::make_shared(*store, node_id, *node_kp); + + // Set up a signing identity so that commit_signature() below can + // later emit a real signature transaction. + constexpr size_t certificate_validity_period_days = 365; + const auto valid_from = ccf::ds::to_x509_time_string( + std::chrono::system_clock::now() - std::chrono::hours(24)); + const auto valid_to = ccf::crypto::compute_cert_valid_to_string( + valid_from, certificate_validity_period_days); + const auto self_signed = + node_kp->self_sign("CN=Node", valid_from, valid_to); + history->set_endorsed_certificate(self_signed); + history->set_service_signing_identity( + service_kp, ccf::COSESignaturesConfig{}); + store->set_history(history); + + raft = std::make_shared( + commit_concurrency_raft_settings(), + std::make_unique>(store), + std::make_unique(node_id), + std::make_shared(), + std::make_shared(node_id), + nullptr); + store->set_consensus(raft); + + ccf::kv::Configuration::Nodes configuration; + configuration.try_emplace(node_id); + raft->add_configuration(0, configuration); + raft->force_become_primary(); + initial_view = raft->get_view(); + } + + // Makes this node aware of a higher term, safely, from any thread. + // + // Aft::become_aware_of_new_term() assumes its caller already holds + // Aft's own (private) state lock, so calling it directly here would + // race against another thread's concurrent Store::commit() -> + // replicate(). recv_message() is Aft's self-locked public entry point + // for this instead, so this constructs a minimal RequestVote from an + // unconfigured phantom peer and delivers it through that path - as a + // real node would learn of a higher term from a real peer. + // term_of_last_committable_idx is set to the new term, which always + // beats this node's own (never advanced after setup), so the vote is + // granted and leadership is relinquished before force_become_primary() + // is next called. + void step_down() + { + const auto next_term = raft->get_view() + 1; + aft::RequestVote rv; + rv.term = next_term; + rv.term_of_last_committable_idx = next_term; + rv.last_committable_idx = 0; + raft->recv_message( + phantom_peer, reinterpret_cast(&rv), sizeof(rv)); + } + + // Loses leadership (rolling back any uncommitted local writes, as a real + // node would when it discovers a higher term) and then wins the next + // election. Returns the new view. + // + // NOTE_IS_PRIMARY_RACE: calling this concurrently with a writer thread + // committing on the same fixture exercises a real, pre-existing data + // race - a transaction reads its own leadership status while this call + // changes it, with no synchronisation between the two. This is + // undefined behaviour: usually tolerated silently by a plain build, but + // reliably caught (and turned into a process abort) by ThreadSanitizer. + // Test cases that exercise this are expected to fail, or abort under + // TSAN, until that race is fixed. + ccf::View reelect() + { + step_down(); + raft->force_become_primary(); + return raft->get_view(); + } + + // Emits a real signature transaction, which - like production CCF's + // periodic signature emission - is the mechanism that marks the current + // point globally committable, letting raft's own commit index advance + // past it. Returns the TxID of the signature transaction itself. + ccf::TxID commit_signature() + { + const auto before = store->current_txid(); + history->emit_signature(); + const auto after = store->current_txid(); + if (after.seqno == before.seqno) + { + throw std::logic_error("emit_signature() did not advance the store"); + } + return after; + } + + // TxHistory's own idea of the last TxID it has recorded. + // + // The returned TxID's view is only refreshed by rollback()/set_term(), + // not by every append_entry() call, so it matches + // store->current_txid().view only immediately after a rollback, before + // any further commit in the new term. For an ordinary in-term commit, + // compare seqnos only (see history_term_of_next_version() below for + // the current term). + ccf::TxID history_txid() + { + auto [txid, root, term_of_next_version] = + history->get_replicated_state_txid_and_root(); + (void)root; + (void)term_of_next_version; + return txid; + } + + // TxHistory's own idea of the current term (i.e. the term new entries + // are expected to be appended in) - the third element of + // get_replicated_state_txid_and_root(), tracked and used independently + // of the TxID's own .view (see history_txid() above). + ccf::kv::Term history_term_of_next_version() + { + auto [txid, root, term_of_next_version] = + history->get_replicated_state_txid_and_root(); + (void)txid; + (void)root; + return term_of_next_version; + } + }; +} diff --git a/src/commit_concurrency/threaded/fuzzer.cpp b/src/commit_concurrency/threaded/fuzzer.cpp new file mode 100644 index 000000000000..7a2bed96feea --- /dev/null +++ b/src/commit_concurrency/threaded/fuzzer.cpp @@ -0,0 +1,261 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#include "commit_concurrency/threaded/fixture.h" + +#define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// The randomised, multi-actor complement to deterministic.cpp's pinned +// scenarios. Drives a real Store + real Aft + real MerkleTxHistory +// (CommitConcurrencyFixture) with: +// - N writer threads, each committing ordinary transactions in a loop. +// - One election-churn actor, repeatedly winning a fresh real election via +// CommitConcurrencyFixture::reelect(). +// - One reader thread, continuously polling +// TxHistory::get_replicated_state_txid_and_root() and +// Store::current_txid() concurrently, checking this suite's core +// invariants on every poll. +// +// All randomness is drawn from a single seed (overridable via the RNG_SEED +// environment variable), logged unconditionally so any CI failure is +// re-runnable with the same seed. A real-OS-thread fuzzer is not +// byte-for-byte replayable purely from a seed - actual thread scheduling +// still varies run to run - so "reproducible" here means the same seed +// reliably exercises the same kind of interleaving, not an identical trace. +// +// The invariant checks below read two pieces of state that are each +// updated independently, with no shared synchronisation between the two +// reads that make up each check. Each check therefore reads its +// "reference" value both before and after the other read, with a short +// sleep in between, and only trusts the comparison if that reference value +// was unchanged across the whole window - this keeps the false-positive +// rate from this kind of read-only race negligible, without requiring any +// change to production code. + +using namespace ccf::kv::test; + +namespace +{ + uint32_t pick_seed() + { + if (const char* env = std::getenv("RNG_SEED")) + { + std::string rng_seed(env); + uint32_t seed = 0; + std::from_chars(rng_seed.data(), rng_seed.data() + rng_seed.size(), seed); + if (seed != 0) + { + return seed; + } + } + return std::random_device{}(); + } + + // Accumulates the first invariant violation found by the reader thread, + // if any. Checked continuously (not just at the end) - see this file's + // top comment. + class InvariantViolations + { + std::mutex lock; + std::optional first; + + public: + void record(const std::string& msg) + { + std::lock_guard guard(lock); + if (!first.has_value()) + { + first = msg; + } + } + + std::optional get() + { + std::lock_guard guard(lock); + return first; + } + }; + + struct FuzzConfig + { + size_t num_writers = 4; + size_t writer_iterations = 150; + size_t reelection_iterations = 60; + std::chrono::microseconds max_writer_delay{100}; + std::chrono::microseconds max_reelection_delay{500}; + bool use_real_crypto = false; + }; + + void run_fuzz(uint32_t seed, const FuzzConfig& cfg) + { + fmt::println( + "commit_concurrency fuzzer seed: {} (rerun with RNG_SEED={} to " + "reproduce)", + seed, + seed); + std::mt19937 seed_rng(seed); + + CommitConcurrencyFixture fixture(cfg.use_real_crypto); + const auto baseline_txid = fixture.commit_signature(); + + InvariantViolations violations; + std::atomic stop{false}; + + // Reader actor: continuously polls TxHistory and the Store concurrently + // and checks that they agree. + std::thread reader([&]() { + while (!stop.load()) + { + // See this file's top comment for why each check below reads its + // "reference" value both before and after the other side, with a + // short sleep in between. + const auto store_txid_before = fixture.store->current_txid(); + const auto history_txid = fixture.history_txid(); + std::this_thread::sleep_for(std::chrono::microseconds(20)); + const auto store_txid_after = fixture.store->current_txid(); + if ( + store_txid_before == store_txid_after && + history_txid.seqno > store_txid_after.seqno) + { + violations.record(fmt::format( + "TxHistory reports seqno {} ahead of Store's own current_txid " + "seqno {} (history TxID {}, store TxID {})", + history_txid.seqno, + store_txid_after.seqno, + history_txid.to_str(), + store_txid_after.to_str())); + } + + // history_term_of_next_version() (unlike history_txid().view - see + // the comment on CommitConcurrencyFixture::history_txid()) is refreshed + // on every rollback() to whatever term Aft passes at that moment, so it + // must never be ahead of Aft's own current view. It can legitimately + // lag transiently, since reelect() is two steps: a message bumping + // Aft's view, then a separate call that performs the rollback syncing + // history to it. + const auto raft_view_before = fixture.raft->get_view(); + const auto history_current_view = + fixture.history_term_of_next_version(); + std::this_thread::sleep_for(std::chrono::microseconds(20)); + const auto raft_view_after = fixture.raft->get_view(); + if ( + raft_view_before == raft_view_after && + history_current_view > raft_view_after) + { + violations.record(fmt::format( + "TxHistory's own idea of the current term ({}) is ahead of " + "Aft's own current view ({})", + history_current_view, + raft_view_after)); + } + } + }); + + std::vector writers; + writers.reserve(cfg.num_writers); + for (size_t w = 0; w < cfg.num_writers; ++w) + { + const uint32_t writer_seed = seed_rng(); + writers.emplace_back([&fixture, &cfg, w, writer_seed]() { + std::mt19937 rng(writer_seed); + for (size_t i = 0; i < cfg.writer_iterations; ++i) + { + random_delay(rng, cfg.max_writer_delay); + auto tx = fixture.store->create_tx(); + tx.rw(fixture.table)->put((w * 1'000'000) + i, i); + // Any result is acceptable here - conflicts and rollback-induced + // failures are expected and simply retried with a fresh + // transaction on the next iteration. + tx.commit(); + } + }); + } + + const uint32_t churn_seed = seed_rng(); + std::thread election_churn([&fixture, &cfg, churn_seed]() { + std::mt19937 rng(churn_seed); + for (size_t i = 0; i < cfg.reelection_iterations; ++i) + { + random_delay(rng, cfg.max_reelection_delay); + fixture.reelect(); + } + }); + + for (auto& w : writers) + { + w.join(); + } + election_churn.join(); + + // Stop the reader only once all mutating actors are done, then take one + // final poll before it exits. + stop = true; + reader.join(); + + const auto mid_run_violation = violations.get(); + DOCTEST_INFO(fmt::format("Seed was {}", seed)); + DOCTEST_REQUIRE_MESSAGE( + !mid_run_violation.has_value(), mid_run_violation.value_or("")); + + DOCTEST_INFO( + "After all actors quiesce, one final, fully deterministic election " + "settles the Store at the permanently-safe baseline used throughout " + "this fuzz run (no further signature was emitted during the fuzzing, " + "so the one committed at the start remains the only globally " + "committable index, and every election - including this final one - " + "rolls back to it)"); + fixture.reelect(); + + const auto final_txid = fixture.store->current_txid(); + DOCTEST_INFO(fmt::format("Seed was {}", seed)); + DOCTEST_CHECK(final_txid == baseline_txid); + DOCTEST_CHECK(fixture.history_txid() == final_txid); + DOCTEST_CHECK(fixture.raft->get_committed_seqno() == final_txid.seqno); + DOCTEST_CHECK(fixture.raft->get_view(final_txid.seqno) == final_txid.view); + } +} + +DOCTEST_TEST_CASE( + "Fuzz: concurrent writers, election churn, and a continuous reader keep " + "TxHistory consistent with the Store (fast, NullTxEncryptor)" * + doctest::test_suite("commit_concurrency_fuzz")) +{ + // The writer threads and election_churn thread spawned by run_fuzz() + // below run fully concurrently with no synchronisation between them, so + // this currently exercises NOTE_IS_PRIMARY_RACE (see + // CommitConcurrencyFixture::reelect() in fixture.h). Expect this test to fail + // occasionally, or to abort the whole process under ThreadSanitizer, + // until that race is fixed. + run_fuzz(pick_seed(), FuzzConfig{}); +} + +DOCTEST_TEST_CASE( + "Soak: as above, with real crypto and more iterations" * + doctest::test_suite("commit_concurrency_fuzz_soak")) +{ + // See NOTE_IS_PRIMARY_RACE (fixture.h) - applies here too. + if (std::getenv("REAL_STACK_SOAK") == nullptr) + { + DOCTEST_MESSAGE( + "Skipping soak variant - set REAL_STACK_SOAK=1 to run it (real " + "AES-GCM encryption per transaction, and more iterations, so this is " + "deliberately not part of the default fast test run)"); + return; + } + + FuzzConfig cfg; + cfg.use_real_crypto = true; + cfg.num_writers = 8; + cfg.writer_iterations = 500; + cfg.reelection_iterations = 200; + run_fuzz(pick_seed(), cfg); +} diff --git a/src/commit_concurrency/threaded/main.cpp b/src/commit_concurrency/threaded/main.cpp new file mode 100644 index 000000000000..01dd621c24a4 --- /dev/null +++ b/src/commit_concurrency/threaded/main.cpp @@ -0,0 +1,18 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. + +// Doctest entry point for the commit-concurrency suite: real OS threads +// exercising a real Store, Aft, and MerkleTxHistory together. See +// fixture.h for the harness, and deterministic.cpp/fuzzer.cpp for what +// each covers. + +#define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES +#define DOCTEST_CONFIG_IMPLEMENT +#include + +int main(int argc, char** argv) +{ + doctest::Context context; + context.applyCommandLine(argc, argv); + return context.run(); +} diff --git a/src/commit_concurrency/threaded/smoke.cpp b/src/commit_concurrency/threaded/smoke.cpp new file mode 100644 index 000000000000..3db1ea6988bd --- /dev/null +++ b/src/commit_concurrency/threaded/smoke.cpp @@ -0,0 +1,51 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#include "commit_concurrency/threaded/fixture.h" + +#define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES +#include + +// Sanity checks for CommitConcurrencyFixture itself, with no concurrency at +// all: establishes that the real Store + real Aft + real MerkleTxHistory wiring +// behaves as expected before any interleaving is layered on top. + +DOCTEST_TEST_CASE( + "CommitConcurrencyFixture wires a real Store, Aft, and MerkleTxHistory in " + "agreement" * + doctest::test_suite("commit_concurrency_smoke")) +{ + ccf::kv::test::CommitConcurrencyFixture fixture; + + DOCTEST_REQUIRE(fixture.raft->is_primary()); + DOCTEST_REQUIRE(fixture.store->current_txid() == ccf::TxID(0, 0)); + + DOCTEST_INFO("Commit a handful of ordinary transactions"); + for (size_t i = 0; i < 5; ++i) + { + auto tx = fixture.store->create_tx(); + tx.rw(fixture.table)->put(i, i * 10); + DOCTEST_REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + + const auto store_txid = fixture.store->current_txid(); + DOCTEST_CHECK(store_txid == ccf::TxID(fixture.initial_view, 5)); + DOCTEST_CHECK(fixture.raft->get_last_idx() == 5); + DOCTEST_CHECK(fixture.history_txid() == store_txid); + + for (size_t i = 0; i < 5; ++i) + { + DOCTEST_CHECK( + ccf::kv::test::read_value(*fixture.store, fixture.table, i) == i * 10); + } + + DOCTEST_INFO( + "Nothing is committed (in the raft sense) until a signature marks a " + "point as globally committable - exactly as in production"); + DOCTEST_CHECK(fixture.raft->get_committed_seqno() == 0); + + DOCTEST_INFO("Emitting a real signature transaction advances commit_idx"); + const auto sig_txid = fixture.commit_signature(); + DOCTEST_CHECK(sig_txid == ccf::TxID(fixture.initial_view, 6)); + DOCTEST_CHECK(fixture.raft->get_committed_seqno() == 6); + DOCTEST_CHECK(fixture.history_txid() == sig_txid); +} diff --git a/src/consensus/aft/impl/state.h b/src/consensus/aft/impl/state.h index 248cb34ab149..905dda0115e2 100644 --- a/src/consensus/aft/impl/state.h +++ b/src/consensus/aft/impl/state.h @@ -2,6 +2,11 @@ // Licensed under the Apache 2.0 License. #pragma once +#if defined(CCF_STATIC_LIBRARY_BUILD) +# error \ + "consensus/aft/impl/state.h must never be compiled into ccf_kv, ccf_tasks, or ccfcrypto: their compiled objects are linked, unmodified, by every test binary, including one that recompiles this header with a different ccf::pal::Mutex (see src/commit_concurrency/interleaving_lock_override.h) - two different memory layouts for the same class name in one binary would be an ODR violation." +#endif + #include "ccf/crypto/verifier.h" #include "ccf/pal/locking.h" #include "ccf/tx_status.h" diff --git a/src/consensus/aft/raft.h b/src/consensus/aft/raft.h index 9056c818e5ff..d656a2bf7f91 100644 --- a/src/consensus/aft/raft.h +++ b/src/consensus/aft/raft.h @@ -2,6 +2,11 @@ // Licensed under the Apache 2.0 License. #pragma once +#if defined(CCF_STATIC_LIBRARY_BUILD) +# error \ + "consensus/aft/raft.h must never be compiled into ccf_kv, ccf_tasks, or ccfcrypto: their compiled objects are linked, unmodified, by every test binary, including one that recompiles this header with a different ccf::pal::Mutex (see src/commit_concurrency/interleaving_lock_override.h) - two different memory layouts for the same class name in one binary would be an ODR violation." +#endif + #include "ccf/pal/locking.h" #include "ccf/service/reconfiguration_type.h" #include "ccf/tx_id.h" @@ -269,7 +274,7 @@ namespace aft bool can_replicate() override { - std::unique_lock guard(state->lock); + ccf::pal::unique_lock guard(state->lock); return can_replicate_unsafe(); } @@ -284,14 +289,14 @@ namespace aft { return false; } - std::unique_lock guard(state->lock); + ccf::pal::unique_lock guard(state->lock); return state->leadership_state == ccf::kv::LeadershipState::Leader && (state->last_idx - state->commit_idx >= max_uncommitted_tx_count); } Consensus::SignatureDisposition get_signature_disposition() override { - std::unique_lock guard(state->lock); + ccf::pal::unique_lock guard(state->lock); if (can_sign_unsafe()) { if (should_sign) @@ -396,7 +401,7 @@ namespace aft { // When receiving append entries as a follower, all security domains will // be deserialised - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); public_only = false; } @@ -410,7 +415,8 @@ namespace aft "Can't force leadership if there is already a leader"); } - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard( + state->lock, "force this node to become primary"); state->current_view += starting_view_change; become_leader(true); } @@ -429,7 +435,8 @@ namespace aft "Can't force leadership if there is already a leader"); } - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard( + state->lock, "force this node to become primary from a known index"); state->current_view = term; state->last_idx = index; state->commit_idx = commit_idx_; @@ -447,7 +454,7 @@ namespace aft { // This should only be called when the node resumes from a snapshot and // before it has received any append entries. - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); state->last_idx = index; state->commit_idx = index; @@ -466,26 +473,26 @@ namespace aft Index get_committed_seqno() override { - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); return get_commit_idx_unsafe(); } Term get_view() override { - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); return state->current_view; } std::pair get_committed_txid() override { - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); ccf::SeqNo commit_idx = get_commit_idx_unsafe(); return {get_term_internal(commit_idx), commit_idx}; } Term get_view(Index idx) override { - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); return get_term_internal(idx); } @@ -591,14 +598,14 @@ namespace aft Configuration::Nodes get_latest_configuration() override { - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); return get_latest_configuration_unsafe(); } ccf::kv::ConsensusDetails get_details() override { ccf::kv::ConsensusDetails details; - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); details.primary_id = leader_id; details.current_view = state->current_view; details.ticking = ticking; @@ -623,7 +630,7 @@ namespace aft bool replicate(const ccf::kv::BatchVector& entries, Term term) override { - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); if (state->leadership_state != ccf::kv::LeadershipState::Leader) { @@ -834,7 +841,7 @@ namespace aft void periodic(std::chrono::milliseconds elapsed) override { - std::unique_lock guard(state->lock); + ccf::pal::unique_lock guard(state->lock); timeout_elapsed += elapsed; if (state->leadership_state == ccf::kv::LeadershipState::Leader) @@ -1107,7 +1114,7 @@ namespace aft const uint8_t* data, size_t size) { - std::unique_lock guard(state->lock); + ccf::pal::unique_lock guard(state->lock); RAFT_DEBUG_FMT( "Recv {} to {} from {}: {}.{} to {}.{} in term {}", @@ -1581,7 +1588,7 @@ namespace aft void recv_append_entries_response( const ccf::NodeId& from, AppendEntriesResponse r) { - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); auto node = all_other_nodes.find(from); if (node == all_other_nodes.end()) @@ -1861,7 +1868,7 @@ namespace aft void recv_request_vote(const ccf::NodeId& from, RequestVote r) { - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); #ifdef CCF_RAFT_TRACING nlohmann::json j = {}; @@ -1878,7 +1885,7 @@ namespace aft void recv_request_pre_vote(const ccf::NodeId& from, RequestPreVote r) { - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); #ifdef CCF_RAFT_TRACING nlohmann::json j = {}; @@ -1941,7 +1948,7 @@ namespace aft RequestVoteResponse r, ElectionType election_type) { - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); #ifdef CCF_RAFT_TRACING nlohmann::json j = {}; @@ -2067,7 +2074,7 @@ namespace aft void recv_propose_request_vote( const ccf::NodeId& from, ProposeRequestVote r) { - std::lock_guard guard(state->lock); + ccf::pal::unique_lock guard(state->lock); #ifdef CCF_RAFT_TRACING nlohmann::json j = {}; diff --git a/src/kv/store.h b/src/kv/store.h index 2a4175280836..c36fd4480e2a 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -2,6 +2,11 @@ // Licensed under the Apache 2.0 License. #pragma once +#if defined(CCF_STATIC_LIBRARY_BUILD) +# error \ + "kv/store.h must never be compiled into ccf_kv, ccf_tasks, or ccfcrypto: their compiled objects are linked, unmodified, by every test binary, including one that recompiles this header with a different ccf::pal::Mutex (see src/commit_concurrency/interleaving_lock_override.h) - two different memory layouts for the same class name in one binary would be an ODR violation." +#endif + #include "apply_changes.h" #include "ccf/kv/read_only_store.h" #include "ccf/pal/locking.h" @@ -133,7 +138,8 @@ namespace ccf::kv ccf::kv::ConsensusHookPtrs& hooks, bool track_deletes_on_missing_keys) override { - std::unique_lock maps_guard(maps_lock, std::defer_lock); + ccf::pal::unique_lock maps_guard( + maps_lock, std::defer_lock); if (!new_maps.empty()) { maps_guard.lock(); @@ -152,7 +158,7 @@ namespace ccf::kv return false; } { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); version = v; last_replicated = version; term_of_last_version = term; @@ -298,7 +304,7 @@ namespace ccf::kv std::shared_ptr get_map( ccf::kv::Version v, const std::string& map_name) override { - std::lock_guard mguard(maps_lock); + ccf::pal::unique_lock mguard(maps_lock); return get_map_internal(v, map_name); } @@ -473,7 +479,7 @@ namespace ccf::kv std::vector hash_at_snapshot; std::vector view_history_; { - std::lock_guard mguard(maps_lock); + ccf::pal::unique_lock mguard(maps_lock); for (auto& it : maps) { @@ -570,7 +576,7 @@ namespace ccf::kv } { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); version = v; last_replicated = v; } @@ -610,7 +616,7 @@ namespace ccf::kv chunker->compacted_to(v); } - std::lock_guard mguard(maps_lock); + ccf::pal::unique_lock mguard(maps_lock); if (v > current_version()) { @@ -636,7 +642,7 @@ namespace ccf::kv } { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); compacted = v; auto h = get_history(); @@ -669,10 +675,11 @@ namespace ccf::kv chunker->rolled_back_to(tx_id.seqno); } - std::lock_guard mguard(maps_lock); + ccf::pal::unique_lock mguard(maps_lock); { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard( + version_lock, "roll version and history back to tx_id"); if (tx_id.seqno < compacted) { throw std::logic_error(fmt::format( @@ -751,7 +758,7 @@ namespace ccf::kv { // Note: This should only be called once, when the store is first // initialised. term_of_next_version is later updated via rollback. - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); if (term_of_next_version != 0) { throw std::logic_error("term_of_next_version is already initialised"); @@ -832,7 +839,7 @@ namespace ccf::kv // rather than with the actual value read. As a result, they don't // need snapshot isolation on the map state, and so do not need to // lock each of the maps before creating the transaction. - std::lock_guard mguard(maps_lock); + ccf::pal::unique_lock mguard(maps_lock); for (auto r = d.start_map(); r.has_value(); r = d.start_map()) { @@ -935,14 +942,14 @@ namespace ccf::kv ccf::TxID current_txid() override { // Must lock in case the version or read term is being incremented. - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); return current_txid_unsafe(); } std::pair current_txid_and_commit_term() override { // Must lock in case the version or commit term is being incremented. - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); return {current_txid_unsafe(), term_of_next_version}; } @@ -968,7 +975,8 @@ namespace ccf::kv return CommitResult::SUCCESS; } - std::lock_guard cguard(commit_lock); + ccf::pal::unique_lock cguard( + commit_lock, "serialise concurrent Store::commit() calls"); LOG_DEBUG_FMT( "Store::commit {}{}", @@ -986,7 +994,9 @@ namespace ccf::kv auto h = get_history(); { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard( + version_lock, + "assign version and enqueue pending tx for replication"); if (txid.view != term_of_next_version && get_consensus()->is_primary()) { // This can happen when a transaction started before a view change, @@ -1104,7 +1114,8 @@ namespace ccf::kv if (c->replicate(batch, replication_view)) { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard( + version_lock, "advance last_replicated after successful replicate()"); if ( last_replicated == previous_last_replicated && previous_rollback_count == rollback_count) @@ -1120,7 +1131,7 @@ namespace ccf::kv bool should_schedule_snapshot() { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); if (snapshotter) { return snapshotter->should_schedule_snapshot(last_committable); @@ -1130,7 +1141,7 @@ namespace ccf::kv bool should_create_ledger_chunk(Version version) override { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); return should_create_ledger_chunk_unsafe(version); } @@ -1172,13 +1183,13 @@ namespace ccf::kv bool check_rollback_count(Version count) override { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); return rollback_count == count; } std::tuple next_version(bool commit_new_map) override { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); Version v = next_version_unsafe(); auto previous_last_new_map = last_new_map; @@ -1192,13 +1203,13 @@ namespace ccf::kv Version next_version() override { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); return next_version_unsafe(); } TxID next_txid() override { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); next_version_unsafe(); return {term_of_next_version, version}; @@ -1206,7 +1217,7 @@ namespace ccf::kv size_t committable_gap() override { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); return version - last_committable; } @@ -1377,25 +1388,25 @@ namespace ccf::kv ReservedTx create_reserved_tx(const TxID& tx_id) { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); return {this, term_of_last_version, tx_id, rollback_count}; } void set_flag(StoreFlag f) override { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); set_flag_unsafe(f); } void unset_flag(StoreFlag f) override { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); unset_flag_unsafe(f); } bool flag_enabled(StoreFlag f) override { - std::lock_guard vguard(version_lock); + ccf::pal::unique_lock vguard(version_lock); return flag_enabled_unsafe(f); } diff --git a/src/node/history.h b/src/node/history.h index db80959d33e6..b006ac2ce6b8 100644 --- a/src/node/history.h +++ b/src/node/history.h @@ -2,6 +2,11 @@ // Licensed under the Apache 2.0 License. #pragma once +#if defined(CCF_STATIC_LIBRARY_BUILD) +# error \ + "node/history.h must never be compiled into ccf_kv, ccf_tasks, or ccfcrypto: their compiled objects are linked, unmodified, by every test binary, including one that recompiles this header with a different ccf::pal::Mutex (see src/commit_concurrency/interleaving_lock_override.h) - two different memory layouts for the same class name in one binary would be an ODR violation." +#endif + #include "ccf/crypto/cose_verifier.h" #include "ccf/ds/x509_time_fmt.h" #include "ccf/node/ledger_sign_mode.h" @@ -650,8 +655,8 @@ namespace ccf const auto delay = std::chrono::milliseconds(sig_ms_interval); emit_signature_periodic_task = ccf::tasks::make_basic_task([this]() { - std::unique_lock mguard( - this->signature_lock, std::defer_lock); + ccf::pal::unique_lock mguard( + this->signature_lock, std::defer_lock, "periodic signature emission"); bool should_emit_signature = false; @@ -734,7 +739,7 @@ namespace ccf // Delay taking this lock until _after_ the read above, to avoid lock // inversions - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); CCF_ASSERT_FMT( !replicated_state_tree.in_range(1), @@ -753,14 +758,14 @@ namespace ccf ccf::crypto::Sha256Hash get_replicated_state_root() override { - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); return replicated_state_tree.get_root(); } std::tuple get_replicated_state_txid_and_root() override { - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); return { {term_of_last_version, static_cast(replicated_state_tree.end_index())}, @@ -875,7 +880,7 @@ namespace ccf std::vector serialise_tree(size_t to) override { - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); if (to <= replicated_state_tree.end_index()) { return replicated_state_tree.serialise( @@ -889,7 +894,7 @@ namespace ccf { // This should only be called once, when the store first knows about its // term - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); term_of_last_version = t; term_of_next_version = t; } @@ -897,7 +902,7 @@ namespace ccf void rollback( const ccf::TxID& tx_id, ccf::kv::Term term_of_next_version_) override { - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); LOG_TRACE_FMT("Rollback to {}.{}", tx_id.view, tx_id.seqno); term_of_last_version = tx_id.view; term_of_next_version = term_of_next_version_; @@ -907,7 +912,7 @@ namespace ccf void compact(ccf::kv::Version v) override { - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); // Receipts can only be retrieved to the flushed index. Keep a range of // history so that a range of receipts are available. if (v > MAX_HISTORY_LEN) @@ -921,7 +926,8 @@ namespace ccf void try_emit_signature() override { - std::unique_lock mguard(signature_lock, std::defer_lock); + ccf::pal::unique_lock mguard( + signature_lock, std::defer_lock, "on-demand signature emission"); if (store.committable_gap() < sig_tx_interval || !mguard.try_lock()) { return; @@ -977,20 +983,20 @@ namespace ccf std::vector get_proof(ccf::kv::Version index) override { - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); return replicated_state_tree.get_proof(index).to_v(); } bool verify_proof(const std::vector& v) override { Proof proof(v); - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); return replicated_state_tree.verify(proof); } std::vector get_raw_leaf(uint64_t index) override { - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); auto leaf = replicated_state_tree.get_leaf(index); return {leaf.h.begin(), leaf.h.end()}; } @@ -999,7 +1005,7 @@ namespace ccf { ccf::crypto::Sha256Hash rh(data); log_hash(rh, APPEND); - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); replicated_state_tree.append(rh); } @@ -1009,7 +1015,7 @@ namespace ccf std::nullopt) override { log_hash(digest, APPEND); - std::lock_guard guard(state_lock); + ccf::pal::unique_lock guard(state_lock); if (expected_term_of_next_version.has_value()) { if (expected_term_of_next_version.value() != term_of_next_version)