tokio_rcu 0.2.1

//! a rust library providing an RCU (read-copy-update) algorithm specifically made for async rust with tokio.
//!
//! this provides a lock-free and wait-free way to update a shared piece of state while it is concurrently being read and updated by
//! other tasks.
//!
//! the core primitive provided by this crate is [`synchronize_rcu`], which works just like the `synchronize_rcu` function in the
//! linux kernel - it waits for an rcu grace period, which allows writers to track when exactly they can reclaim swapped out data.
//!
//! the low level [`synchronize_rcu`] primitive can be used to build a bunch of higher level abstractions.
//! one very simple abstraction - a single pointer to a heap-allocated piece of shared data (an "rcu box") - is implemented in this crate
//! by the [`RcuBox`] type.
//!
//! # performance
//!
//! NOTE: this section specifically refers to [`RcuBox`], the main high level abstraction provided by this crate, but will probably also
//! apply to most other abstractions which can be implemented using the rcu primitive.
//!
//! this crate is speicifcally useful for read-mostly data, as it makes readers extremely fast at the cost of making the writers slower.
//! when a reader reads the data stored in an rcu box (e.g. using [`RcuBox::read`]), the read operation is only a single load of an atomic
//! pointer. that's it. no branches, no book-keeping, just a single pointer load. it is basically the fastest a read can get.
//!
//! during a read operation, no memory writes are performed, as opposed to spinlocks and mutexes which requires memory writes to shared
//! data, and sometimes even syscalls, just to access the underlying data.
//!
//! specifically, for read-mostly data, the cache line containing the pointer can be shared between all readers, and the read operation
//! becomes just a single load from the cpu cache, which is extremely fast.
//! compared to spinlocks and mutexes which usually require exclusive ownership over the cacheline due to writes and other atomic
//! operations, this is much faster and provides much better reader latency.
//!
//! also note that the time spent on a read operation is very predictable and static. other users of the data, such as concurrent readers
//! and even writers, do not affect the time it takes for a reader to read the data. a read is always just a single pointer load.
//! specifically a writer can slightly delay this load due to invalidating the cacheline containing the rcu protected pointer when
//! writing to it, but this is mostly negligible.
//! this consistency of the read operation can be very important in latency-critical applications which require a high-performance
//! fast path with predictable latency.
//!
//! also see [benchmnarks](#benchmarks).
//!
//! # quick start
//!
//! ```rust
//! use tokio_rcu::{rcu_block_on, rcu_box::RcuBox};
//!
//! fn main() {
//!     rcu_block_on(async move {
//!         let numbers = RcuBox::new(Box::new(vec![1, 2, 3, 4]));
//!
//!         // the rcu box's data can safely be accessed using the `with` function.
//!         numbers.with(|numbers| {
//!             assert!(numbers.contains(&3));
//!             assert!(!numbers.contains(&5));
//!         });
//!
//!         // the rcu box's data can be modified while readers are using it.
//!         // and, the old allocation is returned.
//!         let new_numbers = Box::new(vec![5, 6, 7, 8]);
//!         let _old_numbers: Box<Vec<i32>> = numbers.swap(new_numbers).await;
//!
//!         numbers.with(|numbers| {
//!             assert!(numbers.contains(&6));
//!             assert!(!numbers.contains(&10));
//!         });
//!     })
//! }
//! ```
//!
//! # realistic use case
//!
//! for a more realistic use case, see `examples/basic.rs`.
//!
//! # how does it work?
//!
//! when a writer swaps out an old data pointer with a new data pointer containing updated data, he must then know when the previous
//! data pointer can be freed.
//!
//! to free the data, the writer must first wait for all potential readers, who have already read the pointer and are now using it, to
//! finish using that pointer, to avoid a UAF (use-after-free) situation.
//!
//! this problem can be solved in many ways, but rcu usually solves it by defining a state called a "quiescent state", such that when
//! a specific execution context (which can be a cpu core, or an OS thread) reaches that quiescent state, it is guaranteed to not hold
//! any rcu-protected pointer.
//!
//! in this specific crate, the execution contexts are tokio threads, and the quiescent state was chosen to be tokio's
//! [`on_after_task_poll`] hook.
//!
//! this works because this crate limits the usage of rcu protected pointers in a way that prevents them from being held across await
//! points.
//! so, when the runtime reaches the [`on_after_task_poll`] hook, it is guaranteed that no future is currently being executed on
//! the current thread, and since rcu protected pointers can't be held across await points, it is basically guaranteed that the current
//! thread is not holding any rcu protected pointers.
//!
//! waiting an rcu grace period thus means first ensuring that all of our previous memory writes (e.g. rcu pointer swaps) are visible to
//! all other threads, and then just waiting for each other thread to pass at least once through a quiescent state.
//! after such a grace period, it is guaranteed that any swapped out pointers are no longer used by any of the threads, so their memory
//! can be reclaimed.
//!
//! # enabling rcu support
//!
//! to use the rcu primitives, you must use an rcu enabled tokio runtime.
//!
//! the easiest way to do this is to use the [`rcu_block_on`] function which creates a tokio runtime with rcu support enabled, and then
//! runs the provided future inside that runtime using tokio's [`block_on`](tokio::runtime::Runtime::block_on).
//!
//! if you wish to manually configure your runtime, you can use the more low-level [`enable_rcu`](TokioRuntimeBuilderExt::enable_rcu) and
//! [`rcu_block_on`](TokioRuntimeExt::rcu_block_on) functions.
//!
//! # performance overhead
//!
//! enabling rcu for a tokio runtime does introduce a little bit of overhead.
//!
//! specifically, this crate uses tokio hooks (e.g. [`on_after_task_poll`]) to track quiescent states of tokio's worker threads.
//!
//! but, this crate performs a lot of efforts to make this overhead as small as possible, especially in hooks like [`on_after_task_poll`]
//! which are called very often.
//!
//! for example, the current implementation of the [`on_after_task_poll`] hook is basically just a couple of atomic loads and stores,
//! and is unnoticeable in terms of performance.
//!
//! # other async runtimes
//!
//! this crate could quite easily be ported to work with other runtimes other than tokio. i chose tokio because it is the most popular
//! runtime, and because it already provides hooks which allow me to track quiescent states quite easily.
//!
//! # stability
//!
//! this crate currently requires using the `tokio_unstable` configuration of tokio. this is required since the [`on_after_task_poll`]
//! hook is currently unstable, and is needed to make this crate work.
//!
//! # benchmarks
//!
//! to run the benchmarks, use:
//! ```bash
//! cargo bench
//! ```
//!
//! the benchmarks mostly compare this crate with the `arc_swap` crate, which solves the same problem in a different way and provides
//! a very similar interface.
//!
//! here are the results of running the benchmarks on my 20-core `12th Gen Intel(R) Core(TM) i7-12700` cpu:
//! ```text
//! Timer precision: 47 ns
//! comparison                              fastest       │ slowest       │ median        │ mean          │ samples │ iters
//! ├─ read_only_arc_swap                                 │               │               │               │         │
//! │  ├─ 1                                 22.33 ms      │ 34.62 ms      │ 22.82 ms      │ 23.87 ms      │ 100     │ 100
//! │  ├─ 8                                 24.86 ms      │ 46.24 ms      │ 28.97 ms      │ 29.83 ms      │ 100     │ 100
//! │  ├─ 16                                38.22 ms      │ 44.77 ms      │ 43.65 ms      │ 42.8 ms       │ 100     │ 100
//! │  ├─ 32                                65.91 ms      │ 75.48 ms      │ 68.38 ms      │ 69.26 ms      │ 100     │ 100
//! │  ╰─ 64                                129.6 ms      │ 137.4 ms      │ 133.1 ms      │ 132.9 ms      │ 100     │ 100
//! ├─ read_only_rcu_box                                  │               │               │               │         │
//! │  ├─ 1                                 1.005 ms      │ 6.71 ms       │ 1.806 ms      │ 2.63 ms       │ 100     │ 100
//! │  ├─ 8                                 1.193 ms      │ 7.442 ms      │ 2.384 ms      │ 2.733 ms      │ 100     │ 100
//! │  ├─ 16                                1.29 ms       │ 4.381 ms      │ 2.521 ms      │ 2.626 ms      │ 100     │ 100
//! │  ├─ 32                                1.811 ms      │ 3.923 ms      │ 1.992 ms      │ 2.166 ms      │ 100     │ 100
//! │  ╰─ 64                                3.173 ms      │ 4.016 ms      │ 3.427 ms      │ 3.496 ms      │ 100     │ 100
//! ├─ read_while_writing_arc_swap                        │               │               │               │         │
//! │  ├─ 1 reader tasks, 1 writer tasks    24.81 ms      │ 29.73 ms      │ 25.46 ms      │ 26.09 ms      │ 100     │ 100
//! │  ├─ 8 reader tasks, 1 writer tasks    27.52 ms      │ 46.04 ms      │ 33.75 ms      │ 34.25 ms      │ 100     │ 100
//! │  ├─ 8 reader tasks, 2 writer tasks    35.69 ms      │ 58.73 ms      │ 39.27 ms      │ 40.4 ms       │ 100     │ 100
//! │  ├─ 16 reader tasks, 1 writer tasks   50.47 ms      │ 60.23 ms      │ 55.65 ms      │ 55.25 ms      │ 100     │ 100
//! │  ├─ 16 reader tasks, 2 writer tasks   58.23 ms      │ 79.99 ms      │ 67.49 ms      │ 67.51 ms      │ 100     │ 100
//! │  ├─ 32 reader tasks, 1 writer tasks   67.38 ms      │ 81.49 ms      │ 71.44 ms      │ 72.57 ms      │ 100     │ 100
//! │  ├─ 32 reader tasks, 2 writer tasks   69.12 ms      │ 88.03 ms      │ 72.5 ms       │ 74.51 ms      │ 100     │ 100
//! │  ├─ 64 reader tasks, 1 writer tasks   130.8 ms      │ 140 ms        │ 134.5 ms      │ 134.5 ms      │ 100     │ 100
//! │  ╰─ 64 reader tasks, 2 writer tasks   131.8 ms      │ 143.1 ms      │ 136.8 ms      │ 136.5 ms      │ 100     │ 100
//! ├─ read_while_writing_rcu_box                         │               │               │               │         │
//! │  ├─ 1 reader tasks, 1 writer tasks    1.09 ms       │ 4.477 ms      │ 1.415 ms      │ 1.822 ms      │ 100     │ 100
//! │  ├─ 8 reader tasks, 1 writer tasks    1.445 ms      │ 5.593 ms      │ 2.373 ms      │ 2.724 ms      │ 100     │ 100
//! │  ├─ 8 reader tasks, 2 writer tasks    1.692 ms      │ 4.929 ms      │ 2.576 ms      │ 2.769 ms      │ 100     │ 100
//! │  ├─ 16 reader tasks, 1 writer tasks   1.375 ms      │ 3.961 ms      │ 1.786 ms      │ 1.946 ms      │ 100     │ 100
//! │  ├─ 16 reader tasks, 2 writer tasks   1.74 ms       │ 3.345 ms      │ 2.025 ms      │ 2.056 ms      │ 100     │ 100
//! │  ├─ 32 reader tasks, 1 writer tasks   1.94 ms       │ 3.088 ms      │ 2.479 ms      │ 2.481 ms      │ 100     │ 100
//! │  ├─ 32 reader tasks, 2 writer tasks   2.034 ms      │ 3.504 ms      │ 2.764 ms      │ 2.73 ms       │ 100     │ 100
//! │  ├─ 64 reader tasks, 1 writer tasks   3.395 ms      │ 5.704 ms      │ 4.037 ms      │ 4.074 ms      │ 100     │ 100
//! │  ╰─ 64 reader tasks, 2 writer tasks   3.332 ms      │ 4.919 ms      │ 4.352 ms      │ 4.263 ms      │ 100     │ 100
//! ├─ write_while_reading_arc_swap                       │               │               │               │         │
//! │  ├─ 1 reader tasks, 1 writer tasks    447 µs        │ 889.9 µs      │ 534.5 µs      │ 572.5 µs      │ 100     │ 100
//! │  ├─ 1 reader tasks, 8 writer tasks    622.1 µs      │ 1.843 ms      │ 870 µs        │ 959.1 µs      │ 100     │ 100
//! │  ├─ 1 reader tasks, 16 writer tasks   841.5 µs      │ 4.713 ms      │ 1.96 ms       │ 2.201 ms      │ 100     │ 100
//! │  ├─ 1 reader tasks, 32 writer tasks   1.304 ms      │ 8.912 ms      │ 2.989 ms      │ 3.186 ms      │ 100     │ 100
//! │  ├─ 1 reader tasks, 64 writer tasks   2.711 ms      │ 10.06 ms      │ 4.755 ms      │ 5.233 ms      │ 100     │ 100
//! │  ├─ 2 reader tasks, 2 writer tasks    510.9 µs      │ 921.8 µs      │ 656.5 µs      │ 667.8 µs      │ 100     │ 100
//! │  ├─ 4 reader tasks, 8 writer tasks    777.4 µs      │ 2.603 ms      │ 1.518 ms      │ 1.541 ms      │ 100     │ 100
//! │  ├─ 8 reader tasks, 8 writer tasks    1.274 ms      │ 4.425 ms      │ 2.293 ms      │ 2.441 ms      │ 100     │ 100
//! │  ├─ 16 reader tasks, 16 writer tasks  1.845 ms      │ 10.21 ms      │ 2.272 ms      │ 3.091 ms      │ 100     │ 100
//! │  ├─ 32 reader tasks, 32 writer tasks  4.576 ms      │ 12.01 ms      │ 6.949 ms      │ 6.731 ms      │ 100     │ 100
//! │  ╰─ 64 reader tasks, 64 writer tasks  10.51 ms      │ 22.3 ms       │ 12.26 ms      │ 12.6 ms       │ 100     │ 100
//! ╰─ write_while_reading_rcu_box                        │               │               │               │         │
//!    ├─ 1 reader tasks, 1 writer tasks    494.8 µs      │ 1.73 ms       │ 570.5 µs      │ 628.4 µs      │ 100     │ 100
//!    ├─ 1 reader tasks, 8 writer tasks    801.8 µs      │ 7.725 ms      │ 2.019 ms      │ 2.388 ms      │ 100     │ 100
//!    ├─ 1 reader tasks, 16 writer tasks   872.4 µs      │ 73.89 ms      │ 1.671 ms      │ 3.982 ms      │ 100     │ 100
//!    ├─ 1 reader tasks, 32 writer tasks   931.4 µs      │ 194 ms        │ 4.074 ms      │ 21.04 ms      │ 100     │ 100
//!    ├─ 1 reader tasks, 64 writer tasks   1.056 ms      │ 199 ms        │ 4.969 ms      │ 42.72 ms      │ 100     │ 100
//!    ├─ 2 reader tasks, 2 writer tasks    648.5 µs      │ 1.945 ms      │ 818.2 µs      │ 864.3 µs      │ 100     │ 100
//!    ├─ 4 reader tasks, 8 writer tasks    1.44 ms       │ 9.372 ms      │ 2.512 ms      │ 2.869 ms      │ 100     │ 100
//!    ├─ 8 reader tasks, 8 writer tasks    2.514 ms      │ 6.697 ms      │ 2.94 ms       │ 3.121 ms      │ 100     │ 100
//!    ├─ 16 reader tasks, 16 writer tasks  6.412 ms      │ 17.66 ms      │ 6.888 ms      │ 7.19 ms       │ 100     │ 100
//!    ├─ 32 reader tasks, 32 writer tasks  10.35 ms      │ 41.39 ms      │ 13.26 ms      │ 13.1 ms       │ 100     │ 100
//!    ╰─ 64 reader tasks, 64 writer tasks  15.94 ms      │ 29.01 ms      │ 19.01 ms      │ 19.01 ms      │ 100     │ 100
//! ```
//!
//! as you can see, `tokio_rcu`'s reads are faster than `arc_swap`'s reads (about 9x-40x faster on average), while `tokio_rcu`'s writes
//! are slower than `arc_swap`'s writes (about 2x slower on average). for a read-heavy situation, this is ideal.
//!
//! furthermore, note that when using `arc_swap`, the time it takes for a single read operation seems to scale with the number of
//! concurrent readers (see the results of the `arc_swap_read_only` and `arc_swap_read_while_writing` benchmarks), while `tokio_rcu`'s
//! read operation takes roughly the same amount of time regardless of the number of concurrent readers, up until the point where there
//! are more readers than cpu cores (more than 20 reader tasks), at which point the readers start sharing cpu cores and competing for
//! their runtime, which obviously takes its toll on the performance.
//!
//! also note that this constant time for the read operation holds even when writers are concurrently modifying the data - the time
//! spent on a single read operation remains roughly the same (see `rcu_box_read_while_writing`), unlike `arc_swap` (see
//! `arc_swap_read_while_writing`).
//!
//! moreover, while `tokio_rcu`'s writes are slower, it is mostly because the writers are sleeping while waiting for other threads to
//! pass through a quiescent state, so they are NOT slower in the sense that they perform more cpu-bound work, only in the total time it
//! takes for a swap operation to complete after fully awaiting it. in practice the writes may actually spend less cpu time than
//! `arc_swap`'s write.
//!
//! # testing
//!
//! to run the tests, use:
//! ```bash
//! cargo all-features nextest run --release
//! cargo all-features test --doc --release
//! ```
//!
//! (NOTE: this requires installing `cargo-all-features` and `cargo-nextest`)
//!
//! `cargo-nextest` is used since it allows running each test as a separate process, which is important for testing this crate, since
//! this crate heavily relies on thread local variables and generally assumes that only a single tokio runtime is used per process.
//! furthermore, bugs in the rcu primitives can cause UAFs which may crash the process. if the process crashes when using `cargo test`,
//! all tests stop running and no diagnostics are reported. with `cargo-nextest`, such failures are gracefully reported as test failures.
//!
//! sadly, `cargo-nextest` currently does not support running doctests, so we must run them separately. note that `cargo test --doc`
//! already runs each test in its own process, so luckily for us, we don't need `cargo-nextest` for process isolation in this case.
//!
//! furthermore, `cargo-all-features` is used to also test the crate under the `small_epoch_id` feature flag, which is for testing mode
//! only, and allows testing some internal edge cases of this crate which are extremely hard to reach in the default configuration.
//!
//! it is also recommended to run the tests in release mode since it increases the probability of being able to find race conditions and
//! other hard to catch edge cases.
//!
//! # platform support
//!
//! currently, this crate only works on linux and windows.
//!
//! the limitation stems from the membarrier operation, which is currently only implemented for linux (using the membarrier syscall),
//! and windows (using FlushProcessWriteBuffers).
//!
//! more platforms can be added in the future if needed, and given that they have a way to emulate the behaviour of membarrier.
//!
//! # license
//!
//! This project is licensed under the MIT license.
//!
//! [`on_after_task_poll`]: tokio::runtime::Builder::on_after_task_poll
//! [`RcuBox`]: rcu_box::RcuBox
//! [`RcuBox::read`]: rcu_box::RcuBox::read
use std::{sync::atomic, task::Poll};

