hybrid-rc 0.1.0

/*!
 * Thread-safe hybrid reference counting pointers
 *
 * Loosely based on the algorithm described in
 * ["Biased reference counting: minimizing atomic operations in garbage collection"][doi:10.1145/3243176.3243195]
 * by Jiho Choi et. al. but adapted to Rust's type system and its lack of a managed runtime
 * environment.
 *
 * The type `HybridRc<T, State>` provides thread-safe shared ownership of a value of type `T`
 * allocated on the heap, just like `std::sync::Arc<T>` does. The main difference is that one
 * thread at a time can use non-atomic reference counting for better performance. That means that
 * `HybridRc` is especially suited for workloads where one thread accesses the shared value
 * significantly more often than others.
 *
 * There a two variants of [`HybridRc`]:
 * - `HybridRc<T, `[`Local`]`>` (type aliased as [`Rc`]): very fast but only usable on one thread.
 * - `HybridRc<T, `[`Shared`]`>` (type aliased as [`Arc`]): slower but universally usable.
 *
 * Instances of both variants are convertible into each other. Especially, an `Rc` can always be
 * converted into an `Arc` using [`HybridRc::to_shared(&rc)`] or [`.into()`].
 *
 * An `Arc` on the other hand can only be converted into an `Rc` using [`HybridRc::to_local(&arc)`]
 * or [`.try_into()`] if no other thread has `Rc`s for the same value. The thread holding `Rc`s to
 * a value is called the "owner thread". Once all `Rc`s are dropped, the shared value becomes
 * ownerless again.
 *
 * `HybridRc` is designed as a drop-in replacement for `std::sync::Arc` and `std::rc::Rc`, so except
 * for the conversion functionality outlined above the usage is similar to these and other smart
 * pointers.
 *
 * # Thread Safety
 *
 * `HybridRc` uses two separate reference counters - one modified non-atomically and one using
 * atomic operations - and keeps track of a owner thread that is allowed to modify the "local"
 * reference counter. This means that it is thread-safe, while one thread is exempted from
 * the disadvantage of atomic operations being more expensive than ordinary memory accesses.
 *
 *
 * # Examples
 *
 * Multiple threads need a reference to a shared value while one thread needs to clone references
 * to the value significantly more often than the others.
 * ```
 * use hybrid_rc::{Rc, Arc};
 * use std::thread;
 * use std::sync::mpsc::channel;
 *
 * # type SomeComplexType = std::collections::BinaryHeap<()>;
 * # fn expensive_computation<T>(x: impl AsRef<T>, i: i32) -> i32 { let _ = x.as_ref(); i }
 * # fn do_something<T>(x: impl AsRef<T>, _i: i32) { let _ = x.as_ref(); }
 * # fn main() -> Result<(), Box<dyn std::any::Any + Send + 'static>> {
 * let local = Rc::new(SomeComplexType::new());
 * let (sender, receiver) = channel();
 *
 * // Spawn of threads for multiple expensive computations
 * for i in 1..=4 {
 * 	let sender = sender.clone();
 * 	let shared = Rc::to_shared(&local);
 * 	thread::spawn(move || {
 * 		sender.send(expensive_computation(shared, i));
 * 	});
 * }
 *
 * // Do something that needs single-thread reference counting
 * for i in 1..=1000 {
 * 	do_something(local.clone(), i);
 * }
 *
 * // Collect expensive computation results
 * for i in 1..=4 {
 * 	println!("{:?}", receiver.recv().unwrap());
 * }
 * # Ok(())
 * # }
 * ```
 *
 * A library wants to give library consumers flexibility for multithreading but also internally
 * have the performance of `std::rc::Rc` for e.g. a complex tree structure that is mutated on
 * the main thread.
 * ```
 * use hybrid_rc::Rc;
 * use std::thread;
 *
 * # fn get_local_hybridrc_from_some_library() -> Rc<()> { Rc::default() }
 * # fn do_something(_: &()) { }
 * # fn main() -> Result<(), Box<dyn std::any::Any + Send + 'static>> {
 * let reference = get_local_hybridrc_from_some_library();
 * let shared = Rc::to_shared(&reference);
 *
 * // do the work in another thread
 * let worker = thread::spawn(move || {
 * 	do_something(&*shared);
 * });
 *
 * // Do something useful with the library
 *
 * worker.join()?;
 * # Ok(())
 * # }
 * ```
 *
 * [`HybridRc::to_shared(&rc)`]: HybridRc::to_shared
 * [`HybridRc::to_local(&arc)`]: HybridRc::to_local
 * [`.into()`]: HybridRc#impl-From<HybridRc<T%2C%20Local>>
 * [`.try_into()`]: HybridRc#impl-TryFrom<HybridRc<T%2C%20Shared>>
 * [doi:10.1145/3243176.3243195]: https://dl.acm.org/doi/10.1145/3243176.3243195
 */

use core::any::Any;
use core::borrow::Borrow;
use core::cell::Cell;
use core::convert::Infallible;
use core::convert::TryFrom;
use core::hash::{Hash, Hasher};
use core::marker::PhantomData;
use core::ops::Deref;
use core::ptr::NonNull;
use core::sync::atomic;
use core::sync::atomic::Ordering;
use core::{fmt, mem, ptr};
use std::error::Error;
use std::panic::{RefUnwindSafe, UnwindSafe};

mod thread_id;
use thread_id::{AtomicOptionThreadId, ThreadId};

