Skip to content

Commit

Permalink
LenCachingMutex (openethereum#9988)
Browse files Browse the repository at this point in the history
New util LenCachingMutex can be used in place of a Mutex, when working
with collections, or other types with a len() method. When
the Guard is dropped, the value returned from len() is stored
into an AtomicUsize and can be queried using load_len() without
needing to lock the Mutex. Implementations for ```Vec``` and
```VecDeque```.

Now used in [Verification](https://github.com/paritytech/parity-ethereum/blob/4ded4181a651fe9c26850f908e4ab63bf952cab6/ethcore/src/verification/queue/mod.rs#L196) so that calls to ```VerificationQueue.queue_info()```
no longer require locks.
  • Loading branch information
mattrutherford authored and insipx committed Dec 17, 2018
1 parent d32c713 commit 4a5a9fe
Show file tree
Hide file tree
Showing 6 changed files with 170 additions and 11 deletions.
8 changes: 8 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions ethcore/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ kvdb-rocksdb = "0.1.3"
serde = "1.0"
serde_derive = "1.0"
tempdir = {version="0.3", optional = true}
len-caching-mutex = { path = "../util/len-caching-mutex" }

[target.'cfg(any(target_os = "linux", target_os = "macos", target_os = "windows", target_os = "android"))'.dependencies]
hardware-wallet = { path = "../hw" }
Expand Down
1 change: 1 addition & 0 deletions ethcore/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,7 @@ extern crate journaldb;
extern crate serde;
#[cfg(any(test, feature = "json-tests", feature = "test-helpers"))]
extern crate tempdir;
extern crate len_caching_mutex;

#[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows", target_os = "android"))]
extern crate hardware_wallet;
Expand Down
23 changes: 12 additions & 11 deletions ethcore/src/verification/queue/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ use io::*;
use error::{BlockError, ImportErrorKind, ErrorKind, Error};
use engines::EthEngine;
use client::ClientIoMessage;
use len_caching_mutex::LenCachingMutex;

use self::kind::{BlockLike, Kind};

Expand Down Expand Up @@ -195,9 +196,9 @@ impl QueueSignal {

struct Verification<K: Kind> {
// All locks must be captured in the order declared here.
unverified: Mutex<VecDeque<K::Unverified>>,
verifying: Mutex<VecDeque<Verifying<K>>>,
verified: Mutex<VecDeque<K::Verified>>,
unverified: LenCachingMutex<VecDeque<K::Unverified>>,
verifying: LenCachingMutex<VecDeque<Verifying<K>>>,
verified: LenCachingMutex<VecDeque<K::Verified>>,
bad: Mutex<HashSet<H256>>,
sizes: Sizes,
check_seal: bool,
Expand All @@ -207,9 +208,9 @@ impl<K: Kind> VerificationQueue<K> {
/// Creates a new queue instance.
pub fn new(config: Config, engine: Arc<EthEngine>, message_channel: IoChannel<ClientIoMessage>, check_seal: bool) -> Self {
let verification = Arc::new(Verification {
unverified: Mutex::new(VecDeque::new()),
verifying: Mutex::new(VecDeque::new()),
verified: Mutex::new(VecDeque::new()),
unverified: LenCachingMutex::new(VecDeque::new()),
verifying: LenCachingMutex::new(VecDeque::new()),
verified: LenCachingMutex::new(VecDeque::new()),
bad: Mutex::new(HashSet::new()),
sizes: Sizes {
unverified: AtomicUsize::new(0),
Expand Down Expand Up @@ -332,7 +333,7 @@ impl<K: Kind> VerificationQueue<K> {
return;
}

wait.wait(&mut unverified);
wait.wait(unverified.inner_mut());
}

if let State::Exit = *state.0.lock() {
Expand Down Expand Up @@ -453,7 +454,7 @@ impl<K: Kind> VerificationQueue<K> {
pub fn flush(&self) {
let mut unverified = self.verification.unverified.lock();
while !unverified.is_empty() || !self.verification.verifying.lock().is_empty() {
self.empty.wait(&mut unverified);
self.empty.wait(unverified.inner_mut());
}
}

Expand Down Expand Up @@ -595,18 +596,18 @@ impl<K: Kind> VerificationQueue<K> {
use std::mem::size_of;

let (unverified_len, unverified_bytes) = {
let len = self.verification.unverified.lock().len();
let len = self.verification.unverified.load_len();
let size = self.verification.sizes.unverified.load(AtomicOrdering::Acquire);

(len, size + len * size_of::<K::Unverified>())
};
let (verifying_len, verifying_bytes) = {
let len = self.verification.verifying.lock().len();
let len = self.verification.verifying.load_len();
let size = self.verification.sizes.verifying.load(AtomicOrdering::Acquire);
(len, size + len * size_of::<Verifying<K>>())
};
let (verified_len, verified_bytes) = {
let len = self.verification.verified.lock().len();
let len = self.verification.verified.load_len();
let size = self.verification.sizes.verified.load(AtomicOrdering::Acquire);
(len, size + len * size_of::<K::Verified>())
};
Expand Down
10 changes: 10 additions & 0 deletions util/len-caching-mutex/Cargo.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
[package]
description = "Mutex with cached len, for use with collections"
homepage = "http://parity.io"
license = "GPL-3.0"
name = "len-caching-mutex"
version = "0.1.0"
authors = ["Parity Technologies <[email protected]>"]

[dependencies]
parking_lot = "0.6"
138 changes: 138 additions & 0 deletions util/len-caching-mutex/src/lib.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,138 @@
// Copyright 2015-2018 Parity Technologies (UK) Ltd.
// This file is part of Parity.

// Parity is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.

// Parity is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.

// You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
extern crate parking_lot;

use std::collections::VecDeque;
use std::ops::{Deref, DerefMut};
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;

use parking_lot::{Mutex, MutexGuard};

/// Implement to allow a type with a len() method to be used
/// with [`LenCachingMutex`](struct.LenCachingMutex.html)
pub trait Len {
fn len(&self) -> usize;
}

impl<T> Len for Vec<T> {
fn len(&self) -> usize { self.len() }
}

impl<T> Len for VecDeque<T> {
fn len(&self) -> usize { self.len() }
}

/// Can be used in place of a `Mutex` where reading `T`'s `len()` without
/// needing to lock, is advantageous.
/// When the Guard is released, `T`'s `len()` will be cached.
/// The cached `len()` may be at most 1 lock behind current state.
pub struct LenCachingMutex<T> {
data: Mutex<T>,
len: AtomicUsize,
}

impl<T: Len> LenCachingMutex<T> {
pub fn new(data: T) -> LenCachingMutex<T> {
LenCachingMutex {
len: AtomicUsize::new(data.len()),
data: Mutex::new(data),
}
}

/// Load the most recent value returned from your `T`'s `len()`
pub fn load_len(&self) -> usize {
self.len.load(Ordering::SeqCst)
}

pub fn lock(&self) -> Guard<T> {
Guard {
mutex_guard: self.data.lock(),
len: &self.len,
}
}

pub fn try_lock(&self) -> Option<Guard<T>> {
Some( Guard {
mutex_guard: self.data.try_lock()?,
len: &self.len,
})
}
}

pub struct Guard<'a, T: Len + 'a> {
mutex_guard: MutexGuard<'a, T>,
len: &'a AtomicUsize,
}

impl<'a, T: Len> Guard<'a, T> {
pub fn inner_mut(&mut self) -> &mut MutexGuard<'a, T> {
&mut self.mutex_guard
}

pub fn inner(&self) -> &MutexGuard<'a, T> {
&self.mutex_guard
}
}

impl<'a, T: Len> Drop for Guard<'a, T> {
fn drop(&mut self) {
self.len.store(self.mutex_guard.len(), Ordering::SeqCst);
}
}

impl<'a, T: Len> Deref for Guard<'a, T> {
type Target = T;
fn deref(&self) -> &T {
self.mutex_guard.deref()
}
}

impl<'a, T: Len> DerefMut for Guard<'a, T> {
fn deref_mut(&mut self) -> &mut T {
self.mutex_guard.deref_mut()
}
}

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

#[test]
fn caches_len() {
let v = vec![1,2,3];
let lcm = LenCachingMutex::new(v);
assert_eq!(lcm.load_len(), 3);
lcm.lock().push(4);
assert_eq!(lcm.load_len(), 4);
}

#[test]
fn works_with_vec() {
let v: Vec<i32> = Vec::new();
let lcm = LenCachingMutex::new(v);
assert!(lcm.lock().is_empty());
}

#[test]
fn works_with_vecdeque() {
let v: VecDeque<i32> = VecDeque::new();
let lcm = LenCachingMutex::new(v);
lcm.lock().push_front(4);
assert_eq!(lcm.load_len(), 1);
}
}

0 comments on commit 4a5a9fe

Please sign in to comment.