use crate::{
    epoch::{EPOCH_ID_MIN, EpochId, epoch_id_get, epoch_id_inc, epoch_id_set},
    notify::Notify,
    per_thread_storage::{
        this_thread_alloc_storage_slot, this_thread_dealloc_storage_slot,
        this_thread_does_have_allocated_storage_slot, this_thread_get_storage_slot_id,
        thread_storage_slot_get_all,
    },
    thread_state::ThreadState,
};

mod atomic_type;
mod epoch;
mod membarrier;
mod notify;
mod per_thread_storage;
pub mod rcu_box;
mod thread_state;
mod utils;

use branches::{likely, unlikely};
use tokio::runtime::RuntimeFlavor;

/// a notification which is notified when threads update their last seen epoch id or change their status in any other meaningful
/// way (e.g. become non-busy). used by waiters to wait for notifications in a blocking manner while waiting for threads to see
/// their new epoch id, instead of constantly busy polling all threads.
static THREAD_EPOCH_UPDATED_NOTIFY: Notify = Notify::new();

/// a lock used to synchronize the reset operation.
/// a reset operation is performed when the epoch id overflows, in order to reset the epoch id back to its minimum value.
///
/// when some thread increments the epoch id and causes it to exceed its max threshold, this thread begins a reset operation.
/// for resetting the epoch id, the thread must reset the global epoch id back to its initial value, then wait for all threads to
/// see this new state while blocking any further increments of the epoch id until all threads see the reset value.
///
/// in order to prevent the further increments of the epoch id during the reset operation, this lock is used.
/// all incrementors of the epoch id lock it for reading before incrementing, and during the reset operation, the leader of the reset (the
/// first one to increment the epoch id past its max threshold) locks this lock for writing, thus preventing any new incrementors from
/// incrementing the epoch id.
///
/// this also ensures that we don't start performing a reset operation while some incrementor thread is still waiting for threads to see
/// his incremented epoch id. if we were to start the reset while we was waiting, we would get stuck until the next overflow of the epoch
/// id.
static EPOCH_ID_RESET_SYNC_LOCK: tokio::sync::RwLock<()> = tokio::sync::RwLock::const_new(());

