Easy Concurrency Mastery: Exploring the Read-Write Lock Pattern in Rust for Performance

Introduction

In another article we discussed the Lock pattern. In this article we used the Mutex type. The problem with this type is, is that it doesn’t distinguish between reading from a resource, like accessing an element in a vector, and writing to it.

In cases where many threads need to read a resource at one, and there are a few write-operations, the Read-Write Lock pattern can help. This pattern allows concurrent access for readers of the resource. In the case however of writing an exclusive lock is granted, and all read operations are blocked. This last part could be a source of a deadlock, in case the writer takes a long time, or in some bad cases, never finishes for some reason.

In Rust, this pattern in implemented using the RwLock struct. This struct has a write() method, which grants an exclusive lock, and a read() which grants shared read access.

Implementation in Rust

A typical use for a read write lock would be to build a cache. A cache usually has quite a few more reads than writes, and therefore is a good candidate to show this pattern.

Let’s start with our preliminaries:

use std::sync::{Arc, RwLock, PoisonError};
use std::thread;
use std::collections::HashMap;
use std::fmt;

Line by line:

1. The first line imports three synchronization modules:
   • Arc: Atomic Reference Counted. This type allows data to be share ownership of the same resources.
   • RwLock: This is the read write lock. The basic idea behind this is that multiple threads can read the same data, but only one thread at a time can modify the data.
   • PoisonError: in this exampe we use to handle situations where a thread panics while holding a lock. This could leave shared data in an inconsistent state.
2. The std::thread module is used to provide threading capabilities.
3. Because the example cache will be implemented using key value pairs, we use the std::collections::HashMap type.
4. The std::fmt module finally is used to format errors.

The CacheError type

Error handling is very important, especially in production code. That is why we define a custom error. This is implemented as an enum so it can be extended when needed.

#[derive(Debug)]
enum CacheError {
    LockPoisoned,
}

Next we implement the fmt::Display trait so the error can be formatted as user friendly string:

impl fmt::Display for CacheError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            CacheError::LockPoisoned => write!(f, "Lock was poisoned"),
        }
    }
}

We also implement the std::error::Error trait to make the error compatible Rust’s error handling ecosystem, allowing CacheError to be used with functions and methods which expect types implementing this trait:

impl std::error::Error for CacheError {}

Lastly we implement a From implementation. This enables automatic conversion from PoisonError<T>, the error type returned when the application encounters a poisoned lock into our custom CacheError. Having this conversion will allow us to use the ? operator in functions returning a Result<T,CacheError> as any PoisonError<T> will be automatically converted.

impl<T> From<PoisonError<T>> for CacheError {
    fn from(_: PoisonError<T>) -> Self {
        CacheError::LockPoisoned
    }
}

The Cache type

The basic Cache struct is quite simple:

struct Cache {
    data: RwLock<HashMap<String, i32>>,
}

It basically contains a HashMap with strings for keys and integer values wrapped in an RwLock.

The implementation is not that hard either, we will start with the constructor:

    fn new() -> Self {
        Cache {
            data: RwLock::new(HashMap::new()),
        }
    }

All this does is create a new HashMap wrapped in a new RwLock

Inserting an item into the cache is a little more involved:

    fn insert(&self, key: String, value: i32) -> Result<(), CacheError> {
        let mut data = self.data.write()?;
        data.insert(key, value);
        Ok(())
    }

Line by line:

1. The method receives the key and the value to be inserted. It returns a Result<(),CacheError> which means that when things go well, and empty Ok() result is returned. When something does go wrong, for example when a write lock can not be obtained it returns an error.
2. The first line tries and get exclusive write access to the HashMap. The ? at the the end of the line means any errors will be propagated.
3. Because we now have exclusive write access, we can insert the key and value.
4. Everything went well, so we return an Ok(()) result.
5. How is the lock released. The data variable is of type RwLockWriteGuard which holds the lock. When this goes out of scope Rust automatically calls the drop() method, which releases the lock.

The get() method is used to retrieve the value associated with the key:

    fn get(&self, key: &str) -> Result<Option<i32>, CacheError> {
        let data = self.data.read()?;
        Ok(data.get(key).copied())
    }

Line by line:

1. The method receives the key as a string, and returns a Result<Option<i32>,CacheError>. Why would it return an Option<i32>. This is because a key might not have an associated value. Also, failure to optain a read lock, will result in a cache error.
2. The method starts by trying to obtain a read lock on the data. If this fails, the resulting error will be propagated.
3. The call to data.get(key) can result in two things, either a Some(&i32) when the value is available or a None when it is not.
4. Why the call to copied()? This converts Some(&i32) to Some(i32) which will be returned when a value has been found. If no value was found, None is returned.

The get_all() method more or less works the same:

    fn get_all(&self) -> Result<HashMap<String, i32>, CacheError> {
        let data = self.data.read()?;
        Ok(data.clone())
    }

Finally the len() method simply returns the length of the hash map:

    fn len(&self) -> Result<usize, CacheError> {
        let data = self.data.read()?;
        Ok(data.len())
    }

The full implementation:

impl Cache {
    fn new() -> Self {
        Cache {
            data: RwLock::new(HashMap::new()),
        }
    }

    fn insert(&self, key: String, value: i32) -> Result<(), CacheError> {
        let mut data = self.data.write()?;
        data.insert(key, value);
        Ok(())
    }

    fn get(&self, key: &str) -> Result<Option<i32>, CacheError> {
        let data = self.data.read()?;
        Ok(data.get(key).copied())
    }

    fn get_all(&self) -> Result<HashMap<String, i32>, CacheError> {
        let data = self.data.read()?;
        Ok(data.clone())
    }

    fn len(&self) -> Result<usize, CacheError> {
        let data = self.data.read()?;
        Ok(data.len())
    }
}

Testing time

Now we can test our cache in the main method:

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let cache = Arc::new(Cache::new());


    cache.insert("user1".to_string(), 100)?;
    cache.insert("user2".to_string(), 200)?;
    cache.insert("user3".to_string(), 300)?;

    println!("Cache size: {}", cache.len()?);
    println!("user1 score: {:?}", cache.get("user1")?);
    println!("user2 score: {:?}", cache.get("user2")?);
    println!("nonexistent score: {:?}", cache.get("nonexistent")?);


    let mut handles = vec![];

    // Writer threads
    for i in 0..3 {
        let cache_clone = Arc::clone(&cache);
        let handle = thread::spawn(move || -> Result<(), CacheError> {
            for j in 0..5 {
                let key = format!("writer{}_item{}", i, j);
                cache_clone.insert(key, i * 100 + j)?;
            }
            Ok(())
        });
        handles.push(handle);
    }


    for i in 0..2 {
        let cache_clone = Arc::clone(&cache);
        let handle = thread::spawn(move || -> Result<(), CacheError> {
            for j in 0..3 {
                let key = format!("user{}", j + 1);
                if let Some(value) = cache_clone.get(&key)? {
                    println!("Reader {}: {} = {}", i, key, value);
                }
            }
            Ok(())
        });
        handles.push(handle);
    }


    for handle in handles {
        if let Err(e) = handle
            .join()
            .unwrap_or_else(|_| Err(CacheError::LockPoisoned))
        {
            eprintln!("Thread error: {}", e);
        }
    }

    println!("Final cache size: {}", cache.len()?);
    println!("All items: {:?}", cache.get_all()?);

    Ok(())
}

Line by line:

1. The first thing to note is the fact that the main function has a return value. This means that we can use the ? operator inside the function, meaning any errors will be propagated to the caller, i.e. if for example acquiring a lock fails, the program will terminate and print the error message.
2. We start by creating a new cache. It is wrapped in an Arc so it can be safely shared across threads.
3. Next we insert some values into the cache, and retrieve cache.
4. Now we can spawn 3 writer threads, each inserting 5 unique items in the cache.
5. Next we spawn some reader threads.
6. With the threads spawned we wait for them to finish, and print any occurred which may have occurred.
7. Finally print the final cache size, and all current items.

Conclusion

Although the Mutex type performs quite well, and is stable, sometime in the case of many read operations and few write operations, you can boost performance by using the RwLock struct.

If the writer-thread blocks or panics, there is a chance of deadlock. Since only a write can block access to the resource, the cause of this problem can usually be found relatively easily.