/// Provides a senitel pointer value for dangling `Weak`s.
///
/// This is not NULL to allow optimizations through [`NonNull`] but cannot ever be a valid pointer
/// to a [`RcBox`].
#[inline]
const fn senitel<T>() -> NonNull<T> {
	unsafe { NonNull::new_unchecked(usize::MAX as *mut T) }
}

/// Checks if the provided pointer is the [`senitel`]
#[inline]
fn is_dangling<T: ?Sized>(ptr: *mut T) -> bool {
	ptr.cast::<()>() == senitel().as_ptr()
}

/// Internal module for non-public definition of `RcState`.
mod state_trait {
	use core::fmt::Debug;

	/// Internal trait for type-level enumeration of `Shared` and `Local`.
	pub trait RcState: Debug {
		const SHARED: bool;
	}
}
use state_trait::RcState;

/// Marker types for the states of a [`HybridRc`]
pub mod state {
	/// Marks a [`HybridRc`] as shared.
	///
	/// `HybridRc<_, Shared>` atomically updates the shared reference counter.
	///
	/// # See also
	/// - [`Local`]
	///
	/// [`HybridRc`]: super::HybridRc
	#[derive(Debug, Clone, Copy)]
	pub enum Shared {}
	impl super::RcState for Shared {
		const SHARED: bool = true;
	}

	/// Marks a [`HybridRc`] as local.
	///
	/// `HybridRc<_, Local>` non-atomically updates the local reference counter.
	///
	/// # See also
	/// - [`Shared`]
	///
	/// [`HybridRc`]: super::HybridRc
	#[derive(Debug, Clone, Copy)]
	pub enum Local {}
	impl super::RcState for Local {
		const SHARED: bool = false;
	}
}
use state::{Local, Shared};

/// An enumeration of possible errors when upgrading a [`Weak`].
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum UpgradeError {
	/// The referenced value was already dropped because no strong references to it exists anymore.
	ValueDropped,
	/// The requested action would have created a new [`Rc`] while at least one `Rc` still existed
	/// on another thread.
	WrongThread,
}

impl fmt::Display for UpgradeError {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		match *self {
			Self::ValueDropped => f.write_str("value was already dropped"),
			Self::WrongThread => {
				f.write_str("tried to get a local reference while another thread was the owner")
			}
		}
	}
}

impl Error for UpgradeError {}

impl From<Infallible> for UpgradeError {
	fn from(x: Infallible) -> UpgradeError {
		match x {}
	}
}

/// Metadata part of a shared allocation.
#[derive(Debug)]
struct RcMeta {
	/// Id for the thread which may use local references
	owner: AtomicOptionThreadId,
	/// Strong local reference count
	strong_local: Cell<usize>,
	/// Strong shared reference count (+ 1 for all strong local references combined)
	strong_shared: atomic::AtomicUsize,

	/// Weak reference count (+ 1 for all strong references combined)
	///
	/// If `usize::MAX`, the ability to downgrade strong pointers is temporarily locked to avoid
	/// races in `get_mut()`.
	weak: atomic::AtomicUsize,
}

/// Heap struct for shared allocations of `T`.
///
/// `repr(C)` to future-proof against possible layout optimizations which could interfere with
/// `[into|from]_raw()` of transmutable data types.
#[repr(C)]
struct RcBox<T: ?Sized> {
	meta: RcMeta,
	data: T,
}

impl RcMeta {
	/// Increments the local reference counter unconditionally.
	///
	/// *Only safe to use on the owner thread and as long as at least one local reference exists.*
	///
	/// # Panics
	/// Panics if the counter overflowed.
	#[inline(always)]
	fn inc_strong_local(&self) {
		let counter = self.strong_local.get();

		if counter == usize::MAX {
			panic!("reference counter overflow");
		}

		self.strong_local.set(counter + 1);
	}

	/// Increment the local reference counter.
	///
	/// Also adjusts the shared reference counter if neccessary.
	///
	/// Fails if this would resurrect an already dropped
	/// value.
	///
	/// *Only safe to use on the owner thread.*
	///
	/// # Panics
	/// Panics if one of the counters overflowed.
	#[inline]
	fn try_inc_strong_local(&self) -> Result<(), ()> {
		let counter = self.strong_local.get();

		if counter == usize::MAX {
			panic!("reference counter overflow");
		} else if counter == 0 {
			self.try_inc_strong_shared()?;
		}

		self.strong_local.set(counter + 1);
		Ok(())
	}

	/// Decrements the local reference counter.
	///
	/// Also adjusts the shared reference counter and
	/// the `owner` if neccessary.
	///
	/// Returns **true** if no strong references remain at all.
	///
	/// *Only safe to use on the owner thread.*
	///
	/// # Panics
	/// Panics if the shared reference counter was already zero.
	#[inline(always)]
	fn dec_strong_local(&self) -> bool {
		let counter = self.strong_local.get();
		self.strong_local.set(counter - 1);
		if counter == 1 {
			self.remove_last_local_reference()
		} else {
			false
		}
	}

	/// Decrements the shared counter and sets the `owner` to `None`.
	///
	/// Used internally by `dec_strong_local()`
	///
	/// # Panics
	/// Panics if the counter was already zero.
	fn remove_last_local_reference(&self) -> bool {
		let old_shared = self.strong_shared.fetch_sub(1, Ordering::Release);
		if old_shared == 0 {
			panic!("reference counter underflow");
		}
		self.owner.store(None, Ordering::Release);
		old_shared == 1
	}