/// when a thread increments the epoch id past its max threshold, this thread begins a reset operation.
/// while that thread was incrementing the epoch id, another thread may have also been incrementing the epoch id, and also saw that it
/// reached its max threshold. so, that thread also begins the reset operation.
///
/// in practice, the reset is actually only performed by a single thread - the leader, and all other threads that entered reset just wait
/// for him to finish resetting.
///
/// so, this notification used by the leader of a reset operation to notify all other threads that have also entered reset that the reset
/// operation is done.
static RESET_FINISHED_NOTIFICATION: Notify = Notify::new();

/// wait for an RCU grace period.
///
/// this function first performs a membarrier to synchronize all previous writes performed by the current thread with all other
/// threads in the process.
///
/// after performing the membarrier, this function waits for every thread that was active during the membarrier operation to pass
/// through a quiescent state or to became unactive.
///
/// a quiescent state of a thread is defined as a state where the thread is not executing any user-defined task, and is instead executing
/// code inside tokio's task scheduling logic.
///
/// if `include_calling_thread` is set, this function also waits for the calling thread itself to pass through quiescent state after the
/// membarrier operation. this is usually not needed and should be set to `false`.
/// this flag exists as a workaround to remove overhead from the fast-path of the rcu to the slow path.
/// specifically, this helps preventing a specific category of misuse where a user tries to swap an rcu pointer while simultaneously
/// holding a read guard to it on the same thread, for example by manually polling the swap future.
/// making this also wait for the calling thread prevents this misuse from causing a UAF, instead converting it to a deadlock - the
/// synchronize rcu operation will never finish unless the caller actually passes through a quiescent state, at which point he can no
/// longer be holding any read guards.
/// a deadlock is not ideal, but this should never happen during proper use of this library anyway, and it prevents the UAF without
/// adding overhead of checks in the fast path, which is a big win.
/// also note that setting this flag means that the synchronize rcu operation will always yield at least once, to let the calling thread
/// pass through a quiescent state, even if all threads immediately pass through a quiescent state after the membarrier.
pub async fn synchronize_rcu(include_calling_thread: bool) {
    // perform a membarrier to make sure that all other threads see the new rcu pointer.
    membarrier::perform();

    // after the membarrier, all threads are guaranteed to have seen our new pointer.
    // we only need to wait for any potential existing users of the old pointer to finish using it.
    //
    // note that due to the membarrier, we don't need to worry about just-starting threads or just-unparking threads which
    // may access the old pointer.
    //
    // if during the check below, we see that some thread is currently parked, or we don't see the slot of some just-started
    // thread, then it means that this thread's update of its own state happens strictly after the membarrier, and the state
    // update always happens before polling the future, so there's no way for the polled future to see the old pointer.
    // the relationship is:
    // pointer swap -> membarrier -> thread's update of his own state -> thread's load of the rcu protected pointer
    // thus, all such threads are guaranteed to see the new pointer, and we can thus ignore them when waiting for all existing
    // users.

    // lock the reset sync lock for reading.
    //
    // this ensures that if any reset operation is currently ongoing, we don't interrupt it by incremented the epoch id while it
    // is being reset, and we instead wait for it to finish and only then go on with our increment.
    //
    // this exclusivity is guaranteed since during reset the leader of the reset locks the reset sync lock for writing.
    //
    // this also ensures that a reset operation is not initiated while we are still waiting for threads to see our incremented epoch id,
    // since we hold this until we finish waiting.
    let mut reset_sync_read_guard = EPOCH_ID_RESET_SYNC_LOCK.read().await;

    // increment the epoch id.
    //
    // this is used as a communication primitive with the worker threads.
    // worker threads will then update their last seen epoch id by reading the global epoch id every time they pass through a quiescent
    // state.
    //
    // we can then sample their published last seen epoch id to know when they saw our increment, and once they did, we know that they
    // passed through a quiescent state.
    let new_epoch_id = match epoch_id_inc() {
        Ok(v) => v,
        Err(err) => {
            // epoch id overflow.

            // perform a reset of the epoch id
            if err.am_i_the_leader {
                // re-lock the reset sync lock for writing.
                //
                // once we succeed grabbing the write lock, it is guaranteed that:
                // - all previous waiters finished waiting for their grace period
                // - all non-leader waiters that also entered reset mode have started listening to the reset
                // - all new waiters will be blocked until we finish.
                drop(reset_sync_read_guard);
                let reset_sync_write_guard = EPOCH_ID_RESET_SYNC_LOCK.write().await;

                // reset the epoch id
                epoch_id_set(EPOCH_ID_MIN, atomic::Ordering::Relaxed);

                // make sure that all threads see the reset of the epoch id.
                membarrier::perform();

                // wait for all threads to update their last seen epoch id to the reset value.
                //
                // note that parked and not yet started threads are not relevant here, since once they wake up they will see the updated
                // reset value of the epoch id due to the membarrier, and they will fetch and publish it along with the enabling of the
                // busy flag as soon as they unpark.
                wait_for_running_threads_to_see_epoch_id(
                    |last_seen_epoch_id| last_seen_epoch_id == EPOCH_ID_MIN,
                    false,
                )
                .await;

                // at this point, all running threads have reset their last seen epoch id, and new threads are guaranteed
                // to see at least the reset value.

                // now that we finished resetting the epoch id, we can now let new waiters in.
                drop(reset_sync_write_guard);

                // wake all non-leader waiters that are in reset mode waiting for us to finish.
                RESET_FINISHED_NOTIFICATION.notify();
            } else {
                // start listening to reset notification from the leader.
                //
                // this must be done before dropping the read lock, so that the leader doesn't start acting before we are listening
                // to notifications from him.
                //
                // as for the overflow behaviour of `notified`, the time window where we hold the returned future before awaiting it
                // is very small, so we shouldn't expect overflow to occur here.
                let event = RESET_FINISHED_NOTIFICATION.notified();

                // let the leader start doing its thing.
                drop(reset_sync_read_guard);

                // wait for the leader to finish the reset operation and notify us.
                event.await;
            }

            // done resetting epoch id

            // re-lock the reset sync guard just in case, even though we shouldn't expect another reset any time soon.
            // note that the lock should be unlocked now since the writer unlocks it before waking us up.
            reset_sync_read_guard = EPOCH_ID_RESET_SYNC_LOCK.try_read().unwrap_or_else(|_| {
                panic!("another epoch id reset right after the previous reset")
            });

            let Ok(new_epoch_id) = epoch_id_inc() else {
                // avoid poisoning the lock
                drop(reset_sync_read_guard);

                // we should never get another overflow right after we finish resetting.
                // the epoch id should take some time to grow before it wraps around again.
                panic!("overflow when incrementing epoch id after reset")
            };

            new_epoch_id
        }
    };

    // note that parked and not-yet-started threads are irrelevant here since they are guaranteed to see the new pointer
    // due to the membarrier.
    wait_for_running_threads_to_see_epoch_id(
        |last_seen_epoch_id| last_seen_epoch_id >= new_epoch_id,
        include_calling_thread,
    )
    .await;

    // ensure that the reset sync read guard is held up until this point.
    // this is important to make sure that a reset operation is not initiated while we are still waiting for threads to see our new
    // epoch id, otherwise we would keep waiting until the next overflow of the epoch id.
    drop(reset_sync_read_guard);
}

