tokio/util/
sharded_list.rs

1use std::ptr::NonNull;
2use std::sync::atomic::Ordering;
3
4use crate::loom::sync::{Mutex, MutexGuard};
5use crate::util::metric_atomics::{MetricAtomicU64, MetricAtomicUsize};
6
7use super::linked_list::{Link, LinkedList};
8
9/// An intrusive linked list supporting highly concurrent updates.
10///
11/// It currently relies on `LinkedList`, so it is the caller's
12/// responsibility to ensure the list is empty before dropping it.
13///
14/// Note: Due to its inner sharded design, the order of nodes cannot be guaranteed.
15pub(crate) struct ShardedList<L: ShardedListItem> {
16    lists: Box<[Mutex<LinkedList<L>>]>,
17    added: MetricAtomicU64,
18    count: MetricAtomicUsize,
19    shard_mask: usize,
20}
21
22/// Determines which linked list an item should be stored in.
23///
24/// # Safety
25///
26/// Implementations must guarantee that the id of an item does not change from
27/// call to call.
28pub(crate) unsafe trait ShardedListItem: Link {
29    /// # Safety
30    ///
31    /// The provided pointer must point at a valid list item.
32    unsafe fn get_shard_id(target: NonNull<Self::Target>) -> usize;
33}
34
35/// Used to get the lock of shard.
36pub(crate) struct ShardGuard<'a, L: Link> {
37    lock: MutexGuard<'a, LinkedList<L>>,
38    added: &'a MetricAtomicU64,
39    count: &'a MetricAtomicUsize,
40    id: usize,
41}
42
43impl<L: ShardedListItem> ShardedList<L> {
44    /// Creates a new and empty sharded linked list with the specified size.
45    pub(crate) fn new(sharded_size: usize) -> Self {
46        assert!(sharded_size.is_power_of_two());
47
48        let shard_mask = sharded_size - 1;
49        let lists = std::iter::repeat_with(|| Mutex::new(LinkedList::new()));
50        Self {
51            lists: lists.take(sharded_size).collect(),
52            added: MetricAtomicU64::new(0),
53            count: MetricAtomicUsize::new(0),
54            shard_mask,
55        }
56    }
57
58    /// Removes the last element from a list specified by `shard_id` and returns it, or None if it is
59    /// empty.
60    pub(crate) fn pop_back(&self, shard_id: usize) -> Option<L::Handle> {
61        let mut lock = self.shard_inner(shard_id);
62        let node = lock.pop_back();
63        if node.is_some() {
64            self.count.decrement();
65        }
66        node
67    }
68
69    /// Removes the specified node from the list.
70    ///
71    /// # Safety
72    ///
73    /// The caller **must** ensure that exactly one of the following is true:
74    /// - `node` is currently contained by `self`,
75    /// - `node` is not contained by any list,
76    /// - `node` is currently contained by some other `GuardedLinkedList`.
77    pub(crate) unsafe fn remove(&self, node: NonNull<L::Target>) -> Option<L::Handle> {
78        let id = unsafe { L::get_shard_id(node) };
79        let mut lock = self.shard_inner(id);
80        // SAFETY: Since the shard id cannot change, it's not possible for this node
81        // to be in any other list of the same sharded list.
82        let node = unsafe { lock.remove(node) };
83        if node.is_some() {
84            self.count.decrement();
85        }
86        node
87    }
88
89    /// Gets the lock of `ShardedList`, makes us have the write permission.
90    pub(crate) fn lock_shard(&self, val: &L::Handle) -> ShardGuard<'_, L> {
91        let id = unsafe { L::get_shard_id(L::as_raw(val)) };
92        ShardGuard {
93            lock: self.shard_inner(id),
94            added: &self.added,
95            count: &self.count,
96            id,
97        }
98    }
99
100    /// Gets the count of elements in this list.
101    pub(crate) fn len(&self) -> usize {
102        self.count.load(Ordering::Relaxed)
103    }
104
105    cfg_unstable_metrics! {
106        cfg_64bit_metrics! {
107            /// Gets the total number of elements added to this list.
108            pub(crate) fn added(&self) -> u64 {
109                self.added.load(Ordering::Relaxed)
110            }
111        }
112    }
113
114    /// Returns whether the linked list does not contain any node.
115    pub(crate) fn is_empty(&self) -> bool {
116        self.len() == 0
117    }
118
119    /// Gets the shard size of this `SharedList`.
120    ///
121    /// Used to help us to decide the parameter `shard_id` of the `pop_back` method.
122    pub(crate) fn shard_size(&self) -> usize {
123        self.shard_mask + 1
124    }
125
126    #[inline]
127    fn shard_inner(&self, id: usize) -> MutexGuard<'_, LinkedList<L>> {
128        // Safety: This modulo operation ensures that the index is not out of bounds.
129        unsafe { self.lists.get_unchecked(id & self.shard_mask).lock() }
130    }
131}
132
133impl<'a, L: ShardedListItem> ShardGuard<'a, L> {
134    /// Push a value to this shard.
135    pub(crate) fn push(mut self, val: L::Handle) {
136        let id = unsafe { L::get_shard_id(L::as_raw(&val)) };
137        assert_eq!(id, self.id);
138        self.lock.push_front(val);
139        self.added.add(1, Ordering::Relaxed);
140        self.count.increment();
141    }
142}
143
144cfg_taskdump! {
145    impl<L: ShardedListItem> ShardedList<L> {
146        pub(crate) fn for_each<F>(&self, mut f: F)
147        where
148            F: FnMut(&L::Handle),
149        {
150            let mut guards = Vec::with_capacity(self.lists.len());
151            for list in self.lists.iter() {
152                guards.push(list.lock());
153            }
154            for g in &mut guards {
155                g.for_each(&mut f);
156            }
157        }
158    }
159}
tokio/util/sharded_list.rs

tokio/util/
sharded_list.rs