	/// Increments the shared reference counter unconditionally.
	///
	/// *Only safe to use as long as at least one shared reference exists.*
	///
	/// # Panics
	/// Panics if the counter overflowed.
	#[inline]
	fn inc_strong_shared(&self) {
		let old_counter = self.strong_shared.fetch_add(1, Ordering::Relaxed);
		if old_counter == usize::MAX {
			panic!("reference counter overflow");
		}
	}

	/// Increments the shared reference counter.
	///
	/// Also adjusts the shared reference counter and the `owner` if neccessary.
	///
	/// Fails if this would resurrect an already dropped value.
	///
	/// # Panics
	/// Panics if the counter overflowed.
	#[inline]
	fn try_inc_strong_shared(&self) -> Result<(), ()> {
		self.strong_shared
			.fetch_update(
				Ordering::Relaxed,
				Ordering::Relaxed,
				|old_counter| match old_counter {
					0 => None,
					usize::MAX => panic!("reference counter overflow"),
					_ => Some(old_counter + 1),
				},
			)
			.map(|_| ())
			.map_err(|_| ())
	}

	/// Decrements the shared reference counter.
	///
	/// Returns **true** if no strong references remain at all.
	///
	/// # Panics
	/// Panics if the counter was already zero.
	#[inline]
	fn dec_strong_shared(&self) -> bool {
		let old_counter = self.strong_shared.fetch_sub(1, Ordering::Release);
		if old_counter == 0 {
			panic!("reference counter underflow");
		}
		old_counter == 1 && self.owner.load(Ordering::Relaxed).is_none()
	}

	/// Increments the weak reference counter.
	///
	/// # Panics
	/// Panics if the counter overflowed or was already zero.
	#[inline]
	fn inc_weak(&self) {
		const MAX_COUNT: usize = usize::MAX - 1;
		let mut counter = self.weak.load(Ordering::Relaxed);

		// CAS loop
		loop {
			match counter {
				usize::MAX => {
					core::hint::spin_loop();
					counter = self.weak.load(Ordering::Relaxed);
					continue;
				}
				MAX_COUNT => panic!("weak counter overflow"),
				0 => panic!("BUG: weak resurrection of dead counted reference"),
				_ => {
					let result = self.weak.compare_exchange_weak(
						counter,
						counter + 1,
						Ordering::Acquire,
						Ordering::Relaxed,
					);
					match result {
						Ok(_) => break,
						Err(old) => counter = old,
					}
				}
			}
		}
	}

	/// Increments the weak reference counter (without a spin loop).
	///
	/// # Panics
	/// Panics if the counter is locked, overflowed or was already zero.
	#[inline]
	fn inc_weak_nolock(&self) {
		const MAX_COUNT: usize = usize::MAX - 1;
		match self.weak.fetch_add(1, Ordering::Relaxed) {
			usize::MAX => panic!("BUG: weak counter locked"),
			MAX_COUNT => panic!("weak counter overflow"),
			0 => panic!("BUG: weak resurrection of dead counted reference"),
			_ => (),
		}
	}

	/// Decrements the weak reference counter.
	///
	/// Returns **true** if the counter reached zero.
	///
	/// # Panics
	/// Panics if the counter was already zero.
	#[inline]
	fn dec_weak(&self) -> bool {
		let old_counter = self.weak.fetch_sub(1, Ordering::Release);
		if old_counter == 0 {
			panic!("weak counter underflow");
		}
		old_counter == 1
	}

	/// Checks if there is only one unique reference.
	///
	/// Temporarily locks the weak reference counter to prevent race conditions.
	#[inline]
	fn has_unique_ref(&self) -> bool {
		let result =
			self.weak
				.compare_exchange(1, usize::MAX, Ordering::Acquire, Ordering::Relaxed);
		if result.is_ok() {
			let mut count = self.strong_shared.load(Ordering::Acquire);

			if count == 1 {
				let owner = self.owner.load(Ordering::Relaxed);
				if owner == Some(ThreadId::current_thread()) {
					count = self.strong_local.get();
				}
			}

			self.weak.store(1, Ordering::Release);

			count == 1
		} else {
			false
		}
	}
}

/// A hybrid reference-counting pointer.
///
/// - [`HybridRc<T, Shared>`][Arc] behaves mostly like [`std::sync::Arc`]
/// - [`HybridRc<T, Local>`][Rc] behaves mostly like [`std::rc::Rc`].
///
/// See the [module-level documentation][crate] for more details.
///
/// The inherent methods of `HybridRc` are all associated functions, which means that you have to
/// call them as e.g. [`HybridRc::get_mut(&mut x)`] instead of `x.get_mut()`. This avoids conflicts
/// with methods of the inner type `T`.
///
/// [`HybridRc::get_mut(&mut x)`]: Self::get_mut
#[must_use]
pub struct HybridRc<T: ?Sized, State: RcState> {
	ptr: NonNull<RcBox<T>>,
	phantom: PhantomData<(State, RcBox<T>)>,
}

/// Type alias for a local reference counting pointer.
///
/// Provided to ease migrating from [`std::rc::Rc`].
///
/// See the [module-level documentation][crate] for more details.
///
/// The inherent methods of `Rc` are all associated functions, which means that you have to call
/// them as e.g. [`Rc::to_shared(&x)`] instead of `x.to_shared()`. This avoids conflicts with
/// methods of the inner type `T`.
///
/// [`Rc::to_shared(&x)`]: Self::to_shared
pub type Rc<T> = HybridRc<T, Local>;