/// wait for all other threads in the process other than the current thread to see some epoch id as implemented in the given predicate
/// which processes the last seen epoch id of each thread.
///
/// this function does not take into account new threads just starting, nor new threads just existing the busy state.
///
/// if `include_calling_thread` is set, this function also waits for the calling thread itself to see the updated epoch id as implemented
/// in the given predicate. this is usually not needed and should be set to `false`. see [`synchronize_rcu`] for more info.
async fn wait_for_running_threads_to_see_epoch_id<F: Fn(EpochId) -> bool>(
    last_seen_epoch_id_predicate: F,
    include_calling_thread: bool,
) {
    loop {
        // start subscribing to the notified waiters event before checking the current state.
        //
        // if we first check the state and only then start listening, there may be a small window after we finish
        // checking the values but before we start listening where some thread updates its counter and notifies
        // all wakers, but we will miss that notification, which is problematic.
        //
        // so, we start listening before checking the values, so that even notifications that are issued while
        // or right after we finished checking are still received.
        //
        // note that this registration operation provides acquire ordering against any previous notifiers, so we won't miss
        // any state updates.
        // to prove this, we can split our situation with the readers into 2 cases:
        // 1. a thread already notified before we registered.
        // 2. a thread hasn't already notified when we registered.
        // in case 1, we are guaranteed to see this thread's state update since the notify operation has release ordering, and paired
        // with the acquire ordering of our registration, it guarantees that we see the state update as happened before the notify
        // operation.
        // in case 2, we are guaranteed to at some point see either the state update or the notification, since the notification
        // hasn't yet been observed by us.
        //
        // as for the overflow behaviour of `notified`, the time window where we hold the returned future before awaiting it
        // is very small, so we shouldn't expect overflow to occur here.
        let notified = THREAD_EPOCH_UPDATED_NOTIFY.notified();

        // we must re-calculate this every iteration since our task may be sent between threads every time we await the notified future.
        let this_thread_storage_slot_id = this_thread_get_storage_slot_id();

        // check if all threads have seen our new epoch id
        if thread_storage_slot_get_all()
            .iter_enumerated()
            .all(|(storage_slot_id, storage_slot)| {
                if !include_calling_thread
                    && unlikely(storage_slot_id == this_thread_storage_slot_id)
                {
                    // this slot represents the current thread.
                    // we may or may not need to wait for ourselves, depending on the caller's choice.
                    return true;
                }
                let encoded_state = storage_slot.state.load(
                    // we use acquire ordering paired with a release ordering for the store to make sure that the stores to the data
                    // pointed at by the rcu protected pointer happen before we see the store to the state.
                    // this is important in order to guarantee that we don't see those writes after we free the protected pointer, which will
                    // lead to a UAF.
                    atomic::Ordering::Acquire,
                );

                let Some(state) = ThreadState::decode(encoded_state) else {
                    // if the slot is empty, ignore it.
                    // it may at some point be allocated by some new thread that just started, but in this function we explicitly ignore
                    // new threads.
                    return true;
                };

                if !state.is_busy {
                    // this thread is currently not busy running any future.
                    // it may start running as soon as we finished checking it, but in this function we explicitly ignore non busy threads.
                    return true;
                }

                last_seen_epoch_id_predicate(state.last_seen_epoch_id)
            })
        {
            // all threads saw our new epoch id, we are done waiting
            break;
        }

        // some of the threads haven't yet seen our new epoch id.
        // so, wait for them to go through a quiescent state and see our new epoch id, or to go to sleep.
        notified.await;
    }
}

