Use std::sync::RwLock instead of tokio::sync::RwLock (by @aarroyoc)
This commit is contained in:
@@ -10,7 +10,6 @@ use crate::read::*;
|
|||||||
|
|
||||||
use crate::parser::dashu::{Integer, Rational};
|
use crate::parser::dashu::{Integer, Rational};
|
||||||
use ordered_float::OrderedFloat;
|
use ordered_float::OrderedFloat;
|
||||||
use tokio::sync::RwLock;
|
|
||||||
|
|
||||||
use std::alloc;
|
use std::alloc;
|
||||||
use std::cell::UnsafeCell;
|
use std::cell::UnsafeCell;
|
||||||
@@ -20,6 +19,7 @@ use std::mem;
|
|||||||
use std::net::TcpListener;
|
use std::net::TcpListener;
|
||||||
use std::ops::{Deref, DerefMut};
|
use std::ops::{Deref, DerefMut};
|
||||||
use std::ptr;
|
use std::ptr;
|
||||||
|
use std::sync::RwLock;
|
||||||
|
|
||||||
#[macro_export]
|
#[macro_export]
|
||||||
macro_rules! arena_alloc {
|
macro_rules! arena_alloc {
|
||||||
@@ -90,7 +90,8 @@ pub fn lookup_float(
|
|||||||
offset: F64Offset,
|
offset: F64Offset,
|
||||||
) -> RcuRef<RawBlock<F64Table>, UnsafeCell<OrderedFloat<f64>>> {
|
) -> RcuRef<RawBlock<F64Table>, UnsafeCell<OrderedFloat<f64>>> {
|
||||||
let f64table = global_f64table()
|
let f64table = global_f64table()
|
||||||
.blocking_read()
|
.read()
|
||||||
|
.unwrap()
|
||||||
.upgrade()
|
.upgrade()
|
||||||
.expect("We should only be looking up floats while there is a float table");
|
.expect("We should only be looking up floats while there is a float table");
|
||||||
|
|
||||||
@@ -108,12 +109,12 @@ pub fn lookup_float(
|
|||||||
impl F64Table {
|
impl F64Table {
|
||||||
#[inline]
|
#[inline]
|
||||||
pub fn new() -> Arc<Self> {
|
pub fn new() -> Arc<Self> {
|
||||||
let upgraded = global_f64table().blocking_read().upgrade();
|
let upgraded = global_f64table().read().unwrap().upgrade();
|
||||||
// don't inline upgraded, otherwise temporary will be dropped too late in case of None
|
// don't inline upgraded, otherwise temporary will be dropped too late in case of None
|
||||||
if let Some(atom_table) = upgraded {
|
if let Some(atom_table) = upgraded {
|
||||||
atom_table
|
atom_table
|
||||||
} else {
|
} else {
|
||||||
let mut guard = global_f64table().blocking_write();
|
let mut guard = global_f64table().write().unwrap();
|
||||||
// try to upgrade again in case we lost the race on the write lock
|
// try to upgrade again in case we lost the race on the write lock
|
||||||
if let Some(atom_table) = guard.upgrade() {
|
if let Some(atom_table) = guard.upgrade() {
|
||||||
atom_table
|
atom_table
|
||||||
|
|||||||
@@ -12,12 +12,12 @@ use std::slice;
|
|||||||
use std::str;
|
use std::str;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::sync::Mutex;
|
use std::sync::Mutex;
|
||||||
|
use std::sync::RwLock;
|
||||||
use std::sync::Weak;
|
use std::sync::Weak;
|
||||||
|
|
||||||
use indexmap::IndexSet;
|
use indexmap::IndexSet;
|
||||||
|
|
||||||
use modular_bitfield::prelude::*;
|
use modular_bitfield::prelude::*;
|
||||||
use tokio::sync::RwLock;
|
|
||||||
|
|
||||||
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
|
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
|
||||||
pub struct Atom {
|
pub struct Atom {
|
||||||
@@ -74,7 +74,7 @@ fn global_atom_table() -> &'static RwLock<Weak<AtomTable>> {
|
|||||||
|
|
||||||
#[inline(always)]
|
#[inline(always)]
|
||||||
fn arc_atom_table() -> Option<Arc<AtomTable>> {
|
fn arc_atom_table() -> Option<Arc<AtomTable>> {
|
||||||
global_atom_table().blocking_read().upgrade()
|
global_atom_table().read().unwrap().upgrade()
|
||||||
}
|
}
|
||||||
|
|
||||||
impl RawBlockTraits for AtomTable {
|
impl RawBlockTraits for AtomTable {
|
||||||
@@ -310,12 +310,12 @@ impl InnerAtomTable {
|
|||||||
impl AtomTable {
|
impl AtomTable {
|
||||||
#[inline]
|
#[inline]
|
||||||
pub fn new() -> Arc<Self> {
|
pub fn new() -> Arc<Self> {
|
||||||
let upgraded = global_atom_table().blocking_read().upgrade();
|
let upgraded = global_atom_table().read().unwrap().upgrade();
|
||||||
// don't inline upgraded, otherwise temporary will be dropped too late in case of None
|
// don't inline upgraded, otherwise temporary will be dropped too late in case of None
|
||||||
if let Some(atom_table) = upgraded {
|
if let Some(atom_table) = upgraded {
|
||||||
atom_table
|
atom_table
|
||||||
} else {
|
} else {
|
||||||
let mut guard = global_atom_table().blocking_write();
|
let mut guard = global_atom_table().write().unwrap();
|
||||||
// try to upgrade again in case we lost the race on the write lock
|
// try to upgrade again in case we lost the race on the write lock
|
||||||
if let Some(atom_table) = guard.upgrade() {
|
if let Some(atom_table) = guard.upgrade() {
|
||||||
atom_table
|
atom_table
|
||||||
|
|||||||
11
src/rcu.rs
11
src/rcu.rs
@@ -6,19 +6,17 @@ use std::{
|
|||||||
ptr::NonNull,
|
ptr::NonNull,
|
||||||
sync::{
|
sync::{
|
||||||
atomic::{AtomicPtr, AtomicU8},
|
atomic::{AtomicPtr, AtomicU8},
|
||||||
Arc, Weak,
|
Arc, Weak, RwLock
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
|
|
||||||
use tokio::sync::RwLock;
|
|
||||||
|
|
||||||
// the epoch counters of all threads that have ever accessed an Rcu
|
// the epoch counters of all threads that have ever accessed an Rcu
|
||||||
// threads that have finished will have a dangling Weak reference and can be cleand up
|
// threads that have finished will have a dangling Weak reference and can be cleand up
|
||||||
// having this be shared between all Rcu's is a tradeof,
|
// having this be shared between all Rcu's is a tradeof,
|
||||||
// writes will be slower as more epoch counters need to be waited for
|
// writes will be slower as more epoch counters need to be waited for
|
||||||
// reads should be faster as a thread only needs to register itself once on the first read
|
// reads should be faster as a thread only needs to register itself once on the first read
|
||||||
//
|
//
|
||||||
static EPOCH_COUNTERS: RwLock<Vec<Weak<AtomicU8>>> = RwLock::const_new(Vec::new());
|
static EPOCH_COUNTERS: RwLock<Vec<Weak<AtomicU8>>> = RwLock::new(Vec::new());
|
||||||
|
|
||||||
thread_local! {
|
thread_local! {
|
||||||
// odd value means the current thread is about to access the active_epoch of an Rcu
|
// odd value means the current thread is about to access the active_epoch of an Rcu
|
||||||
@@ -53,7 +51,8 @@ impl<T> Rcu<T> {
|
|||||||
let epoch_counter = Arc::new(AtomicU8::new(0));
|
let epoch_counter = Arc::new(AtomicU8::new(0));
|
||||||
// register the current threads epoch counter on init
|
// register the current threads epoch counter on init
|
||||||
EPOCH_COUNTERS
|
EPOCH_COUNTERS
|
||||||
.blocking_write()
|
.write()
|
||||||
|
.unwrap()
|
||||||
.push(Arc::downgrade(&epoch_counter));
|
.push(Arc::downgrade(&epoch_counter));
|
||||||
epoch_counter
|
epoch_counter
|
||||||
});
|
});
|
||||||
@@ -113,7 +112,7 @@ impl<T> Rcu<T> {
|
|||||||
// - the Rcu itself holds one strong count
|
// - the Rcu itself holds one strong count
|
||||||
let arc = unsafe { ManuallyDrop::new(Arc::from_raw(arc_ptr)) };
|
let arc = unsafe { ManuallyDrop::new(Arc::from_raw(arc_ptr)) };
|
||||||
|
|
||||||
let epochs = EPOCH_COUNTERS.blocking_read().clone();
|
let epochs = EPOCH_COUNTERS.read().unwrap().clone();
|
||||||
let mut epochs = epochs
|
let mut epochs = epochs
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.flat_map(|elem| {
|
.flat_map(|elem| {
|
||||||
|
|||||||
Reference in New Issue
Block a user