/// Type alias for a shared reference counting pointer.
///
/// Provided to ease migrating from [`std::sync::Arc`].
///
/// See the [module-level documentation] for more details.
///
/// The inherent methods of `Arc` are all associated functions, which means that you have to call
/// them as e.g. [`Arc::to_local(&x)`] instead of `x.to_local()`. This avoids conflicts with
/// methods of the inner type `T`.
///
/// [`Arc::to_local(&x)`]: Self::to_local
/// [module-level documentation]: crate
pub type Arc<T> = HybridRc<T, Shared>;

impl<T: ?Sized, State: RcState> HybridRc<T, State> {
	/// Creates a new `HybridRc` from a pointer to a shared allocation.
	///
	/// The reference counters must have been updated by the caller.
	#[inline(always)]
	fn from_inner(ptr: NonNull<RcBox<T>>) -> Self {
		Self {
			ptr,
			phantom: PhantomData,
		}
	}

	/// Provides a reference to the inner value.
	#[inline(always)]
	fn data(&self) -> &T {
		// Safety: as long as one HybridRc or Weak for this item exists, the memory stays allocated.
		unsafe { &(*self.ptr.as_ptr()).data }
	}

	/// Provides a reference to the shared metadata.
	#[inline(always)]
	fn meta(&self) -> &RcMeta {
		// Safety: as long as one HybridRc or Weak for this item exists, the memory stays allocated.
		unsafe { &(*self.ptr.as_ptr()).meta }
	}

	/// Returns a mutable reference to the value, without checking for uniqueness.
	///
	/// # See also
	/// - [`get_mut()`], which is safe.
	///
	/// # Safety
	/// No other `HybridRc` or [`Weak`] for the same value must be dereferenced for the duration of
	/// the returned borrow.
	///
	/// [`get_mut()`]: Self::get_mut
	#[must_use]
	#[inline]
	pub unsafe fn get_mut_unchecked(this: &mut Self) -> &mut T {
		unsafe { &mut (*this.ptr.as_ptr()).data }
	}

	/// Returns a mutable reference to the value, iff the value is not shared
	/// with another `HybridRc` or [`Weak`].
	///
	/// Returns `None` otherwise.
	#[must_use]
	pub fn get_mut(this: &mut Self) -> Option<&mut T> {
		if this.meta().has_unique_ref() {
			unsafe { Some(Self::get_mut_unchecked(this)) }
		} else {
			None
		}
	}

	/// Provides a raw pointer to the referenced value
	///
	/// The counts are not affected in any way and the `HybridRc` is not consumed.
	/// The pointer is valid for as long there exists at least one `HybridRc` for the value.
	#[must_use]
	#[inline]
	pub fn as_ptr(this: &Self) -> *const T {
		this.data()
	}

	/// Creates a new [`Weak`] for the referenced value.
	///
	/// # Example
	/// ```
	/// use hybrid_rc::{Rc, Weak};
	///
	/// let strong = Rc::new(42i32);
	/// let weak = Rc::downgrade(&strong);
	///
	/// assert_eq!(Rc::as_ptr(&strong), Weak::as_ptr(&weak));
	/// ```
	#[inline]
	pub fn downgrade(this: &Self) -> Weak<T> {
		this.meta().inc_weak();
		Weak { ptr: this.ptr }
	}

	/// Drops the contained value and also drops the shared `RcBox` if there are no other `Weak`
	/// references.
	///
	/// # Safety
	/// Only safe to use in `drop()` or a consuming function after verifying that no other strong
	/// reference exists. Otherwise after calling this e.g. dereferencing the `HybridRc` WILL
	/// cause undefined behaviour and even dropping it MAY cause undefined behaviour.
	unsafe fn drop_contents_and_maybe_box(&mut self) {
		// Safety: only called if this was the last strong reference
		unsafe {
			ptr::drop_in_place(Self::get_mut_unchecked(self));
		}

		if self.meta().dec_weak() {
			// Safety: ManuallyDrop doesn't change the memory layout
			let ptr: *mut RcBox<mem::ManuallyDrop<T>> =
				unsafe { mem::transmute(self.ptr.as_mut()) };
			// Safety: only called if this was the last (weak) reference so only called once on a
			// valid pointer that was once a leaked box.
			mem::drop(unsafe { Box::from_raw(ptr) });
		}
	}
}

impl<State: RcState> HybridRc<dyn Any, State> {
	/// Tries to downcast the `HybridRc<dyn Any, _>` to a concrete type.
	///
	/// # Errors
	/// If a downcast failed, the original `HybridRc` is returned in `Err`
	///
	/// # Example
	///
	/// ```
	/// use std::any::Any;
	/// use std::mem::drop;
	/// use hybrid_rc::Rc;
	///
	/// let value = 42i32;
	/// let concrete = Rc::new(value);
	/// let any: Rc<dyn Any> = Rc::into(concrete);
	///
	/// let any = any.downcast::<String>().unwrap_err();
	///
	/// assert_eq!(*any.downcast::<i32>().unwrap(), 42);
	/// ```
	#[inline]
	pub fn downcast<T: Any>(self) -> Result<HybridRc<T, State>, Self> {
		if (*self).is::<T>() {
			let ptr = self.ptr.cast::<RcBox<T>>();
			std::mem::forget(self);
			Ok(HybridRc::from_inner(ptr))
		} else {
			Err(self)
		}
	}
}