/// "see" a new epoch id in the current thread.
/// this fetches the current epoch id with a proper memory ordering - a release memory ordering, which provides the required
/// guaranteed, for example it guarantees that once we see an updated epoch id, we see the swap of the rcu protected pointer
/// as happened before that store to the epoch id.
fn this_thread_see_new_epoch_id() -> EpochId {
    epoch_id_get(
        // we use acquire ordering coupled with a release ordering when incrementing the epoch id to make sure that we see swap of the rcu
        // protected pointer before we see the increment of the epoch id.
        //
        // if we were to first see the increment of the epoch id, and only then see the swap of the pointer, we may publish that we have
        // seen the new epoch id, causing the waiter to free the memory, and then still use the old and now freed pointer since we haven't
        // yet seen the pointer swap.
        atomic::Ordering::Acquire,
    )
}

fn on_thread_stop() {
    // note that at this point, this thread may or may not have a slot allocated to it.
    // this hook is called by both tokio worker threads, and tokio blocking threads.
    // blocking threads will not have a slot at all, since we only allocate a slot in the `on_before_task_poll` hook.
    // tokio worker threads may or may not have a slot, depending on whether they have polled any task throughout their
    // lifetime.
    this_thread_dealloc_storage_slot();
}

fn on_thread_park() {
    // the `on_thread_park` hook may be called before a slot is allocated, since a slot is only allocated in `on_before_task_poll`,
    // but a worker thread may decide to park even before polling its first future, for example if there are no tasks to be executed by it.
    if unlikely(!this_thread_does_have_allocated_storage_slot()) {
        return;
    }

    {
        let storage_slot = &thread_storage_slot_get_all()[this_thread_get_storage_slot_id()];

        // mark this thread as non-busy.
        storage_slot.state.fetch_and(
            !1,
            // no special ordering needed here.
            // note that this relaxed store doesn't break the release-sequence of this variable (see c++ memory model for more
            // info), so it doesn't prevent the loader from synchronizing with any previous release ordered store.
            atomic::Ordering::Relaxed,
        );
    }

    // wake all waiters since some waiters may be waiting for us to see their new epoch id, and we are instead going to sleep
    // so we will never see it.
    // wake them so that they will see that we are no longer busy and thus we are no longer using any of their rcu protected
    // pointers.
    THREAD_EPOCH_UPDATED_NOTIFY.notify();
}

