Jack thread safety testing

main.rs

use std::sync::mpsc::{sync_channel, SyncSender, Receiver};
use std::thread::{ThreadId, current, sleep};
use std::time::{Duration, Instant};
use std::collections::{HashSet, HashMap};

use jack::{Client, ClientStatus, Control, ProcessScope, Frames};

#[derive(Debug, PartialEq, Eq, Hash, Clone)]
enum NotifType {
    AudioProcessing,
    ThreadInit,
    Shutdown,
    FreewheelOn,
    FreewheelOff,
    BufferSize,
    SampleRate,
    ClientRegistration,
    GraphReorder,
    XRun,
}

use NotifType::*;

struct Notifications {
    thread: SyncSender<(NotifType, ThreadId)>
}

impl jack::NotificationHandler for Notifications {
    fn thread_init(&self, _: &Client) {
        self.thread.try_send((ThreadInit, current().id()));
    }

    fn shutdown(&mut self, _status: ClientStatus, _reason: &str) {
        self.thread.try_send((Shutdown, current().id()));
    }

    fn freewheel(&mut self, _: &Client, is_freewheel_enabled: bool) {
        self.thread.try_send((if is_freewheel_enabled {
            FreewheelOn
        } else {
            FreewheelOff
        }, current().id()));
    }

    fn buffer_size(&mut self, _: &Client, _size: Frames) -> Control {
        self.thread.try_send((BufferSize, current().id()));
        Control::Continue
    }

    fn sample_rate(&mut self, _: &Client, _size: Frames) -> Control {
        self.thread.try_send((SampleRate, current().id()));
        Control::Continue
    }

    fn client_registration(&mut self, _: &Client, _name: &str, _is_registered: bool) {
        self.thread.try_send((ClientRegistration, current().id()));
    }

    fn graph_reorder(&mut self, _: &Client) -> Control {
        self.thread.try_send((GraphReorder, current().id()));
        Control::Continue
    }

    fn xrun(&mut self, _: &Client) -> Control {
        self.thread.try_send((XRun, current().id()));
        Control::Continue
    }
}

struct Process {
    thread: SyncSender<(NotifType, ThreadId)>
}

impl jack::ProcessHandler for Process {
    fn process(&mut self, _: &Client, _: &ProcessScope) -> Control {
        self.thread.try_send((AudioProcessing, current().id()));

        Control::Continue
    }
}

fn main() {
    let interval = 60;
    let repeat = false;

    let client = jack::Client::new("Sync Test", jack::ClientOptions::NO_START_SERVER).unwrap().0;

    let (notif_tx, notif_rx) = sync_channel(5);

    let out_port = client
        .register_port("out", jack::AudioOut::default())
        .unwrap();

    let running = client.activate_async(Notifications {thread: notif_tx.clone()}, Process {thread: notif_tx}).unwrap();

    let mut notifications: HashMap<NotifType, HashSet<ThreadId>> = HashMap::new();

    let mut last_clear = Instant::now();

    loop {
        if last_clear.elapsed().as_secs() > interval {
            if repeat {
                last_clear = Instant::now();
                for (_ty, v) in notifications.iter_mut() {
                    v.clear();
                }
            }else{
                println!("");
                println!("---- FINAL SUMMARY ---");
                let mut thread_map: HashMap<ThreadId, HashSet<NotifType>> = HashMap::new();

                for (ty, threads) in notifications.iter() {
                    for thread in threads.iter() {
                        match thread_map.get_mut(thread) {
                            None => {
                                let mut set = HashSet::new();
                                set.insert(ty.clone());
                                thread_map.insert(thread.clone(), set);
                            },
                            Some(set) => {
                                set.insert(ty.clone());
                            }
                        }
                    }
                }

                for (thread, tys) in thread_map {
                    println!("{:?} => {:?}", thread, tys);
                }

                return;
            }
        }
        if let Ok((ty, thread)) = notif_rx.try_recv() {
            if !notifications.contains_key(&ty) {
                notifications.insert(ty.clone(), HashSet::new());
            }
            let set = notifications.get_mut(&ty).unwrap();
            if !set.contains(&thread) {
                println!("Got {:?} on thread {:?}", ty, thread);
                set.insert(thread);
                continue;
            }
        }
        
        sleep(Duration::from_millis(2));
    }
}