impl<T: ?Sized> Rc<T> {
	/// Creates a new shared reference (`Arc`) for the referenced value.
	///
	/// # Example
	/// ```
	/// use hybrid_rc::{Rc, Arc};
	/// # fn main() -> Result<(), Box<dyn std::any::Any + Send + 'static>> {
	///
	/// let local = Rc::new(42i32);
	/// let shared = Rc::to_shared(&local);
	///
	/// // `shared` can be safely transferred to another thread
	/// std::thread::spawn(move || assert_eq!(*shared, 42i32)).join()?;
	/// # Ok(())
	/// # }
	/// ```
	#[inline]
	pub fn to_shared(this: &Self) -> Arc<T> {
		this.meta().inc_strong_shared();
		Arc::from_inner(this.ptr)
	}
}

impl<T: ?Sized> Arc<T> {
	/// Creates a new local reference (`Rc`) for the referenced value.
	///
	/// Returns `None` if at least one `Rc` already exists on another thread.
	///
	/// # Example
	/// ```
	/// use hybrid_rc::{Rc, Arc};
	/// # fn main() -> Result<(), Box<dyn std::any::Any + Send + 'static>> {
	///
	/// let local = Rc::new(42i32);
	/// let shared = Rc::to_shared(&local);
	///
	/// // `shared` can be safely transferred to another thread
	/// std::thread::spawn(move || assert_eq!(*shared, 42i32)).join()?;
	/// # Ok(())
	/// # }
	/// ```
	#[must_use]
	#[inline]
	pub fn to_local(this: &Self) -> Option<Rc<T>> {
		let meta = this.meta();
		let current_thread = ThreadId::current_thread();
		let owner = match meta.owner.compare_exchange(
			None,
			Some(current_thread),
			Ordering::Acquire,
			Ordering::Relaxed,
		) {
			Ok(_) => None,
			Err(owner) => owner,
		};

		match owner {
			None => {
				meta.try_inc_strong_local()
					.expect("inconsistent reference count (shared == 0)");
				Some(Rc::from_inner(this.ptr))
			}
			Some(v) if v == current_thread => {
				meta.inc_strong_local();
				Some(Rc::from_inner(this.ptr))
			}
			Some(_) => None,
		}
	}
}

impl<T> Rc<T> {
	/// Creates a new `Rc<T>`, moving `data` into a reference counted allocation.
	///
	/// The shared value is initially owned by the calling thread, so for another thread to assume
	/// ownership [`to_shared()`] must be used and all `Rc`s for the value must be dropped.
	///
	/// # Example
	/// ```
	/// use hybrid_rc::Rc;
	///
	/// let rc = Rc::new(42i32);
	/// ```
	/// ```compile_fail
	/// # let rc = hybrid_rc::Rc::new(42i32);
	/// // Cannot be used in another thread without using rc.to_shared()
	/// std::thread::spawn(move || *rc).join(); // does not compile
	/// ```
	///
	/// [`to_shared()`]: Self::to_shared
	#[inline]
	pub fn new(data: T) -> Self {
		let inner = Box::new(RcBox::<T> {
			meta: RcMeta {
				owner: ThreadId::current_thread().into(),
				strong_local: Cell::new(1),
				strong_shared: 1.into(),
				weak: 1.into(),
			},
			data,
		});
		Self::from_inner(Box::leak(inner).into())
	}
}

impl<T> Arc<T> {
	/// Creates a new `Arc<T>`, moving `data` into a reference counted allocation.
	///
	/// Initially the shared value has no owner thread, so any thread may call [`to_local()`] to
	/// assume ownership.
	///
	/// Only useful for types that are `Sync + Send`, as otherwise the `Arc`
	/// can't be sent to other threads.
	///
	/// # Example
	/// ```
	/// use hybrid_rc::Arc;
	/// # fn main() -> Result<(), Box<dyn std::any::Any + Send + 'static>> {
	///
	/// let arc = Arc::new(42i32);
	///
	/// std::thread::spawn(move || assert!(*arc == 42)).join()?;
	/// # Ok(())
	/// # }
	/// ```
	///
	/// [`to_local()`]: Self::to_local
	#[inline]
	pub fn new(data: T) -> Self {
		let inner = Box::new(RcBox::<T> {
			meta: RcMeta {
				owner: None.into(),
				strong_local: Cell::new(0),
				strong_shared: 1.into(),
				weak: 1.into(),
			},
			data,
		});
		Self::from_inner(Box::leak(inner).into())
	}
}

impl<T: ?Sized> Clone for HybridRc<T, Local> {
	/// Creates another `Rc` for the same value.
	///
	/// # Example
	/// ```
	/// use hybrid_rc::Rc;
	///
	/// let first = Rc::new(42i32);
	/// let second = Rc::clone(&first);
	///
	/// assert_eq!(Rc::as_ptr(&first), Rc::as_ptr(&second));
	/// ```
	#[inline]
	fn clone(&self) -> Self {
		self.meta().inc_strong_local();
		Self::from_inner(self.ptr)
	}
}

impl<T: ?Sized> Clone for HybridRc<T, Shared> {
	/// Creates another `Arc` for the same value.
	///
	/// # Example
	/// ```
	/// use hybrid_rc::Arc;
	/// # fn main() -> Result<(), Box<dyn std::any::Any + Send + 'static>> {
	///
	/// let first = Arc::new(42i32);
	/// let second = Arc::clone(&first);
	///
	/// assert_eq!(Arc::as_ptr(&first), Arc::as_ptr(&second));
	///
	/// let value = std::thread::spawn(move || *second)
	///   .join()?;
	/// assert_eq!(*first, value);
	/// # Ok(())
	/// # }
	/// ```
	#[inline]
	fn clone(&self) -> Self {
		self.meta().inc_strong_shared();
		Self::from_inner(self.ptr)
	}
}