fn on_thread_unpark() {
    // the `on_thread_unpark` hook may be called before a slot is allocated, since a slot is only allocated in `on_before_task_poll`,
    // but a worker thread may decide to park (and then unpark) even before polling its first future, for example if there are no tasks to
    // be executed by it.
    if unlikely(!this_thread_does_have_allocated_storage_slot()) {
        return;
    }

    // note that in addition to setting the is busy flag here, we also need to see a new epoch id.
    //
    // this is needed for the case where a reset operation was performed since we last went to sleep.
    // in that case, if we wake up and set the busy flag without updating the epoch id, some thread that had already incremented
    // the epoch id since the reset may think that we saw his epoch id increment since we have a stale high epoch id value, even
    // though in practice we didn't really see his epoch id increment.
    let new_seen_epoch_id = this_thread_see_new_epoch_id();

    let storage_slot = &thread_storage_slot_get_all()[this_thread_get_storage_slot_id()];
    storage_slot.state.store(
        ThreadState {
            last_seen_epoch_id: new_seen_epoch_id,
            is_busy: true,
        }
        .encode(),
        // we use release ordering to make sure that all writes to the data pointed at by the rcu protected pointer happen before this
        // store so that no writes happen after the data is freed.
        // this is needed since we actually fetch a new epoch id here, not only set the busy flag.
        atomic::Ordering::Release,
    );
}

