Suppose we are developing a synthesizer which produces sounds when keys are pressed on a MIDI keyboard. The audio library calls a function we provide once ever 6 or so milliseconds to request a list of samples from us. The library calls our function with 2 arguments: 1) How many samples it wants 2) what key presses we need to handle. The callback function uses a precomputed list of samples to generate sounds every time it is called. To modify the properties of the sounds that are produced, the user edits settings with a user interface.

It would be painful (and incorrect) to attempt to handle UI events in the realtime thread, so we will run a UI thread to handle the UI events. Whenever the UI thread gets an event to handle, it needs to compute a new sample list, then send the list to the realtime thread.

Since we can't lock, let's use a queue to send some sort of message between threads. The queue that we choose needs to have a few properties:

• Must be a lock free queue
• Must be able to preallocate all of its nodes (cant't allocate or free memory for a node on a push or pull)

I want to place messages on the heap so that they do not need to be copied as we move them around. If messages lives on the heap, we must ensure they are allocated and freed outside of the realtime thread (we can't call allocation functions on the realtime thread).

For the sake of these examples, let's assume that the built-in Rust mpsc channel is an appropriate lock free queue. It will be pretty easy to swap this with something different later, and, if we use the standard library, all of the examples will easily run in the Rust playground. We are also going to fake a bunch of the details of the audio library.

We now need to build the GC that I promised we would build, to clean up after us, outside of the realtime thread. Sneak peak, once the GC is implemented, all we have to change is UIThread::run, in a very small way:

/// All of the UI thread code
fn run(&mut self) {
    let mut gc = GC::new(); // + NEW LINE

    // create 5 "ui events"
    for i in 0..5 {
        let volume = i as f32 / 5.0;
        let samples = Arc::new(self.compute_samples(volume));
        self.collector.track(samples.clone()); // + NEW LINE

        // send the samples to the other thread
        println!("[ui] sending new samples. Second sample: {}", samples[1]);
        self.outgoing.send(Message::NewSamples(samples)).unwrap();
    }

    // tell the other thread to shutdown
    self.outgoing.send(Message::Shutdown).unwrap();
}

With that in mind, lets sketch out the interface for the Garbage Collector.

/// A garbage collector for Arc<T> pointers
struct GC<T> {
    // ...
}

impl<T> GC<T> {
    /// Construct a new garbage collector and start the collection thread
    fn new() -> Self {
        // ...
    }

    /// Instruct the garbage collector to monitor this Arc<T>
    /// When no references remain, the collector will `drop` the value
    fn track(&mut self, t: Arc<T>) {
        // ...
    }
}

First think about the track method. All this method needs to do is move it's argument into some list (vector) of pointers. We will keep this vector in the GC thread struct so that each of the references will live until the GC thread is shut down or until the GC drops them.

struct GC<T> {
    pool: Vec<Arc<T>>,
}

impl<T> GC<T> {
    // ...

    pub fn track(&mut self, t: Arc<T>) {
        self.pool.push(t);
    }
}

Now lets think about the garbage collection logic. Since we have a Vec<Arc<T>>, we will want to iterate over it, removing any elements which meet (or fail) a condition. We can use Vec::retain to do this. Something like the following might work:

pool.retain(|e| {
    if /* has more than one reference */ {
        return true
    } else {
        return false
    }
})

Looking at the Arc docs, there are a few ways we can figure out if the Arc has only one remaining reference:

• Attempt to consume the Arc with Arc::try_unwrap, if this fails, we know that it has more than one reference. Unforunately, this method requires moving the Arc out of the vector, which is not ideal if we want to use Vec::retain.
• Arc::strong_count - this is currently marked as unstable. Looks like what we might want to use though.
• Arc::get_mut could possibly be used the same way we would use Arc::try_unwrap, without moving the Arc containing in the vector unless we want to remove it.

We don't have lots of options, so I'm going to go ahead and use Arc::strong_count. This is (for now) the most natural way to solve the problem:

pool.retain(|e: Arc<_>| {
    if Arc::strong_count(&e) {
        return true
    } else {
        return false
    }
})

Let's move on to new. The new method needs to start new thread which will run the pool.retain thing every once and a while. We also need to hold on to a thread handle so that we can eventually join the thread. The join handle is wrapped in an Option, we will see why quite a bit later.

/// A garbage collector for Arc<T> pointers
struct GC<T> {
    pool: Vec<Arc<T>>,
    thread: Option<thread::JoinHandle<()>>,
}

impl<T> GC<T> {
    // private. cleans up any dead pointers in a pool
    fn cleanup(pool: &mut Vec<Arc<T>>) {
        pool.retain(|e: &Arc<_>| {
            if Arc::strong_count(&e) > 1 {
                return true
            } else {
                return false
            }
        });
    }

    pub fn new() -> Self {
        let pool = Vec::new();

        // create a closure which will become a new thread
        let gc = || {
            loop {
                GC::cleanup(&mut pool);

                // wait for 100 milliseconds, then scan again
                let sleep = std::time::Duration::from_millis(100);
                thread::sleep(sleep);
            }
        };

        // spawns a new thread and returns a handle to the thread
        let gc_thread = thread::spawn(gc);

        GC {
            pool:   pool,
            thread: Some(gc_thread),
        }
    }

    pub fn track(&mut self, t: Arc<T>) {
        self.pool.push(t);
    }
}

fn main() {
    let (tx, rx) = mpsc::sync_channel(0);
    let rt = RealtimeThread::new(rx);
    let ui = UIThread::new(tx);
    run_threads(rt, ui);
}

We've written a bunch of new code, better make sure it compiles (Rust playground):

error[E0277]: the trait bound `T: std::marker::Send` is not satisfied
   --> <anon>:154:25
    |
154 |         let gc_thread = thread::spawn(gc);
    |                         ^^^^^^^^^^^^^ the trait `std::marker::Send` is not implemented for `T`
    |
    = help: consider adding a `where T: std::marker::Send` bound
    = note: required because of the requirements on the impl of `std::marker::Send` for `std::sync::Arc<T>`
    = note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<std::sync::Arc<T>>`
    = note: required because it appears within the type `alloc::raw_vec::RawVec<std::sync::Arc<T>>`
    = note: required because it appears within the type `std::vec::Vec<std::sync::Arc<T>>`
    = note: required because of the requirements on the impl of `std::marker::Send` for `&mut std::vec::Vec<std::sync::Arc<T>>`
    = note: required because it appears within the type `[closure@<anon>:143:18: 151:10 pool:&mut std::vec::Vec<std::sync::Arc<T>>]`
    = note: required by `std::thread::spawn`

error[E0277]: the trait bound `T: std::marker::Sync` is not satisfied
   --> <anon>:154:25
    |
154 |         let gc_thread = thread::spawn(gc);
    |                         ^^^^^^^^^^^^^ the trait `std::marker::Sync` is not implemented for `T`
    |
    = help: consider adding a `where T: std::marker::Sync` bound
    = note: required because of the requirements on the impl of `std::marker::Send` for `std::sync::Arc<T>`
    = note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<std::sync::Arc<T>>`
    = note: required because it appears within the type `alloc::raw_vec::RawVec<std::sync::Arc<T>>`
    = note: required because it appears within the type `std::vec::Vec<std::sync::Arc<T>>`
    = note: required because of the requirements on the impl of `std::marker::Send` for `&mut std::vec::Vec<std::sync::Arc<T>>`
    = note: required because it appears within the type `[closure@<anon>:143:18: 151:10 pool:&mut std::vec::Vec<std::sync::Arc<T>>]`
    = note: required by `std::thread::spawn`

error: aborting due to 2 previous errors

Oops, this isn't good. This error makes it feel sort of like Rust hates us, but the compiler is actually doing us a massive favor.

In Rust, there are a few thread safety "marker traits" called Send and Sync. The compiler is telling us that our generic type T doesn't implement either of them.

Put very loosely, if something implements Send, it is safe to send it between threads. Sync is considerably more subtle and quite difficult to wrap your head around, but we can sort of say that, if something implements Sync, we can access the same instance of it from multiple threads. For more info, you can read this blog post, but you shouldn't need any more than what I've given to get through the rest of my post.

So anyway, Rust is telling us that we have a thread safety problem, but we haven't guaranteed that we can safely copy and access values of our type T between the garbage collector thread and any other threads.

I know that T must be Send, because it has to be sent between threads, so let's go ahead and add that restriction:

/// A garbage collector for Arc<T> pointers
struct GC<T: Send> {
    pool: Vec<Arc<T>>,
    thread: Option<thread::JoinHandle<()>>,
}

impl<T: Send> GC<T> {
// ....

Rust playground link

Hoorary, the Send error is gone! Unfortunately, we still have the issue with Sync. Let's look more closely at the error we are getting:

error[E0277]: the trait bound `T: std::marker::Sync` is not satisfied
   --> <anon>:154:25
    |
154 |         let gc_thread = thread::spawn(gc);
    |                         ^^^^^^^^^^^^^ the trait `std::marker::Sync` is not implemented for `T`
    |
    = help: consider adding a `where T: std::marker::Sync` bound
    = note: required because of the requirements on the impl of `std::marker::Send` for `std::sync::Arc<T>`
    = note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<std::sync::Arc<T>>`
    = note: required because it appears within the type `alloc::raw_vec::RawVec<std::sync::Arc<T>>`
    = note: required because it appears within the type `std::vec::Vec<std::sync::Arc<T>>`
    = note: required because of the requirements on the impl of `std::marker::Send` for `&mut std::vec::Vec<std::sync::Arc<T>>`
    = note: required because it appears within the type `[closure@<anon>:143:18: 151:10 pool:&mut std::vec::Vec<std::sync::Arc<T>>]`
    = note: required by `std::thread::spawn`

error: aborting due to previous error

This error is really confusing, and my solution for it is not going to be much better, but stick with me.

The origin of this error is the Arc<T>. If we want an Arc<T> to implement Send, the T contained in it must implement BOTH Send and Sync. It makes sense that T would need to implement Send, but why does T need to be Sync? Basically, this is because the data the Arc<T> is holding will be shared by anyone who can access the Arc<T>. An Arc can be =clone=ed at any time, so, if we are allowed to pass it to other threads, it must also be safe for multiple threads to access the underlying data at the same time.

We could add the Sync constraint to our type T to resolve this problem, but does this really make any sense? Nowhere in our application will a message be accessible by more than one thread at a time.

When the UI thread creates a new message, it immediately surrenders all access to the underlying data, by moving the value into the channel. Once the realtime thread has the data, it will be the only thread that actually accesses the data until the data needs to be freed. The GC also is holding a reference to data, but it will never actually touch the data in any way, until it frees it. When the GC thread frees the memory holding the data, we know that there will be no other references to the memory in the program.

I might be wrong about this (please let me know if I am), but I think that we don't actually need the type T to be Sync. The compiler will never let us get away with this (because it doesn't know all of these properties) but we can let it know that it should trust us, with a new struct:

struct TrustMe<T> {
    pub inner: T
}

unsafe impl<T> Send for TrustMe<T> {}

This will tell the compiler "yes, this thing is Send", even when it actually isn't, so the implementation of the trait Send is unsafe.

Now, we can create a Send=able =TrustMe<Arc<T>>, and the compiler will trust us when we share these =Arc<T>=s between threads.

Now, lets add this to our GC:

/// A garbage collector for Arc<T> pointers
struct GC<T: Send> {
    pool: Vec<TrustMe<Arc<T>>>,
    thread: Option<thread::JoinHandle<()>>,
}

impl<T: Send> GC<T> {
    // private. cleans up any dead pointers in a pool
    fn cleanup(pool: &mut Vec<TrustMe<Arc<T>>>) {
        pool.retain(|e: &TrustMe<Arc<_>>| {
            if Arc::strong_count(&e.inner) > 1 {
                return true
            } else {
                return false
            }
        });
    }

    pub fn new() -> Self {
        let mut pool = Vec::new();

        // create a closure which will become a new thread
        let gc = || {
            loop {
                GC::cleanup(&mut pool);

                // wait for 100 milliseconds, then scan again
                let sleep = std::time::Duration::from_millis(100);
                thread::sleep(sleep);
            }
        };

        // spawns a new thread and returns a handle to the thread
        let gc_thread = thread::spawn(gc);

        GC {
            pool:   pool,
            thread: Some(gc_thread),
        }
    }

    pub fn track(&mut self, t: Arc<T>) {
        let t = TrustMe { inner: t };
        self.pool.push(t);
    }
}

Rust playground link

When we try to compile this, we get YET ANOTHER compiler error. This time, the compiler is whining at us with "the parameter type T may not live long enough". This error message is frustrating, but, we are using Rust because we want to be very careful with memory safety, so lets try to keep going.

The new thread that we have created could run until the termination of the program, so any data which the thread might be holding onto also must be able to live until the termination of the program.

The compiler is telling us that we need to add a "lifetime specifier" to our type T. In this case, it is telling us that the lifetime of any T which is managed by the GC must be 'static. The 'static lifetime indicates that values of type T + 'static might live for the entire duration of the program.

This might seem excessive, but, it is not possible for the compiler to determine when in the program our thread will terminate (if it could we would have solved the halting problem), so the maximum lifetime MUST potentially be the entire duration of the program. Note that, this doesn't mean that all the values stored in the GC will necessarily live for the entire lifetime of the program (if they did, we wouldn't be cleaning up garbage). This condition just means that they might live that long.

Anyway, we can now add the + 'static specifier the compiler has asked us to add, and try to compile this one more time.

/// A garbage collector for Arc<T> pointers
struct GC<T: Send + 'static> {

// ...

impl<T: Send + 'static> GC<T> {

// ...

GUESS WHAT IT DIDN'T WORK.

error[E0373]: closure may outlive the current function, but it borrows `pool`, which is owned by the current function
   --> <anon>:149:18
    |
149 |         let gc = || {
    |                  ^^ may outlive borrowed value `pool`
150 |             loop {
151 |                 GC::cleanup(&mut pool);
    |                                  ---- `pool` is borrowed here
    |
help: to force the closure to take ownership of `pool` (and any other referenced variables), use the `move` keyword, as shown:
    |         let gc = move || {

error: aborting due to previous error

Once again, this is a good thing, I promise! Now, the compiler is trying to tell us that the vector named pool is being accessed from two different places. The compiler wants us to have the new thread take ownership of the vector, but this highlights an interesting problem. We need to allow both the GC thread, and any other non-realtime thread, to access the vector, at the same time.

The compiler has prevented us from accessing the same data from multiple threads.

To solve this, we can just wrap the vector in a Mutex and an Arc. The Arc allows us to create one instance of the vector on the heap, and the Mutex makes sure that only one thread can access the heap allocated vector at any given time.

Here are most of the changes:

// introduce some news type aliases to make life a little bit easier
type TrustedArc<T> = TrustMe<Arc<T>>;
type ArcPool<T> = Vec<TrustedArc<T>>;

/// A garbage collector for Arc<T> pointers
struct GC<T: Send + 'static> {
    pool: Mutex<Arc<ArcPool<T>>>,
    thread: Option<thread::JoinHandle<()>>,
}

// ...

impl<T: Send + 'static> GC<T> {
    // ...
    pub fn new() -> Self {
        let pool = Arc::new(Mutex::new(Vec::new()));

        // create a copy of the pool. The GC thread will own this clone
        // and the reference count will be incremented by one
        let thread_arc_copy = pool.clone();

        // create a closure which will become a new thread
        let gc = move || {
            loop {
                // lock the mutex, then let go of it.
                // If we hold the mutex, the UI thread will be blocked every time it asks the
                // collector to track something.
                {
                    let mut pool = thread_arc_copy.lock().unwrap();
                    GC::cleanup(&mut pool);
                }

                // wait for a bit, then scan again
                let sleep = std::time::Duration::from_millis(5);
                thread::sleep(sleep);

            }
        };

        // ....
    }

    pub fn track(&mut self, t: Arc<T>) {
        let t = TrustMe { inner: t };
        let mut pool = self.pool.lock().unwrap();
        pool.push(t);
    }
}

We can finally compile this! Here's a link to the Rust playground. Note that you will need to make sure you compile with the "Nightly" channel.

There are only a few things left to do.

We did it!

For me, this post exemplifies the reasons I am so excited about Rust. The realtime audio world places us into a world where many programming languages are simply not usable. Languages with runtimes that may behave unpredictably cannot meet the extremely strict requirements we must meet for correct realtime operation. Rust allows us to meet all of those requirements and gives us some nice abstractions.

On top of that, the Rust compiler meticulously checks for thread safety violations and memory safety violations. While writing this post, some of the issues the compiler threw at me ('static, for example), are issues I never considered. The compiler caught me and told me "no," so I had to think about what was actually going on.

These checks are absolutely irritating, and sometimes we might want to work around them (like we did with TrustMe). I'm glad to be exposed to potential issues, even if I have to work around the compiler sometimes (so far).

If you made it this far, thank you for reading. I hope you've learned something interesting (maybe even useful).

Discussion on reddit.