diff --git a/CMakeLists.txt b/CMakeLists.txt index dee27f1f816..69646495afa 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -727,6 +727,73 @@ 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 checkpoint primitive itself + # (src/commit_concurrency/threaded/checkpoint.h) could deadlock if buggy. + add_unit_test( + commit_concurrency_test + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/threaded/main.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/threaded/checkpoint_test.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 + ) + + # Systematically explores the space of legal interleavings of a bounded + # scenario (rather than sampling timing-dependent ones, as + # commit_concurrency_test does), via ccf::kv::test::DeterministicScheduler + # in src/commit_concurrency/scheduled/deterministic_scheduler.h - + # exhaustively where that space is small enough + # (explore_all_interleavings()), or by random sampling where it isn't + # (explore_random_interleavings()). DETECT_DEADLOCKS is passed for the + # same reason as above. + add_unit_test( + commit_concurrency_scheduled_test + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/scheduled/deterministic_scheduler_test.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/scheduled/main.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/scheduled/pthread_mutex_wrap.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/src/commit_concurrency/scheduled/rejected_commit.cpp + DETECT_DEADLOCKS + ) + set_property( + TEST commit_concurrency_scheduled_test + APPEND + PROPERTY LABELS concurrency + ) + # ccf_kv/ccf_tasks are linked normally, unmodified, exactly like every + # other test target - pthread_mutex_wrap.cpp intercepts their real + # ccf::pal::Mutex use at link time instead (see its own comment), so no + # source ever needs recompiling against a different Mutex type. + target_link_libraries( + commit_concurrency_scheduled_test + PRIVATE ccfcrypto http_parser ccf_kv ccf_tasks + ) + # See pthread_mutex_wrap.cpp: __wrap_pthread_mutex_lock/unlock/trylock + # there are called instead of the real pthread_mutex_lock/unlock/ + # trylock for every call in this target (__real_pthread_mutex_* is how + # they still reach the genuine, original function). + target_link_options( + commit_concurrency_scheduled_test + PRIVATE + -Wl,--wrap=pthread_mutex_lock + -Wl,--wrap=pthread_mutex_unlock + -Wl,--wrap=pthread_mutex_trylock + ) + add_unit_test( raft_enclave_test ${CMAKE_CURRENT_SOURCE_DIR}/src/consensus/aft/test/enclave.cpp diff --git a/include/ccf/ds/locking.h b/include/ccf/ds/locking.h index 513a5a63054..6f3ab255172 100644 --- a/include/ccf/ds/locking.h +++ b/include/ccf/ds/locking.h @@ -7,6 +7,7 @@ #include #include #include +#include #include namespace ccf::ds @@ -14,6 +15,24 @@ namespace ccf::ds class ConditionVariable; class MutexGuard; + namespace detail + { + // Set immediately before Mutex's own lock()/try_lock()/unlock() make + // their real call, and consumed immediately by whatever runs next on + // this thread - not read by anything in this header itself. This + // lets a genuinely real, immediately-following OS-level lock/unlock + // call (which a bare mutex address alone cannot carry a label + // through) recover one anyway - see + // src/commit_concurrency/scheduled/pthread_mutex_wrap.cpp, which + // intercepts real pthread_mutex_lock/unlock/trylock calls to + // deterministically explore interleavings, and uses `pending` to + // tell a genuine ccf::ds::Mutex call apart from every other, + // unrelated lock in the process (allocator, iostream, etc.) without + // needing to track any mutex's address at all. + inline thread_local bool pending = false; + inline thread_local const char* pending_label = nullptr; + } + /** * Generic locking primitives shared across CCF components. */ @@ -31,18 +50,24 @@ namespace ccf::ds Mutex(const Mutex&) = delete; Mutex& operator=(const Mutex&) = delete; - void lock() CCF_ACQUIRE() + void lock(const char* label = nullptr) CCF_ACQUIRE() { + detail::pending = true; + detail::pending_label = label; mutex.lock(); } - bool try_lock() CCF_TRY_ACQUIRE(true) + bool try_lock(const char* label = nullptr) CCF_TRY_ACQUIRE(true) { + detail::pending = true; + detail::pending_label = label; return mutex.try_lock(); } - void unlock() CCF_RELEASE() + void unlock(const char* label = nullptr) CCF_RELEASE() { + detail::pending = true; + detail::pending_label = label; mutex.unlock(); } @@ -161,4 +186,100 @@ namespace ccf::ds lock.get(), timeout_time, std::move(predicate)); } }; + + // A drop-in replacement for std::unique_lock (supporting the same + // deferred-locking constructor and lock()/try_lock()/unlock() surface + // used against ccf::ds::Mutex elsewhere in this codebase), with an + // optional label describing why this lock is being taken - passed + // directly into Mutex's own lock()/try_lock()/unlock() call. With no + // label given, it defaults to the call site's source location. + // + // Carries its own CCF_SCOPED_CAPABILITY annotations (mirroring + // MutexGuard above), rather than relying on Clang's built-in, + // name-based special-casing of std::unique_lock, since this needs to + // call LockType's own lock()/try_lock()/unlock() directly (to pass a + // label through) rather than delegating to a real std::unique_lock + // member. This gives real static verification for the ordinary, + // unconditional case - a function using this type's lock/unlock like + // an ordinary scoped guard is checked exactly as if it used + // std::unique_lock or MutexGuard. The one gap: Clang's built-in + // std::unique_lock support additionally understands the + // conditionally-taken pattern (construct with std::defer_lock, only + // sometimes call .lock()/.try_lock() depending on runtime state) well + // enough to statically verify it; that specific pattern is not + // supported for a user-annotated type like this one, and needs + // CCF_NO_THREAD_SAFETY_ANALYSIS on the specific enclosing function that + // does it (a handful of call sites in this codebase - see their own + // comments for why). + template + class CCF_SCOPED_CAPABILITY unique_lock + { + LockType* mtx; + bool owned = false; + const char* label; + std::source_location loc; + + const char* effective_label() const + { + return label != nullptr ? label : loc.function_name(); + } + + public: + explicit unique_lock( + LockType& mtx_, + const char* label_ = nullptr, + std::source_location loc_ = std::source_location::current()) + CCF_ACQUIRE(mtx_) : + mtx(&mtx_), + label(label_), + loc(loc_) + { + lock(); + } + + unique_lock( + LockType& mtx_, + std::defer_lock_t, + const char* label_ = nullptr, + std::source_location loc_ = std::source_location::current()) : + mtx(&mtx_), + label(label_), + loc(loc_) + {} + + ~unique_lock() CCF_RELEASE() + { + if (owned) + { + unlock(); + } + } + + void lock() CCF_ACQUIRE() + { + mtx->lock(effective_label()); + owned = true; + } + + bool try_lock() CCF_TRY_ACQUIRE(true) + { + const bool locked = mtx->try_lock(effective_label()); + owned = locked; + return locked; + } + + void unlock() CCF_RELEASE() + { + mtx->unlock(effective_label()); + owned = false; + } + + bool owns_lock() const + { + return owned; + } + + unique_lock(const unique_lock&) = delete; + unique_lock& operator=(const unique_lock&) = delete; + }; } diff --git a/src/commit_concurrency/scheduled/deterministic_scheduler.h b/src/commit_concurrency/scheduled/deterministic_scheduler.h new file mode 100644 index 00000000000..538a06e8ba7 --- /dev/null +++ b/src/commit_concurrency/scheduled/deterministic_scheduler.h @@ -0,0 +1,829 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#pragma once + +// A cooperative scheduler for deterministically exploring real thread +// interleavings of real ccf::ds::Mutex use, without recompiling any +// production code against a different Mutex type: every real +// pthread_mutex_lock/unlock/trylock call is intercepted at link time (see +// src/commit_concurrency/scheduled/pthread_mutex_wrap.cpp) and, for a +// thread with an active DeterministicScheduler, is redirected to +// before_lock()/after_unlock() below instead of ever reaching the real +// mutex. Each participating actor runs on its own real OS thread, but the +// scheduler only ever lets one actor execute application code at a time; +// each such intercepted call is a point 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 real ccf::ds::Mutex used on a thread with no active +// DeterministicScheduler behaves exactly like an ordinary mutex. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace ccf::kv::test +{ + using ActorId = size_t; + + // What an actor was last known to be doing, for describe() to report + // against whichever decision comes next - whichever of these was + // reported last for that actor, however many decisions ago, stays in + // place until the next one (all three kinds behave identically here; + // none is cleared automatically). Acquired/Released are recorded + // automatically by before_lock()/after_unlock(), in sync with the exact + // lock event that caused them - see pthread_mutex_wrap.cpp's own + // __wrap_pthread_mutex_lock()/__wrap_pthread_mutex_unlock(). + // YieldPoint is for a scenario's own yield_point() label, describing + // something with no specific lock attached. + enum class ActorEventKind + { + YieldPoint, + Requested, + Acquired, + Released + }; + + struct ActorEvent + { + ActorEventKind kind = ActorEventKind::YieldPoint; + std::string label; + }; + + class DeterministicScheduler + { + public: + // One entry per point where the scheduler chose which ready actor + // would run next. `trigger` is the actor whose own progress led here + // (with `trigger_event`, the event it had just reported when it did) + // - std::nullopt only for the very first decision (see kick_off()), + // which has no preceding actor to attribute it to. Since only one + // actor's code ever runs at a time, `trigger` is always exactly + // whichever actor was `ready[chosen_index]` at the previous Decision. + // `ready` and `chosen_index` are the full set of candidates the + // scheduler picked from and which one it picked - not rendered by + // describe() below, but load-bearing for explore_all_interleavings()'s + // backtracking (see its own comments). + struct Decision + { + std::optional trigger; + ActorEvent trigger_event; + std::vector ready; + size_t chosen_index; + }; + + private: + struct MutexState + { + std::optional owner; + std::vector waiters; + }; + + // Per-actor state, indexed by ActorId - one entry per real actor, + // plus one extra for the reserved driver id (see DriverRegistration): + // the driver never gets marked finished or blocked_on_lock (its own + // before_lock()/after_unlock() return early, before touching either), + // but can still report a `current_event` via yield_point() while + // registered. + struct ActorState + { + bool finished = false; + bool blocked_on_lock = false; + ActorEvent current_event; + }; + + std::mutex m; + std::condition_variable cv; + size_t num_actors; + size_t parked_count = 0; + std::vector actors; + std::optional running; + std::function chooser; + std::vector path; + std::vector actor_names; + + // 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 ActorEvent already reported directly to before_lock()/ + // after_unlock()/yield_point() - 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 (!actors[a].finished && !actors[a].blocked_on_lock) + { + 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"); + } + // `running` still holds whichever actor was chosen at the previous + // Decision (or nullopt, only for this very first one) - since only + // one actor's code ever runs at a time, that is exactly the actor + // whose own progress brought execution to this choose_next() call, + // and actors[*running].current_event is exactly the event it just + // reported to get here (see before_lock()/after_unlock()/ + // yield_point()'s own comments, all of which set their actor's + // event immediately before calling this). + const std::optional trigger = running; + const ActorEvent trigger_event = + trigger.has_value() ? actors[*trigger].current_event : ActorEvent{}; + path.push_back(Decision{trigger, trigger_event, ready, 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_), + // One extra slot beyond the real actors, for the reserved driver id + // (see ActorState's own comment, and DriverRegistration). + actors(num_actors_ + 1), + chooser(std::move(chooser_)), + actor_names(std::move(actor_names_)) + {} + + // 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 YieldPoint event 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()) + { + actors[self].current_event = + ActorEvent{ActorEventKind::YieldPoint, std::move(label)}; + } + choose_next(lock); + cv.wait(lock, [&] { return running == self; }); + } + + // Called by __wrap_pthread_mutex_lock() (see pthread_mutex_wrap.cpp) + // for a real ccf::ds::Mutex lock attempt. Blocks until this actor + // actually holds the lock. The attempt itself is a decision point (its + // Requested event, below), before even checking whether the lock is + // free - without this, whichever actor happened to be running when + // it reached an uncontended lock would always win it unconditionally, + // since (only one actor's code ever runs at a time) no other actor + // could otherwise ever get a chance to reach for the same lock first. + // Acquiring it (whether or not this actor had to wait first) is a + // further decision point of its own, with an Acquired event. `label` + // (if given - see ccf::ds::unique_lock, the only real caller that + // supplies one) is used for both events. 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 any + // decision point, or event, of its own, since no other actor thread + // even exists yet to be a candidate. + void before_lock(ActorId self, void* mutex_key, const char* label = nullptr) + { + std::unique_lock lock(m); + auto& mtx = mutex_states[mutex_key]; + if (self >= num_actors) + { + mtx.owner = self; + return; + } + actors[self].current_event = + ActorEvent{ActorEventKind::Requested, label != nullptr ? label : ""}; + choose_next(lock); + cv.wait(lock, [&] { return running == self; }); + + while (mtx.owner.has_value()) + { + mtx.waiters.push_back(self); + actors[self].blocked_on_lock = 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; + actors[self].current_event = + ActorEvent{ActorEventKind::Acquired, label != nullptr ? label : ""}; + choose_next(lock); + cv.wait(lock, [&] { return running == self; }); + } + + // Called by __wrap_pthread_mutex_unlock() (see pthread_mutex_wrap.cpp) + // after a real ccf::ds::Mutex release. 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. `label` + // becomes this actor's Released event, recorded right before that + // same decision, so it is visible from this decision onward. 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, const char* label = nullptr) + { + 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()); + actors[woken].blocked_on_lock = false; + } + actors[self].current_event = + ActorEvent{ActorEventKind::Released, label != nullptr ? label : ""}; + 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); + actors[self].finished = true; + // Only the real actors (not the reserved driver slot, which is + // never marked finished) need to have finished. + if (std::all_of( + actors.begin(), + actors.begin() + static_cast(num_actors), + [](const ActorState& a) { return a.finished; })) + { + running.reset(); + cv.notify_all(); + return; + } + choose_next(lock); + } + + const std::vector& decision_path() const + { + return path; + } + + // A human-readable event stream, one line per decision: what the + // triggering actor (see Decision's own comment) just did, and - only + // when a genuine handoff happens, i.e. a different actor is chosen + // to continue - which actor resumes next. Deliberately does not list + // every other actor that was merely ready at that point (blocked or + // idly-ready-but-not-chosen are not a meaningful distinction here); + // decision_path() above still has that, for anything that needs it + // (e.g. explore_all_interleavings()'s own backtracking). 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) + ": "; + if (decision.trigger.has_value()) + { + out += actor_label(*decision.trigger); + const auto& event = decision.trigger_event; + if (!event.label.empty()) + { + switch (event.kind) + { + case ActorEventKind::Requested: + out += " requests " + event.label; + break; + case ActorEventKind::Acquired: + out += " acquires " + event.label; + break; + case ActorEventKind::Released: + out += " releases " + event.label; + break; + case ActorEventKind::YieldPoint: + default: + out += ": " + event.label; + break; + } + } + const auto chosen = decision.ready[decision.chosen_index]; + if (chosen != *decision.trigger) + { + out += ", " + actor_label(chosen) + " resumes"; + } + } + else + { + // The very first decision (see kick_off()) - nobody's own + // progress caused this one, it is simply who runs first. + out += actor_label(decision.ready[decision.chosen_index]) + " starts"; + } + out += "\n"; + } + return out; + } + + private: + // Keyed by the real pthread_mutex_t*'s own address (see + // pthread_mutex_wrap.cpp) - the scheduler never touches the real + // mutex at all, so this is purely bookkeeping for who is waiting on + // (the identity of) each one. + std::unordered_map mutex_states; + }; + + // Finds, and points a thread at, whichever DeterministicScheduler (if + // any) is exploring interleavings on the calling thread. + // 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: + 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 this actor's + // current YieldPoint event 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)); + } + } + + // 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 real ccf::ds::Mutex 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 real ccf::ds::Mutex 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 real ccf::ds::Mutex 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/scheduled/deterministic_scheduler_test.cpp b/src/commit_concurrency/scheduled/deterministic_scheduler_test.cpp new file mode 100644 index 00000000000..34072e6eed4 --- /dev/null +++ b/src/commit_concurrency/scheduled/deterministic_scheduler_test.cpp @@ -0,0 +1,310 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#include "ccf/ds/locking.h" +#include "commit_concurrency/scheduled/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); +} + +// A real ccf::ds::Mutex lock()/unlock() is intercepted at link time (see +// src/commit_concurrency/scheduled/pthread_mutex_wrap.cpp) rather than +// via a distinct C++ type, so nothing here stops a future change (e.g. +// dropping the -Wl,--wrap=... flags in CMakeLists.txt, or a libc/compiler +// change that stops emitting a plain pthread_mutex_lock call) from +// silently making that interception a no-op: every actor's real lock and +// unlock would then just succeed immediately as ordinary OS-level +// locking, with DeterministicScheduler never told about any of it. Every +// other test in this file could plausibly still pass in that scenario +// (e.g. "explored > 1" could, in principle, come from yield_point() calls +// alone) - this test instead asserts, explicitly and unambiguously, that +// a real lock/unlock actually produced the three events before_lock()/ +// after_unlock() are documented to report, which is only possible if +// interception genuinely engaged. +DOCTEST_TEST_CASE( + "Smoke test: a real ccf::ds::Mutex lock/unlock is genuinely intercepted, " + "not silently left as real, untracked OS-level locking" * + doctest::test_suite("deterministic_scheduler")) +{ + ccf::ds::Mutex mtx; + explore_all_interleavings( + 1, + [&]() -> std::vector> { + return {[&]() { std::lock_guard guard(mtx); }}; + }, + [&](const DeterministicScheduler& scheduler) { + const auto& path = scheduler.decision_path(); + const auto has_kind = [&](ActorEventKind kind) { + return std::any_of(path.begin(), path.end(), [&](const auto& d) { + return d.trigger_event.kind == kind; + }); + }; + DOCTEST_CHECK(has_kind(ActorEventKind::Requested)); + DOCTEST_CHECK(has_kind(ActorEventKind::Acquired)); + DOCTEST_CHECK(has_kind(ActorEventKind::Released)); + }); +} + +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; + ccf::ds::Mutex 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; + ccf::ds::Mutex 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 10968 schedules for this + // scenario now that every lock request/acquire/release (not just + // contended acquisitions) 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 10968)", + min_estimate, + max_estimate)); + DOCTEST_CHECK(min_estimate >= 1.0); + DOCTEST_CHECK(max_estimate <= 200000.0); + } +} diff --git a/src/commit_concurrency/scheduled/main.cpp b/src/commit_concurrency/scheduled/main.cpp new file mode 100644 index 00000000000..434e20d80c8 --- /dev/null +++ b/src/commit_concurrency/scheduled/main.cpp @@ -0,0 +1,23 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. + +// Doctest entry point for the scheduled concurrency-testing 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::DeterministicScheduler +// (src/commit_concurrency/scheduled/deterministic_scheduler.h), which +// systematically explores the space of legal interleavings of a bounded +// scenario - exhaustively where that space is small enough +// (explore_all_interleavings()), or by random sampling where it isn't +// (explore_random_interleavings(), used by every scenario below). + +#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/scheduled/pthread_mutex_wrap.cpp b/src/commit_concurrency/scheduled/pthread_mutex_wrap.cpp new file mode 100644 index 00000000000..7110dc12f74 --- /dev/null +++ b/src/commit_concurrency/scheduled/pthread_mutex_wrap.cpp @@ -0,0 +1,109 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. + +// Intercepts every real pthread_mutex_lock()/unlock()/trylock() call in +// this binary (via -Wl,--wrap=..., see CMakeLists.txt), so that a real +// ccf::ds::Mutex - used, unmodified, by production code linked into this +// binary (Store's version_lock, Aft's state->lock, ccf::tasks' job board, +// etc.) - can be driven by a DeterministicScheduler without recompiling +// any of that code against a different Mutex type at all. +// +// ccf::ds::Mutex's own lock()/try_lock()/unlock() (see +// include/ccf/ds/locking.h) set ccf::ds::detail::pending immediately +// before making the real call intercepted here. This is how the +// functions below tell a genuine ccf::ds::Mutex call apart from every +// other, unrelated pthread_mutex_lock/unlock/trylock call anywhere else +// in the binary (allocator internals, iostream, DeterministicScheduler's +// own bookkeeping mutex, etc.), with no need to track any mutex's address +// at all. The flag is consumed (reset to false) the instant it is read, +// so a nested/recursive real lock call - e.g. DeterministicScheduler's +// own internal std::mutex, locked from inside before_lock()/ +// after_unlock() themselves - correctly sees it already cleared, and +// falls straight through to a real lock, with no risk of infinite +// recursion. + +#include "ccf/ds/locking.h" +#include "commit_concurrency/scheduled/deterministic_scheduler.h" + +#include +#include +#include + +extern "C" int __real_pthread_mutex_lock(pthread_mutex_t* mutex); +extern "C" int __real_pthread_mutex_unlock(pthread_mutex_t* mutex); +extern "C" int __real_pthread_mutex_trylock(pthread_mutex_t* mutex); + +namespace +{ + bool consume_pending() + { + if (!ccf::ds::detail::pending) + { + return false; + } + ccf::ds::detail::pending = false; + return true; + } +} + +extern "C" int __wrap_pthread_mutex_lock(pthread_mutex_t* mutex) +{ + if (!consume_pending()) + { + return __real_pthread_mutex_lock(mutex); + } + auto* scheduler = ccf::kv::test::SchedulerThreadContext::scheduler(); + if (scheduler == nullptr) + { + return __real_pthread_mutex_lock(mutex); + } + scheduler->before_lock( + ccf::kv::test::SchedulerThreadContext::actor(), + mutex, + ccf::ds::detail::pending_label); + return 0; +} + +extern "C" int __wrap_pthread_mutex_unlock(pthread_mutex_t* mutex) +{ + if (!consume_pending()) + { + return __real_pthread_mutex_unlock(mutex); + } + auto* scheduler = ccf::kv::test::SchedulerThreadContext::scheduler(); + if (scheduler == nullptr) + { + return __real_pthread_mutex_unlock(mutex); + } + scheduler->after_unlock( + ccf::kv::test::SchedulerThreadContext::actor(), + mutex, + ccf::ds::detail::pending_label); + return 0; +} + +extern "C" int __wrap_pthread_mutex_trylock(pthread_mutex_t* mutex) +{ + if (!consume_pending()) + { + return __real_pthread_mutex_trylock(mutex); + } + auto* scheduler = ccf::kv::test::SchedulerThreadContext::scheduler(); + if (scheduler == nullptr) + { + return __real_pthread_mutex_trylock(mutex); + } + // 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. + // Aborts directly, rather than throwing, because this is reached from + // std::mutex::try_lock(), which is noexcept - an exception escaping it + // would call std::terminate() anyway, with less control over the + // diagnostic than doing so explicitly here. + std::fprintf( + stderr, + "FATAL: pthread_mutex_trylock intercepted under an active " + "DeterministicScheduler, but try_lock() is not yet implemented for " + "scheduled scenarios\n"); + std::abort(); +} diff --git a/src/commit_concurrency/scheduled/rejected_commit.cpp b/src/commit_concurrency/scheduled/rejected_commit.cpp new file mode 100644 index 00000000000..6d5afe1b0ee --- /dev/null +++ b/src/commit_concurrency/scheduled/rejected_commit.cpp @@ -0,0 +1,152 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#include "commit_concurrency/scheduled/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 +// (also a regression test for #8242). +// 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_scheduled")) +{ + 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)); + } + }; + + 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_scheduled")) +{ + 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)); + } + }; + + 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/checkpoint.h b/src/commit_concurrency/threaded/checkpoint.h new file mode 100644 index 00000000000..1053d0df952 --- /dev/null +++ b/src/commit_concurrency/threaded/checkpoint.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/threaded/checkpoint_test.cpp b/src/commit_concurrency/threaded/checkpoint_test.cpp new file mode 100644 index 00000000000..4be1de3cd83 --- /dev/null +++ b/src/commit_concurrency/threaded/checkpoint_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/threaded/checkpoint.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("checkpoint")) +{ + 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("checkpoint")) +{ + 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("checkpoint")) +{ + 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("checkpoint")) +{ + 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("checkpoint")) +{ + 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/threaded/deterministic.cpp b/src/commit_concurrency/threaded/deterministic.cpp new file mode 100644 index 00000000000..4087c6cac17 --- /dev/null +++ b/src/commit_concurrency/threaded/deterministic.cpp @@ -0,0 +1,420 @@ +// 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/threaded/checkpoint.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(), 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 - just like the " + "test below, which checks the same thing for a transaction whose " + "commit view goes stale before it ever reaches Store::commit()"); + 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); + } +} + +// Regression test for #8242 ("Reject stale-view writes before local +// commit"): kv_test.cpp's "Stale-view writes are rejected before local +// application" checks the same invariant directly, single-threaded, via +// an explicit Store::rollback() call. The test below drives the same +// rejection through a real election instead, and additionally +// cross-checks the result against TxHistory and raft's own replication +// index - neither of which kv_test.cpp's version touches. +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); + 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( + 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 00000000000..e7e76f44aab --- /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/threaded/checkpoint.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 00000000000..7a2bed96fee --- /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 00000000000..01dd621c24a --- /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 00000000000..3db1ea6988b --- /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/raft.h b/src/consensus/aft/raft.h index 2e745cf5e7a..2524b0412b2 100644 --- a/src/consensus/aft/raft.h +++ b/src/consensus/aft/raft.h @@ -410,7 +410,8 @@ namespace aft "Can't force leadership if there is already a leader"); } - std::lock_guard guard(state->lock); + ccf::ds::unique_lock guard( + state->lock, "force this node to become primary"); state->current_view += starting_view_change; become_leader(true); } @@ -429,7 +430,8 @@ namespace aft "Can't force leadership if there is already a leader"); } - std::lock_guard guard(state->lock); + ccf::ds::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_; diff --git a/src/kv/store.h b/src/kv/store.h index 0f32c766a4e..63ef19d8553 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -662,7 +662,8 @@ namespace ccf::kv std::lock_guard mguard(maps_lock); { - std::lock_guard vguard(version_lock); + ccf::ds::unique_lock vguard( + version_lock, "roll version and history back to tx_id"); if (tx_id.seqno < compacted) { throw std::logic_error(fmt::format( @@ -978,7 +979,8 @@ namespace ccf::kv return CommitResult::SUCCESS; } - std::lock_guard cguard(commit_lock); + ccf::ds::unique_lock cguard( + commit_lock, "serialise concurrent Store::commit() calls"); LOG_DEBUG_FMT( "Store::commit {}{}", @@ -996,7 +998,9 @@ namespace ccf::kv auto h = get_history(); { - std::lock_guard vguard(version_lock); + ccf::ds::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, @@ -1125,7 +1129,8 @@ namespace ccf::kv if (c->replicate(batch, replication_view)) { - std::lock_guard vguard(version_lock); + ccf::ds::unique_lock vguard( + version_lock, "advance last_replicated after successful replicate()"); if ( last_replicated == previous_last_replicated && previous_rollback_count == rollback_count) diff --git a/src/node/history.h b/src/node/history.h index 6ebfccf01d8..c531091b4bf 100644 --- a/src/node/history.h +++ b/src/node/history.h @@ -649,56 +649,66 @@ 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); - - bool should_emit_signature = false; - - if (mguard.try_lock()) - { - auto consensus = this->store.get_consensus(); - if (consensus != nullptr) + // CCF_NO_THREAD_SAFETY_ANALYSIS: mguard below is only + // conditionally locked (via std::defer_lock, then .try_lock()) - + // real, correct behaviour that Clang's thread-safety analysis + // cannot statically verify for a ccf::ds::unique_lock used this + // way (unlike its built-in support for std::unique_lock, which + // does handle this exact pattern). + emit_signature_periodic_task = + ccf::tasks::make_basic_task([this]() CCF_NO_THREAD_SAFETY_ANALYSIS { + ccf::ds::unique_lock mguard( + this->signature_lock, + std::defer_lock, + "periodic signature emission"); + + bool should_emit_signature = false; + + if (mguard.try_lock()) { - auto sig_disp = consensus->get_signature_disposition(); - switch (sig_disp) + auto consensus = this->store.get_consensus(); + if (consensus != nullptr) { - case ccf::kv::Consensus::SignatureDisposition::CANT_REPLICATE: - { - break; - } - case ccf::kv::Consensus::SignatureDisposition::CAN_SIGN: + auto sig_disp = consensus->get_signature_disposition(); + switch (sig_disp) { - // To snapshot we need to complete the chunk and to do that we - // need to set the force_chunk_after flag on the last snapshot - // in it. - // At this point the previous signature is already replicating - // and is immutable. - // So if we need to snapshot, we need to emit a new signature to - // ensure we can set the force_chunk_after flag, even if there - // are no other transactions between this and the last snapshot - if ( - this->store.committable_gap() > 0 || - this->store.should_schedule_snapshot()) + case ccf::kv::Consensus::SignatureDisposition::CANT_REPLICATE: + { + break; + } + case ccf::kv::Consensus::SignatureDisposition::CAN_SIGN: + { + // To snapshot we need to complete the chunk and to do that we + // need to set the force_chunk_after flag on the last snapshot + // in it. + // At this point the previous signature is already replicating + // and is immutable. + // So if we need to snapshot, we need to emit a new signature + // to ensure we can set the force_chunk_after flag, even if + // there are no other transactions between this and the last + // snapshot + if ( + this->store.committable_gap() > 0 || + this->store.should_schedule_snapshot()) + { + should_emit_signature = true; + } + break; + } + case ccf::kv::Consensus::SignatureDisposition::SHOULD_SIGN: { should_emit_signature = true; + break; } - break; - } - case ccf::kv::Consensus::SignatureDisposition::SHOULD_SIGN: - { - should_emit_signature = true; - break; } } } - } - if (should_emit_signature) - { - this->emit_signature(); - } - }); + if (should_emit_signature) + { + this->emit_signature(); + } + }); ccf::tasks::add_periodic_task(emit_signature_periodic_task, delay, delay); } @@ -919,9 +929,16 @@ namespace ccf ccf::ds::Mutex signature_lock; - void try_emit_signature() override + // CCF_NO_THREAD_SAFETY_ANALYSIS: mguard below is only conditionally + // locked (via std::defer_lock, then .try_lock()) - real, correct + // behaviour that Clang's thread-safety analysis cannot statically + // verify for a ccf::ds::unique_lock used this way (unlike its + // built-in support for std::unique_lock, which does handle this + // exact pattern). + void try_emit_signature() override CCF_NO_THREAD_SAFETY_ANALYSIS { - std::unique_lock mguard(signature_lock, std::defer_lock); + ccf::ds::unique_lock mguard( + signature_lock, std::defer_lock, "on-demand signature emission"); if (store.committable_gap() < sig_tx_interval || !mguard.try_lock()) { return;