Struct concurrent_map::ConcurrentMap

pub struct ConcurrentMap<K, V, const FANOUT: usize = 64, const LOCAL_GC_BUFFER_SIZE: usize = 128>where
    K: 'static + Clone + Minimum + Ord + Send + Sync,
    V: 'static + Clone + Send + Sync,{ /* private fields */ }

A lock-free B+ tree.

Note that this structure is Send but NOT Sync, despite being a lock-free tree. This is because the inner reclamation system, provided by the ebr crate completely avoids atomic operations in its hot path for efficiency. If you want to share ConcurrentMap between threads, simply clone it, and this will set up a new efficient thread-local memory reclamation state.

If you want to use a custom key type, you must implement the Minimum trait, allowing the left-most side of the tree to be created before inserting any data. If you wish to perform scans in reverse lexicographical order, you may instead implement Maximum for your key type and use std::cmp::Reverse.

The FANOUT const generic must be greater than 3. This const generic controls how large the fixed-size array for either child pointers (for index nodes) or values (for leaf nodes) will be. A higher value may make reads and scans faster, but writes will be slower because each modification performs a read-copy-update of the full node. In some cases, it may be preferable to wrap complex values in an Arc to avoid higher copy costs.

The LOCAL_GC_BUFFER_SIZE const generic must be greater than 0. This controls the epoch-based reclamation granularity. Garbage is placed into fixed-size arrays, and garbage collection only happens after this array fills up and a final timestamp is assigned to it. Lower values will cause replaced values to be dropped more quickly, but the efficiency will be lower. Values that are extremely high may cause undesirable memory usage because it will take more time to fill up each thread-local garbage segment.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

// insert and remove atomically returns the last value, if it was set,
// similarly to a BTreeMap
assert_eq!(map.insert(1, 10), None);
assert_eq!(map.insert(1, 11), Some(10));
assert_eq!(map.remove(&1), Some(11));

// get also functions similarly to BTreeMap, except it
// returns a cloned version of the value rather than a
// reference to it, so that no locks need to be maintained.
// For this reason, it can be a good idea to use types that
// are cheap to clone for values, which can be easily handled
// with `Arc` etc...
assert_eq!(map.insert(1, 12), None);
assert_eq!(map.get(&1), Some(12));

// compare and swap from value 12 to value 20
map.cas(1, Some(&12_usize), Some(20)).unwrap();

assert_eq!(map.get(&1).unwrap(), 20);

// there are a lot of methods that are not covered
// here - check out the docs!

Implementations

impl<K, V, const FANOUT: usize, const LOCAL_GC_BUFFER_SIZE: usize> ConcurrentMap<K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>where K: 'static + Clone + Minimum + Ord + Send + Sync, V: 'static + Clone + Send + Sync,

pub fn new() -> Self

Creates a new empty ConcurrentMap<K, V, ...>.

Examples

use concurrent_map::ConcurrentMap;

let cm: ConcurrentMap<bool, usize> = ConcurrentMap::new();

pub fn get<Q>(&self, key: &Q) -> Option<V>where K: Borrow<Q>, Q: Ord + ?Sized,

Atomically get a value out of the map that is associated with this key.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

map.insert(1, 1);

let actual = map.get(&0);
let expected = None;
assert_eq!(expected, actual);

let actual = map.get(&1);
let expected = Some(1);
assert_eq!(expected, actual);

pub fn contains_key<Q>(&self, key: &Q) -> boolwhere K: Borrow<Q>, Q: Ord + ?Sized,

Returns true if the ConcurrentMap contains the specified key.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

map.insert(1, 1);

assert!(map.contains_key(&1));
assert!(!map.contains_key(&2));

pub fn get_lt<Q>(&self, key: &Q) -> Option<(K, V)>where K: Borrow<Q>, Q: ?Sized + Ord + PartialEq,

Atomically get a key and value out of the map that is associated with the key that is lexicographically less than the provided key.

This will always return None if the key passed to get_lt == K::MIN.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

map.insert(1, 1);

let actual = map.get_lt(&0);
let expected = None;
assert_eq!(expected, actual);

let actual = map.get_lt(&1);
let expected = None;
assert_eq!(expected, actual);

let actual = map.get_lt(&2);
let expected = Some((1, 1));
assert_eq!(expected, actual);

pub fn get_lte<Q>(&self, key: &Q) -> Option<(K, V)>where K: Borrow<Q>, Q: ?Sized + Ord + PartialEq,

