2
\$\begingroup\$

I wrote a small vector with a few lines of unsafe Rust.

The idea is to have a vector, which can be read from simultaneously, but it needs the ability to grow. To realize that, the storage is organized in pages. As page, I use arrayvec:

const ELEMENTS_PER_PAGE: usize = 1024;

type Page<T> = ArrayVec<[T; ELEMENTS_PER_PAGE]>;

The main problem is, when putting those pages in a Vec, the access may gets invalidated, while pushing to the vector. To solve this, I have an array of *mut Page<T> (so basically Vec<Box<Page<T>>>) and another pointer which points to this array: *mut *mut Page<T>. When adding pages, the vector will be copied without cloning the Pages (this is why I need the raw pointer). Now the *mut *mut ptr will be updated to the new vector and everyone will be headed to the correct memory.

As a small optimization, I use Box<[_]> instead of Vec, since the Vec will never grow directly:

pub struct ConstVec<T> {
    // storage, which holds the actual pages
    pages: Mutex<Box<[*mut Page<T>]>>,
    // points to the storage. Used for wait-free access.
    pages_pointer: AtomicPtr<*mut Page<T>>,
    len: AtomicUsize,
}

We need two helper functions to get the indices for the page and the element in a specific page:

const fn page_index(index: usize) -> usize {
    index / ELEMENTS_PER_PAGE
}

const fn element_index(index: usize) -> usize {
    index % ELEMENTS_PER_PAGE
}

and a len method:

pub fn len(&self) -> usize {
    self.len.load(atomic::Ordering::Acquire)                           // 1
}

This is the function for pushing into the vector:

pub fn push(&self, value: T) {
    let mut pages = self.pages.lock().unwrap();

    let index = self.len.load(atomic::Ordering::Acquire);              // 1
    let page_index = Self::page_index(index);

    // do we need an new page?
    if page_index == pages.len() {
        // allocate a new vector, which will replace the old one
        // and copy old elements into it.
        let mut new_pages = Vec::with_capacity(page_index + 1);
        new_pages.extend(pages.iter().cloned());
        new_pages.push(Box::into_raw(Box::new(Page::new())));          // 2

        // Update the pages pointer first. This will be used
        // to receive data. The pointers remains valid.
        self.pages_pointer
            .store(new_pages.as_mut_ptr(), atomic::Ordering::SeqCst);  // 1
        // replace "vector"
        mem::replace(pages.deref_mut(), new_pages.into_boxed_slice());
    }
    unsafe {
        (*pages[page_index]).push(value);                              // 2
    }
    self.len.store(index + 1, atomic::Ordering::Release);              // 1
}

And of course we need to implement Drop:

fn drop(&mut self) {
    for page_ptr in self.pages.lock().unwrap().iter() {
        unsafe { Box::from_raw(*page_ptr) };                           // 2
    }
}

Now two questions came up (see the code numbers for references):

  1. Am I using the right Ordering for atomic access? If I see correctly, I can use Acquire/Release everywhere but in storing the pages_pointer. This needs to be SeqCst, because it would be fatal, if the mem::replace would be executed before (the old pointer would dangle). Also I don't thing, Relaxed can be used anywhere here. Are my assumptions correct?

  2. Are those unsafe actually safe?

playground

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.