impl<T: ?Sized, State: RcState> Drop for HybridRc<T, State> {
	/// Drops the `HybridRc`.
	///
	/// This will decrement the appropriate reference count depending on `State`. If both strong
	/// reference counts reach zero then the only other references (if any) are [`Weak`]. In that
	/// case the inner value is dropped.
	#[inline]
	fn drop(&mut self) {
		let no_more_strong_refs = if State::SHARED {
			self.meta().dec_strong_shared()
		} else {
			self.meta().dec_strong_local()
		};

		if no_more_strong_refs {
			unsafe {
				self.drop_contents_and_maybe_box();
			}
		}
	}
}

// Dereferencing traits

impl<T: ?Sized, State: RcState> Deref for HybridRc<T, State> {
	type Target = T;

	#[inline]
	fn deref(&self) -> &T {
		self.data()
	}
}

impl<T: ?Sized, State: RcState> Borrow<T> for HybridRc<T, State> {
	#[inline]
	fn borrow(&self) -> &T {
		&**self
	}
}

impl<T: ?Sized, State: RcState> AsRef<T> for HybridRc<T, State> {
	#[inline]
	fn as_ref(&self) -> &T {
		&**self
	}
}

// Safety: T: Sync implies that dereferencing the Arc<T> on multiple threads is sound and T: Send
// implies that dropping T on another thread is sound. So T: Sync + Send gives all guarantees we
// need to make Arc Sync + Send.
unsafe impl<T: ?Sized + Sync + Send> Send for HybridRc<T, Shared> {}
unsafe impl<T: ?Sized + Sync + Send> Sync for HybridRc<T, Shared> {}

// Unwind safety: A HybridRc can only be UnwindSafe if the inner type is RefUnwindSafe.
impl<T: RefUnwindSafe + ?Sized, State: RcState> UnwindSafe for HybridRc<T, State> {}

// Unwind safety: An Arc is always RefUnwindSafe because of its use of atomics.
impl<T: RefUnwindSafe> RefUnwindSafe for HybridRc<T, Shared> {}

impl<T: Any + 'static, State: RcState> From<HybridRc<T, State>>
	for HybridRc<dyn Any + 'static, State>
{
	/// Upcasts a `HybridRc<T, State>` into an `HybridRc<dyn Any, State>`
	#[inline]
	fn from(src: HybridRc<T, State>) -> Self {
		let ptr = src.ptr.as_ptr() as *mut RcBox<dyn Any>;
		std::mem::forget(src);
		Self::from_inner(unsafe { NonNull::new_unchecked(ptr) })
	}
}

impl<T: ?Sized> From<Rc<T>> for HybridRc<T, Shared> {
	#[inline]
	/// Converts an `Rc<T>` into an `Arc<T>`.
	///
	/// See [`to_shared()`].
	///
	/// [`to_shared()`]: HybridRc::to_shared
	fn from(src: Rc<T>) -> Self {
		HybridRc::to_shared(&src)
	}
}

impl<T: ?Sized> TryFrom<Arc<T>> for HybridRc<T, Local> {
	type Error = Arc<T>;

	/// Tries to convert an `Arc<T>` into an `Rc<T>`.
	///
	/// See [`to_local()`].
	///
	/// [`to_local()`]: HybridRc::to_local
	#[inline]
	fn try_from(src: Arc<T>) -> Result<Self, Self::Error> {
		match HybridRc::to_local(&src) {
			Some(result) => Ok(result),
			None => Err(src),
		}
	}
}

// Propagate some useful traits implemented by the inner type

impl<T: Default> Default for HybridRc<T, Local> {
	/// Creates a new `Rc`, with the `Default` value for `T`.
	#[inline]
	fn default() -> Self {
		Self::new(Default::default())
	}
}

impl<T: Default> Default for HybridRc<T, Shared> {
	/// Creates a new `Arc`, with the `Default` value for `T`.
	#[inline]
	fn default() -> Self {
		Self::new(Default::default())
	}
}

impl<T: ?Sized + PartialEq, S1: RcState, S2: RcState> PartialEq<HybridRc<T, S2>>
	for HybridRc<T, S1>
{
	/// Equality for `HybridRc`s.
	///
	/// Two `HybridRc`s are equal if their inner values are equal, independent of if they are
	/// stored in the same allocation.
	#[inline]
	fn eq(&self, other: &HybridRc<T, S2>) -> bool {
		**self == **other
	}
}

impl<T: ?Sized + Eq, State: RcState> Eq for HybridRc<T, State> {}

impl<T: ?Sized + Hash, State: RcState> Hash for HybridRc<T, State> {
	#[inline]
	fn hash<H: Hasher>(&self, state: &mut H) {
		Self::data(self).hash(state);
	}
}

impl<T: ?Sized + fmt::Display, State: RcState> fmt::Display for HybridRc<T, State> {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		fmt::Display::fmt(&Self::data(self), f)
	}
}

impl<T: ?Sized + fmt::Debug, State: RcState> fmt::Debug for HybridRc<T, State> {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		fmt::Debug::fmt(&Self::data(self), f)
	}
}

// `HybridRc` can be formatted as a pointer.
impl<T: ?Sized, State: RcState> fmt::Pointer for HybridRc<T, State> {
	/// Formats the value using the given formatter.
	///
	/// If the `#` flag is used, the state (shared/local) is written after the address.
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		if f.alternate() {
			fmt::Pointer::fmt(&Self::as_ptr(self), f)?;
			f.write_str(if State::SHARED {
				" [shared]"
			} else {
				" [local]"
			})
		} else {
			fmt::Pointer::fmt(&Self::as_ptr(self), f)
		}
	}
}

