use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::Duration;
use rand::Rng;

// These structs encapsulate requests and provide a way for the
// state-owning thread to respond.
struct ReadOp {
    key: i32,
    resp: crossbeam::channel::Sender<i32>,
}

struct WriteOp {
    key: i32,
    val: i32,
    resp: crossbeam::channel::Sender<bool>,
}

fn main() {
    // We'll count how many operations we perform.
    let read_ops = Arc::new(AtomicU64::new(0));
    let write_ops = Arc::new(AtomicU64::new(0));

    // The reads and writes channels will be used by other threads to issue
    // read and write requests, respectively.
    let (reads_tx, reads_rx) = crossbeam::channel::unbounded();
    let (writes_tx, writes_rx) = crossbeam::channel::unbounded();

    // This thread owns the state, which is a HashMap in this case.
    let state_thread = thread::spawn(move || {
        let mut state = std::collections::HashMap::new();
        loop {
            crossbeam::select! {
                recv(reads_rx) -> read => {
                    if let Ok(read) = read {
                        let value = state.get(&read.key).cloned().unwrap_or(0);
                        read.resp.send(value).unwrap();
                    }
                }
                recv(writes_rx) -> write => {
                    if let Ok(write) = write {
                        state.insert(write.key, write.val);
                        write.resp.send(true).unwrap();
                    }
                }
            }
        }
    });

    // Start 100 threads to issue reads to the state-owning thread.
    let mut read_threads = vec![];
    for _ in 0..100 {
        let reads = reads_tx.clone();
        let read_ops = Arc::clone(&read_ops);
        let t = thread::spawn(move || {
            let mut rng = rand::thread_rng();
            loop {
                let (resp_tx, resp_rx) = crossbeam::channel::bounded(1);
                let read = ReadOp {
                    key: rng.gen_range(0..5),
                    resp: resp_tx,
                };
                reads.send(read).unwrap();
                resp_rx.recv().unwrap();
                read_ops.fetch_add(1, Ordering::SeqCst);
                thread::sleep(Duration::from_millis(1));
            }
        });
        read_threads.push(t);
    }

    // Start 10 threads to issue writes.
    let mut write_threads = vec![];
    for _ in 0..10 {
        let writes = writes_tx.clone();
        let write_ops = Arc::clone(&write_ops);
        let t = thread::spawn(move || {
            let mut rng = rand::thread_rng();
            loop {
                let (resp_tx, resp_rx) = crossbeam::channel::bounded(1);
                let write = WriteOp {
                    key: rng.gen_range(0..5),
                    val: rng.gen_range(0..100),
                    resp: resp_tx,
                };
                writes.send(write).unwrap();
                resp_rx.recv().unwrap();
                write_ops.fetch_add(1, Ordering::SeqCst);
                thread::sleep(Duration::from_millis(1));
            }
        });
        write_threads.push(t);
    }

    // Let the threads work for a second.
    thread::sleep(Duration::from_secs(1));

    // Capture and report the op counts.
    println!("readOps: {}", read_ops.load(Ordering::SeqCst));
    println!("writeOps: {}", write_ops.load(Ordering::SeqCst));
}