fn on_before_task_poll() {
    // note that we only allocate in the `on_before_task_poll` hook, instead of the more reasonable `on_thread_start` hook,
    // since the `on_thread_start` hook is also called by blocking threads, but we only want to account for tokio worker
    // threads in our rcu book-keeping.
    // and, the `on_before_task_poll` hook is obviously only called for tokio worker threads, so it is the ideal place to
    // perform the slot allocation.
    if likely(this_thread_does_have_allocated_storage_slot()) {
        return;
    }

    let epoch_id = this_thread_see_new_epoch_id();
    this_thread_alloc_storage_slot(ThreadState {
        last_seen_epoch_id: epoch_id,
        is_busy: true,
    });
}

fn on_after_task_poll() {
    let new_seen_epoch_id = this_thread_see_new_epoch_id();

    // extra scope to scope the lifetime of the read guard of the storage slots buffer
    let prev_state_encoded = {
        let storage_slot = &thread_storage_slot_get_all()[this_thread_get_storage_slot_id()];
        // at this point we want to swap the current state with the new state.
        // we could do that using the atomic `swap` operation, but we can do something more performant while still maintaining correctness.
        //
        // the slot's data is loaded from multiple threads, but it is only written to by the current thread who owns that slot.
        // we can use that fact to split the atomic `swap` operation into a `load` and then a `store`, while still being guaranteed that no
        // one will modify the value between the `load` and the `store`, since the current thread are the only one allowed to modify the
        // value.
        //
        // as for why this is more efficient, the load-then-store method requires looser memory ordering guarantees, and thus provides more
        // flexibility for optimization by the hardware's memory subsystem.
        //
        // for example, on x86, the load then store will be translated to just 2 simple `MOV` instructions, while a `swap` would have been
        // translated to a `LOCK XCHG` instruction, which requires much more effort from the hardware.
        let prev_state_encoded = storage_slot.state.load(
            // we don't need any special ordering, since this thread is the only entity which can write to this variable.
            // so, the returned value is sequentially consistent with the execution order of the code in this thread.
            //
            // also, we don't need to synchronize this load against any other shared variables, since the returned value is only used to
            // check whether it was different than the newly written value, and is thus not used in combination with any other shared state.
            atomic::Ordering::Relaxed,
        );
        storage_slot.state.store(
            ThreadState {
                last_seen_epoch_id: new_seen_epoch_id,
                is_busy: true,
            }
            .encode(),
            // we use release ordering to make sure that all writes to the data pointed at by the rcu protected pointer happen before this
            // store so that no writes happen after the data is freed.
            atomic::Ordering::Release,
        );

        prev_state_encoded
    };

    let prev_state = ThreadState::decode(prev_state_encoded).unwrap();

    // we are expected to be in the busy state while not parked
    debug_assert!(prev_state.is_busy);

    if unlikely(prev_state.last_seen_epoch_id != new_seen_epoch_id) {
        // if the last seen epoch id changed, some waiter may now be able to finish waiting. so, notify all waiters.
        THREAD_EPOCH_UPDATED_NOTIFY.notify();
    }
}