Atomically get a key and value out of the map that is associated with the key that is lexicographically less than or equal to the provided key.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

map.insert(1, 1);

let actual = map.get_lte(&0);
let expected = None;
assert_eq!(expected, actual);

let actual = map.get_lte(&1);
let expected = Some((1, 1));
assert_eq!(expected, actual);

let actual = map.get_lte(&2);
let expected = Some((1, 1));
assert_eq!(expected, actual);

pub fn get_gt<Q>(&self, key: &Q) -> Option<(K, V)>where K: Borrow<Q>, Q: ?Sized + Ord + PartialEq,

Atomically get a key and value out of the map that is associated with the key that is lexicographically greater than the provided key.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

map.insert(1, 1);

let actual = map.get_gt(&0);
let expected = Some((1, 1));
assert_eq!(expected, actual);

let actual = map.get_gt(&1);
let expected = None;
assert_eq!(expected, actual);

let actual = map.get_gt(&2);
let expected = None;
assert_eq!(expected, actual);

pub fn get_gte<Q>(&self, key: &Q) -> Option<(K, V)>where K: Borrow<Q>, Q: ?Sized + Ord + PartialEq,

Atomically get a key and value out of the map that is associated with the key that is lexicographically greater than or equal to the provided key.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

map.insert(1, 1);

let actual = map.get_gte(&0);
let expected = Some((1, 1));
assert_eq!(expected, actual);

let actual = map.get_gte(&1);
let expected = Some((1, 1));
assert_eq!(expected, actual);

let actual = map.get_gte(&2);
let expected = None;
assert_eq!(expected, actual);

pub fn first(&self) -> Option<(K, V)>

Get the minimum item stored in this structure.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

map.insert(1, 1);
map.insert(2, 2);
map.insert(3, 3);

let actual = map.first();
let expected = Some((1, 1));
assert_eq!(actual, expected);

pub fn pop_first(&self) -> Option<(K, V)>where V: PartialEq,

Atomically remove the minimum item stored in this structure.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

map.insert(1, 1);
map.insert(2, 2);
map.insert(3, 3);

let actual = map.pop_first();
let expected = Some((1, 1));
assert_eq!(actual, expected);

assert_eq!(map.get(&1), None);

Examples found in repository

examples/mpmc.rs (line 21)

fn consumer(cm: ConcurrentMap<usize, usize>, n: usize) {
    let mut popped = 0;
    while popped < n {
        if let Some((k, v)) = cm.pop_first() {
            assert_eq!(k, v);
            popped += 1;
        }
    }
}

pub fn pop_first_in_range<Q, R>(&self, range: R) -> Option<(K, V)>where R: RangeBounds<Q> + Clone, K: Borrow<Q>, V: PartialEq, Q: ?Sized + Ord + PartialEq,

Pops the first kv pair in the provided range, or returns None if nothing exists within that range.

Panics

This will panic if the provided range’s end_bound() == Bound::Excluded(K::MIN).

Examples

use concurrent_map::ConcurrentMap;

let data = vec![
    ("key 1", 1),
    ("key 2", 2),
    ("key 3", 3)
];

let map: ConcurrentMap<&'static str, usize> = data.iter().copied().collect();

let r1 = map.pop_first_in_range("key 1"..="key 3");
assert_eq!(Some(("key 1", 1_usize)), r1);

let r2 = map.pop_first_in_range("key 1".."key 3");
assert_eq!(Some(("key 2", 2_usize)), r2);

let r3: Vec<_> = map.range("key 4"..).collect();
assert!(r3.is_empty());

let r4 = map.pop_first_in_range("key 2"..="key 3");
assert_eq!(Some(("key 3", 3_usize)), r4);

pub fn last(&self) -> Option<(K, V)>

Get the maximum item stored in this structure.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

map.insert(1, 1);
map.insert(2, 2);
map.insert(3, 3);

let actual = map.last();
let expected = Some((3, 3));
assert_eq!(actual, expected);

pub fn pop_last(&self) -> Option<(K, V)>where V: PartialEq,

Atomically remove the maximum item stored in this structure.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

map.insert(1, 1);
map.insert(2, 2);
map.insert(3, 3);

let actual = map.pop_last();
let expected = Some((3, 3));
assert_eq!(actual, expected);

assert_eq!(map.get(&3), None);