/// `Weak<T>` represents a non-owning reference to a value managed by a [`HybridRc<T, _>`].
/// The value is accessed by calling [`upgrade()`] or [`upgrade_local()`] on `Weak`.
///
/// `Weak` references are typically used to prevent circular references that would keep
/// the shared value alive indefinitely.
///
/// The typical way to obtain a `Weak<T>` is to call [`HybridRc::downgrade()`].
///
/// [`upgrade()`]: Weak::upgrade
/// [`upgrade_local()`]: Weak::upgrade_local
#[must_use]
pub struct Weak<T: ?Sized> {
	ptr: NonNull<RcBox<T>>,
}

impl<T: ?Sized> Weak<T> {
	/// Accesses the metadata area of the shared allocation.
	///
	/// `None` for instances created through `Weak::new()`.
	#[inline]
	fn meta(&self) -> Option<&RcMeta> {
		if is_dangling(self.ptr.as_ptr()) {
			None
		} else {
			// Safety: as long as one Rc or Weak
			// for this item exists, the memory stays
			// allocated.
			Some(unsafe { &(*self.ptr.as_ptr()).meta })
		}
	}

	/// Returns a raw pointer to the value referenced by this `Weak<T>`.
	///
	/// The pointer is valid only if there are some strong references. It may be dangling,
	/// unaligned or even null otherwise.
	///
	/// # Example
	/// ```
	/// use hybrid_rc::Rc;
	///
	/// let strong = Rc::new(42i32);
	/// let weak = Rc::downgrade(&strong);
	/// {
	/// 	let pointer = weak.as_ptr();
	/// 	// As long as strong is not dropped, the pointer stays valid
	/// 	assert_eq!(42, unsafe { *pointer });
	/// }
	/// drop(strong);
	/// {
	/// 	// Calling weak.as_ptr() is still safe, but dereferencing it would lead
	/// 	// to undefined behaviour.
	/// 	let pointer = weak.as_ptr();
	/// 	// assert_eq!(42, unsafe { &*pointer }); // undefined behaviour
	/// }
	#[must_use]
	#[inline]
	pub fn as_ptr(&self) -> *const T {
		let ptr: *mut RcBox<T> = self.ptr.as_ptr();

		if is_dangling(ptr) {
			// If the pointer is dangling, we return the sentinel directly. This cannot be
			// a valid payload address, as the payload is at least as aligned as ArcInner (usize).
			ptr as *const T
		} else {
			// Safety: raw pointer manipulation like in sync::Weak, as the payload may have been
			// dropped at this point and to keep provenance.
			unsafe { ptr::addr_of_mut!((*ptr).data) }
		}
	}

	/// Attempts to upgrade the Weak pointer to an [`Rc`].
	///
	/// **Note:** Only one thread can have `Rc`s for a value at any point in time.
	/// See [`upgrade()`] to upgrade to an [`Arc`].
	///
	/// # Errors
	/// - [`ValueDropped`]: the referenced value has already been dropped.
	/// - [`WrongThread`]: another thread currently holds `Rc`s for the value.
	///
	/// # Example
	/// ```
	/// use hybrid_rc::{Rc, Weak, UpgradeError};
	/// # fn main() -> Result<(), UpgradeError> {
	/// let strong = Rc::new(42i32);
	/// let weak = Rc::downgrade(&strong);
	///
	/// {
	/// 	let strong2 = weak.upgrade_local()?;
	/// 	assert_eq!(Rc::as_ptr(&strong), Rc::as_ptr(&strong2));
	/// }
	///
	/// std::mem::drop(strong);
	///
	/// let error = Weak::upgrade_local(&weak).unwrap_err();
	/// assert_eq!(error, UpgradeError::ValueDropped);
	/// # Ok(())
	/// # }
	/// ```
	///
	/// [`upgrade()`]: Weak::upgrade
	/// [`ValueDropped`]: UpgradeError::ValueDropped
	/// [`WrongThread`]: UpgradeError::WrongThread
	#[inline]
	pub fn upgrade_local(&self) -> Result<Rc<T>, UpgradeError> {
		let meta = self.meta().ok_or(UpgradeError::ValueDropped)?;
		let current_thread = ThreadId::current_thread();

		let owner = match meta.owner.compare_exchange(
			None,
			Some(current_thread),
			Ordering::Acquire,
			Ordering::Relaxed,
		) {
			Ok(_) => Some(current_thread),
			Err(owner) => owner,
		};

		if owner == Some(current_thread) {
			if meta.try_inc_strong_local().is_ok() {
				Ok(HybridRc::<T, Local>::from_inner(self.ptr))
			} else {
				meta.owner.store(None, Ordering::Relaxed);
				Err(UpgradeError::ValueDropped)
			}
		} else {
			Err(UpgradeError::WrongThread)
		}
	}

	/// Attempts to upgrade the Weak pointer to an [`Arc`].
	///
	/// Also see [`upgrade_local()`] to upgrade to an [`Rc`].
	///
	/// # Errors
	/// - [`ValueDropped`]: the referenced value has already been dropped.
	///
	/// [`upgrade_local()`]: Weak::upgrade_local
	/// [`ValueDropped`]: UpgradeError::ValueDropped
	#[inline]
	pub fn upgrade(&self) -> Result<Arc<T>, UpgradeError> {
		let meta = self.meta().ok_or(UpgradeError::ValueDropped)?;
		meta.try_inc_strong_shared()
			.map_err(|_| UpgradeError::ValueDropped)?;
		Ok(HybridRc::<T, Shared>::from_inner(self.ptr))
	}
}