/// extension methods for tokio's runtime builder.
pub trait TokioRuntimeBuilderExt {
    /// enable rcu support for this tokio runtime.
    /// must be called when constructing the runtime in order to use any rcu related primitive inside the runtime.
    ///
    /// # Safety
    ///
    /// when used, in order to use any of the rcu primitives safely, you must wrap the [`rcu_block_on`](TokioRuntimeExt::rcu_block_on)
    /// function to run the main future on the runtime. using [`block_on`](tokio::runtime::Runtime::block_on) directly is forbidden.
    ///
    /// furthermore, after calling this function, you must not register any tokio hooks of your own, since this functions registers the
    /// rcu hooks needed for book-keeping. overriding any of those hooks will lead to undefined behaviour.
    unsafe fn enable_rcu(&mut self) -> &mut Self;
}

impl TokioRuntimeBuilderExt for tokio::runtime::Builder {
    unsafe fn enable_rcu(&mut self) -> &mut Self {
        assert!(membarrier::is_supported());
        membarrier::register();

        self.on_before_task_poll(|_| {
            on_before_task_poll();
        })
        .on_thread_stop(|| {
            on_thread_stop();
        })
        .on_thread_park(|| {
            on_thread_park();
        })
        .on_thread_unpark(|| {
            on_thread_unpark();
        })
        .on_after_task_poll(|_| {
            on_after_task_poll();
        })
    }
}

/// extension methods for tokio's runtime.
pub trait TokioRuntimeExt {
    /// runs a future to completion on the tokio runtime, with RCU support.
    ///
    /// this can only be used with multi-threaded runtimes.
    ///
    /// # Safety
    ///
    /// to use this, you must first call [`enable_rcu`](TokioRuntimeBuilderExt::enable_rcu) when building the runtime.
    unsafe fn rcu_block_on<F: Future>(&self, future: F) -> F::Output;
}
impl TokioRuntimeExt for tokio::runtime::Runtime {
    unsafe fn rcu_block_on<F: Future>(&self, future: F) -> F::Output {
        // rcu is only supported for multithreaded runtimes
        assert_eq!(self.handle().runtime_flavor(), RuntimeFlavor::MultiThread);

        self.block_on(unsafe {
            // SAFETY: we pass the wrapped future directly to `block_on`
            RcuRootFuture::new(future)
        })
    }
}

/// runs the provided future inside a new multi-threaded tokio runtime with all features enabled and with rcu support.
pub fn rcu_block_on<F: Future>(future: F) -> F::Output {
    unsafe {
        // SAFETY: we use `rcu_block_on`
        let rt = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .enable_rcu()
            .build()
            .unwrap();

        // SAFETY: we called `enable_rcu`
        rt.rcu_block_on(future)
    }
}

/// a wrapper around the root future of a tokio `block_on` call.
///
/// this is required since tokio's hooks only apply to tokio's worker threads, but not to the main thread which initially calls `block_on`.
///
/// but, we need the main thread to also perform the book-keeping needed by the rcu primitive, in order for it to be able use the rcu
/// primitives and to interact with the other threads using the rcu primitives.
///
/// so, we wrap the main future passed to `block_on` in a custom wrapper which emulates the call to the different worker hooks.
/// this lets the main thread participate in the book-keeping like any other worker thread.
#[derive(Debug, Clone, Copy)]
struct RcuRootFuture<F> {
    inner_future: F,
    has_already_been_polled: bool,
}
impl<F> RcuRootFuture<F> {
    /// wraps the provided future with the rcu root future logic.
    ///
    /// # Safety
    ///
    /// may only be used to wrap the future provided to tokio's `block_on` function on a multithreaded runtime.
    /// using this incorrectly will lead to undefined behaviour.
    unsafe fn new(inner_future: F) -> Self {
        Self {
            inner_future,
            has_already_been_polled: false,
        }
    }
}
impl<F: Future> Future for RcuRootFuture<F> {
    type Output = F::Output;

    // #[inline] is important here since it increases the chance that some of the redundant branches performed in this function
    // will be eliminated, for example the `has_already_been_polled` branch, which is only used to track the first call to
    // `poll` and can easily be eliminated by unrolling the first iteration of the poll loop.
    #[inline]
    fn poll(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Self::Output> {
        if unlikely(!self.has_already_been_polled) {
            // first time being polled on the main thread.

            // SAFETY: we don't move out of anything
            unsafe { self.as_mut().get_unchecked_mut().has_already_been_polled = true }
        } else {
            // we have already been polled in the previous iteration.
            //
            // we have returned `Poll::Pending` in the previous iteration, so the main thread parked itself and went to sleep,
            // and now we are being polled again.
            //
            // this is basically an unpark.
            on_thread_unpark();
        }

        // before polling the task
        on_before_task_poll();

        // SAFETY: we do not move out of anything, we just project a field, which is safe
        let inner_future = unsafe { self.map_unchecked_mut(|x| &mut x.inner_future) };

        let res = inner_future.poll(cx);

        // just finished polling the task.
        on_after_task_poll();

        match res {
            Poll::Ready(_) => {
                // in this case, the main future is done, so the main thread is also done.
                on_thread_stop();
            }
            Poll::Pending => {
                // if we return `Poll::Pending`, the main thread will park itself until an IO event occurs and wakes it up.
                //
                // so this is basically a thread park.
                on_thread_park();
            }
        }

        res
    }
}