pub fn pop_last_in_range<Q, R>(&self, range: R) -> Option<(K, V)>where R: RangeBounds<Q> + Clone, K: Borrow<Q>, V: PartialEq, Q: ?Sized + Ord + PartialEq,

Pops the last kv pair in the provided range, or returns None if nothing exists within that range.

Panics

This will panic if the provided range’s end_bound() == Bound::Excluded(K::MIN).

Examples

use concurrent_map::ConcurrentMap;

let data = vec![
    ("key 1", 1),
    ("key 2", 2),
    ("key 3", 3)
];

let map: ConcurrentMap<&'static str, usize> = data.iter().copied().collect();

let r1 = map.pop_last_in_range("key 1"..="key 3");
assert_eq!(Some(("key 3", 3_usize)), r1);

let r2 = map.pop_last_in_range("key 1".."key 3");
assert_eq!(Some(("key 2", 2_usize)), r2);

let r3 = map.pop_last_in_range("key 4"..);
assert!(r3.is_none());

let r4 = map.pop_last_in_range("key 2"..="key 3");
assert!(r4.is_none());

let r5 = map.pop_last_in_range("key 0"..="key 3");
assert_eq!(Some(("key 1", 1_usize)), r5);

let r6 = map.pop_last_in_range("key 0"..="key 3");
assert!(r6.is_none());

pub fn insert(&self, key: K, value: V) -> Option<V>

Atomically insert a key-value pair into the map, returning the previous value associated with this key if one existed.

This method has an optimization that skips lock-free RCU when the internal Arc has a strong count of 1, significantly increasing insertion throughput when used from a single thread.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

assert_eq!(map.insert(1, 1), None);
assert_eq!(map.insert(1, 1), Some(1));

Examples found in repository

examples/mpmc.rs (line 14)

fn producer(cm: ConcurrentMap<usize, usize>, min: usize, max: usize) {
    for i in min..max {
        cm.insert(i, i);
    }
}

pub fn remove<Q>(&self, key: &Q) -> Option<V>where K: Borrow<Q>, Q: Ord + ?Sized,

Atomically remove the value associated with this key from the map, returning the previous value if one existed.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

assert_eq!(map.remove(&1), None);
assert_eq!(map.insert(1, 1), None);
assert_eq!(map.remove(&1), Some(1));

pub fn cas<VRef>( &self, key: K, old: Option<&VRef>, new: Option<V> ) -> Result<Option<V>, CasFailure<V>>where V: Borrow<VRef>, VRef: PartialEq + ?Sized,

Atomically compare and swap the value associated with this key from the old value to the new one. An old value of None means “only create this value if it does not already exist”. A new value of None means “delete this value, if it matches the provided old value”. If successful, returns the old value if it existed. If unsuccessful, returns both the proposed new value that failed to be installed as well as the current actual value in a CasFailure struct.

Examples

let map = concurrent_map::ConcurrentMap::<usize, usize>::default();

// key 1 does not yet exist
assert_eq!(map.get(&1), None);

// uniquely create value 10
map.cas(1, None, Some(10)).unwrap();

assert_eq!(map.get(&1).unwrap(), 10);

// compare and swap from value 10 to value 20
map.cas(1, Some(&10_usize), Some(20)).unwrap();

assert_eq!(map.get(&1).unwrap(), 20);

// if we guess the wrong current value, a CasFailure is returned
// which will tell us what the actual current value is (which we
// failed to provide) and it will give us back our proposed new
// value.
let cas_result = map.cas(1, Some(&999999_usize), Some(30));

let expected_cas_failure = Err(concurrent_map::CasFailure {
    actual: Some(20),
    returned_new_value: Some(30),
});

assert_eq!(cas_result, expected_cas_failure);

// conditionally delete
map.cas(1, Some(&20_usize), None).unwrap();

assert_eq!(map.get(&1), None);

pub fn len(&self) -> usize

A lagging, eventually-consistent length count. This is NOT atomically updated with [insert] / [remove] / [cas], but is updated after those operations complete their atomic modifications to the shared map.

pub fn is_empty(&self) -> bool

A lagging, eventually-consistent check for emptiness, based on the correspondingly non-atomic len method.

pub fn iter(&self) -> Iter<'_, K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>

Iterate over the map.

This is not an atomic snapshot, and it caches B+tree leaf nodes as it iterates through them to achieve high throughput. As a result, the following behaviors are possible:

• may (or may not!) return values that were concurrently added to the map after the iterator was created
• may (or may not!) return items that were concurrently deleted from the map after the iterator was created
• If a key’s value is changed from one value to another one after this iterator is created, this iterator might return the old or the new value.

But, you can be certain that any key that existed prior to the creation of this iterator, and was not changed during iteration, will be observed as expected.

Examples

use concurrent_map::ConcurrentMap;

let data = vec![
    ("key 1", 1),
    ("key 2", 2),
    ("key 3", 3)
];

let map: ConcurrentMap<&'static str, usize> = data.iter().copied().collect();

let r: Vec<_> = map.iter().collect();

assert_eq!(&data, &r);

pub fn range<Q, R>( &self, range: R ) -> Iter<'_, K, V, FANOUT, LOCAL_GC_BUFFER_SIZE, R, Q> where R: RangeBounds<Q>, K: Borrow<Q>, Q: ?Sized + Ord + PartialEq,

Iterate over a range of the map.

This is not an atomic snapshot, and it caches B+tree leaf nodes as it iterates through them to achieve high throughput. As a result, the following behaviors are possible:

• may (or may not!) return values that were concurrently added to the map after the iterator was created
• may (or may not!) return items that were concurrently deleted from the map after the iterator was created
• If a key’s value is changed from one value to another one after this iterator is created, this iterator might return the old or the new value.

But, you can be certain that any key that existed prior to the creation of this iterator, and was not changed during iteration, will be observed as expected.

Panics

This will panic if the provided range’s end_bound() == Bound::Excluded(K::MIN).

Examples

use concurrent_map::ConcurrentMap;

let data = vec![
    ("key 1", 1),
    ("key 2", 2),
    ("key 3", 3)
];

let map: ConcurrentMap<&'static str, usize> = data.iter().copied().collect();

let r1: Vec<_> = map.range("key 1"..="key 3").collect();
assert_eq!(&data, &r1);

let r2: Vec<_> = map.range("key 1".."key 3").collect();
assert_eq!(&data[..2], &r2);

let r3: Vec<_> = map.range("key 2"..="key 3").collect();
assert_eq!(&data[1..], &r3);

let r4: Vec<_> = map.range("key 4"..).collect();
assert!(r4.is_empty());

pub fn update_and_fetch<F>(&self, key: K, f: F) -> Option<V>where F: FnMut(Option<&V>) -> Option<V>, V: PartialEq,

Fetch the value, apply a function to it and return the result. Similar to ConcurrentMap::cas, returning a None from the provided closure will cause a deletion of the value.

Note

This may call the function multiple times if the value has been changed from other threads in the meantime. This function essentially implements the common CAS loop pattern for atomically pushing a function to some shared data.

Examples

let map = concurrent_map::ConcurrentMap::<&'static str, usize>::default();

fn increment(old_opt: Option<&usize>) -> Option<usize> {
    let incremented = match old_opt {
        Some(old) => {
            old + 1
        }
        None => 0,
    };

    // returning `None` here means "delete this value"
    Some(incremented)
}

assert_eq!(map.update_and_fetch("counter", increment), Some(0));
assert_eq!(map.update_and_fetch("counter", increment), Some(1));
assert_eq!(map.update_and_fetch("counter", increment), Some(2));
assert_eq!(map.update_and_fetch("counter", increment), Some(3));

// push a "deletion" that returns None
assert_eq!(map.update_and_fetch("counter", |_| None), None);

pub fn fetch_and_update<F>(&self, key: K, f: F) -> Option<V>where F: FnMut(Option<&V>) -> Option<V>, V: PartialEq,

Fetch the value, apply a function to it and return the previous value. Similar to ConcurrentMap::cas, returning a None from the provided closure will cause a deletion of the value.

Note

This may call the function multiple times if the value has been changed from other threads in the meantime. This function essentially implements the common CAS loop pattern for atomically pushing a function to some shared data.

Examples

let map = concurrent_map::ConcurrentMap::<&'static str, usize>::default();

fn increment(old_opt: Option<&usize>) -> Option<usize> {
    let incremented = match old_opt {
        Some(old) => {
            old + 1
        }
        None => 0,
    };

    // returning `None` here means "delete this value"
    Some(incremented)
}

assert_eq!(map.fetch_and_update("counter", increment), None);
assert_eq!(map.fetch_and_update("counter", increment), Some(0));
assert_eq!(map.fetch_and_update("counter", increment), Some(1));
assert_eq!(map.fetch_and_update("counter", increment), Some(2));