impl<T> Weak<T> {
	/// Constructs a dummy `Weak<T>`, without allocating any memory.
	///
	/// Trying to upgrade the result will always result in a [`ValueDropped`] error.
	///
	/// [`ValueDropped`]: UpgradeError::ValueDropped
	pub fn new() -> Weak<T> {
		Weak { ptr: senitel() }
	}
}

impl<T: ?Sized> fmt::Debug for Weak<T> {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		write!(f, "(Weak)")
	}
}

impl<T: ?Sized> fmt::Pointer for Weak<T> {
	/// Formats the value using the given formatter.
	///
	/// If the `#` flag is used, the state (weak) is written after the address.
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		if f.alternate() {
			fmt::Pointer::fmt(&Self::as_ptr(self), f)?;
			f.write_str(" [weak]")
		} else {
			fmt::Pointer::fmt(&Self::as_ptr(self), f)
		}
	}
}

impl<T> Default for Weak<T> {
	/// Constructs a dummy `Weak<T>`, without allocating any memory.
	///
	/// See [`Weak<T>::new()`].
	#[inline]
	fn default() -> Self {
		Self::new()
	}
}

impl<T: ?Sized> Clone for Weak<T> {
	/// Creates another `Weak` reference for the same value.
	///
	/// # Example
	/// ```
	/// use hybrid_rc::{Rc, Weak};
	///
	/// let strong = Rc::new(42i32);
	/// let weak = Rc::downgrade(&strong);
	/// let weak2 = Weak::clone(&weak);
	///
	/// assert_eq!(weak.as_ptr(), weak2.as_ptr());
	/// ```
	#[inline]
	fn clone(&self) -> Self {
		if let Some(meta) = self.meta() {
			// We can ignore the lock in Weak::clone() as the counter is only locked by HybridRc
			// when there are no Weak instances/ (meta.weak == 1).
			meta.inc_weak_nolock();
		}
		Self { ptr: self.ptr }
	}
}

impl<T: ?Sized> Drop for Weak<T> {
	/// Drops the `Weak` reference.
	///
	/// Once all `HybridRc` and `Weak` references to a shared value are dropped, the shared
	/// allocation is fully released.
	#[inline]
	fn drop(&mut self) {
		if let Some(meta) = self.meta() {
			let last_reference = meta.dec_weak();
			if last_reference {
				unsafe {
					// Safety: ManuallyDrop doesn't change the memory layout
					let ptr: *mut RcBox<mem::ManuallyDrop<T>> = mem::transmute(self.ptr.as_mut());
					// Safety: only called if this was the last (weak) reference
					// so only called once on a valid pointer that was once a leaked box.
					mem::drop(Box::from_raw(ptr));
				}
			}
		}
	}
}

#[cfg(test)]
mod tests {
	use super::*;
	use std::thread;

	thread_local! {
		static DROP_COUNTER: Cell<usize> = Cell::new(0);
	}

	#[derive(Debug)]
	struct Test {}
	impl Drop for Test {
		fn drop(&mut self) {
			DROP_COUNTER.with(|x| x.set(x.get() + 1));
		}
	}

	#[test]
	fn test_local() {
		DROP_COUNTER.with(|x| x.set(0));
		let a = Rc::new(Test {});
		let b = a.clone();
		mem::drop(a);
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 0);
		mem::drop(b);
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 1);
	}

	#[test]
	fn test_shared() {
		DROP_COUNTER.with(|x| x.set(0));
		let a = Arc::new(Test {});
		let b = a.clone();
		mem::drop(a);
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 0);
		mem::drop(b);
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 1);
	}

	#[test]
	fn test_shared_to_local() {
		DROP_COUNTER.with(|x| x.set(0));
		let a = Arc::new(Test {});
		let b = Arc::to_local(&a).unwrap();
		let _ = Arc::to_local(&a).unwrap();
		mem::drop(a);
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 0);
		mem::drop(b);
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 1);
	}

	#[test]
	fn test_shared_to_local_on_wrong_thread() {
		DROP_COUNTER.with(|x| x.set(0));
		let a = Arc::new(Test {});
		let b = Arc::to_local(&a).unwrap();
		assert!(thread::spawn(move || Arc::to_local(&a).is_none())
			.join()
			.unwrap());
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 0);
		mem::drop(b);
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 1);
	}

	#[test]
	fn test_local_to_shared() {
		DROP_COUNTER.with(|x| x.set(0));
		let a = Rc::new(Test {});
		let b = Rc::to_shared(&a);
		mem::drop(a);
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 0);
		mem::drop(b);
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 1);
	}

	#[test]
	fn test_dangling_weak() {
		DROP_COUNTER.with(|x| x.set(0));
		let w = Weak::<Test>::new();
		assert!(w.upgrade_local().is_err());
		assert!(w.upgrade().is_err());
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 0);
	}

	#[test]
	fn test_weak() {
		DROP_COUNTER.with(|x| x.set(0));
		let a = Rc::new(Test {});
		let b = Rc::to_shared(&a);
		mem::drop(a);
		let w = Arc::downgrade(&b);
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 0);
		assert!(w.upgrade_local().is_ok());
		mem::drop(b);
		assert_eq!(DROP_COUNTER.with(|x| x.get()), 1);
		assert!(w.upgrade_local().is_err());
	}
}