// push a "deletion" that returns the previous value, essentially
// mimicking the ConcurrentMap::remove method.
assert_eq!(map.fetch_and_update("counter", |_| None), Some(3));

// verify that it's not present
assert_eq!(map.get("counter"), None);

impl<K, V, const FANOUT: usize, const LOCAL_GC_BUFFER_SIZE: usize> ConcurrentMap<K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>where K: 'static + Clone + Minimum + Ord + Send + Sync, V: 'static + Clone + Send + Sync + Ord,

pub fn fetch_min(&self, key: K, value: V) -> Option<V>

Similar to std::sync::atomic::AtomicU64::fetch_min in spirit, this atomically sets the value to the minimum of the previous value and the provided value.

The previous value is returned. None is returned if there was no previous value, in which case the value is set to the provided value. The value is unchanged if the current value is already lower than the provided value.

Examples

let map = concurrent_map::ConcurrentMap::<&'static str, usize>::default();

// acts as an insertion if no value is present
assert_eq!(map.fetch_min("key 1", 5), None);

// sets the value to the new lower value, returns the old value
assert_eq!(map.fetch_min("key 1", 2), Some(5));

// fails to set the value to a lower number, returns the
// current value.
assert_eq!(map.fetch_min("key 1", 10), Some(2));

pub fn fetch_max(&self, key: K, value: V) -> Option<V>

Similar to std::sync::atomic::AtomicU64::fetch_max in spirit, this atomically sets the value to the maximum of the previous value and the provided value.

The previous value is returned. None is returned if there was no previous value, in which case the value is set to the provided value. The value is unchanged if the current value is already higher than the provided value.

Examples

let map = concurrent_map::ConcurrentMap::<&'static str, usize>::default();

// acts as an insertion if no value is present
assert_eq!(map.fetch_max("key 1", 5), None);

// sets the value to the new higher value, returns the old value
assert_eq!(map.fetch_max("key 1", 10), Some(5));

// fails to set the value to a higher number, returns the
// current value.
assert_eq!(map.fetch_max("key 1", 2), Some(10));

Trait Implementations

impl<K, V, const FANOUT: usize, const LOCAL_GC_BUFFER_SIZE: usize> Clone for ConcurrentMap<K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>where K: 'static + Clone + Minimum + Ord + Send + Sync + Clone, V: 'static + Clone + Send + Sync + Clone,

fn clone(&self) -> ConcurrentMap<K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<K, V, const FANOUT: usize, const LOCAL_GC_BUFFER_SIZE: usize> Debug for ConcurrentMap<K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>where K: 'static + Debug + Clone + Minimum + Ord + Send + Sync, V: 'static + Debug + Clone + Send + Sync,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<K, V, const FANOUT: usize, const LOCAL_GC_BUFFER_SIZE: usize> Default for ConcurrentMap<K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>where K: 'static + Clone + Minimum + Ord + Send + Sync, V: 'static + Clone + Send + Sync,

fn default() -> ConcurrentMap<K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>

Returns the “default value” for a type.

impl<K, V, const FANOUT: usize, const LOCAL_GC_BUFFER_SIZE: usize> FromIterator<(K, V)> for ConcurrentMap<K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>where K: 'static + Clone + Minimum + Ord + Send + Sync, V: 'static + Clone + Send + Sync,

fn from_iter<I: IntoIterator<Item = (K, V)>>(iter: I) -> Self

Creates a value from an iterator.

impl<'a, K, V, const FANOUT: usize, const LOCAL_GC_BUFFER_SIZE: usize> IntoIterator for &'a ConcurrentMap<K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>where K: 'static + Clone + Minimum + Ord + Send + Sync, V: 'static + Clone + Send + Sync,

type Item = (K, V)

The type of the elements being iterated over.

type IntoIter = Iter<'a, K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>

Which kind of iterator are we turning this into?

fn into_iter(self) -> Self::IntoIter

Creates an iterator from a value.

impl<K, V, const FANOUT: usize, const LOCAL_GC_BUFFER_SIZE: usize> PartialEq for ConcurrentMap<K, V, FANOUT, LOCAL_GC_BUFFER_SIZE>where K: 'static + Debug + Clone + Minimum + Ord + Send + Sync + PartialEq, V: 'static + Debug + Clone + Send + Sync + PartialEq,

fn eq(&self, other: &Self) -> bool

This method tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

This method tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.