concurrency - 在 Rust 中实现 MultiplexStream

Question

我想用 Rust 实现多路复用器/多路分解器。它应该通过一个“下游”DuplexStream 发送多个“上游”DuplexStream 的数据，只需在上游 port_num 标识符前添加 DuplexStream 到数据包。当然这也应该反过来工作:从从下游接收的数据包中读取 port_num 并将其发送到正确的上游流。

我开始实现这样一个MultiplexStream(下面的代码不会编译)。但是，我面临一个问题:将 port_num 映射到相应上游 DuplexStream 的 open_ports 变量必须可供多个任务访问，这在 Rust 中是不允许的。

这里可以应用什么设计模式来解决我的问题？

impl MultiplexStream<T,U> {
    fn new(downstream: DuplexStream<(u32,T), U>) -> MultiplexStream<T,U> {
        let mut open_ports = HashMap::<u32, DuplexStream<(u32,T), U>>new();

        spawn do {
            let res = try {
                loop {
                    let (port_num, data) = downstream.recv();

                    match open_ports.find(port_num) {
                        Some(intermediate) => {
                            let res = try {
                                intermediate.send(data)
                            }

                            if res.is_err() {
                                open_ports.remove(port_num);
                            }
                        }
                        None => {}
                    }
                }
            }

            // downstream was closed => cleanup
            for intermediate in open_ports.values() {
                intermediate.close();
            }
            open_ports.clear();
        }

    }

    fn open<V: Send>(port_num: u32) -> Result(DuplexStream<V,T>, ()) {
        if open_ports.contains_key(port_num) {
            return Err(());
        }

        let (upstream, intermediate) = DuplexStream<V,T>::new();
        open_ports.insert(port_num, intermediate);

        spawn do {
            let res = try {
                loop {
                    let data = intermediate.recv();
                    downstream.send(~(port_num, data));
                }
            }

            // upstream was closed => cleanup
            intermediate.close();
            open_ports.remove(port_num);
        }

        return Ok(upstream);
    }
}

Answer 1

在 Rust 中，共享数据是通过 Arc(来自 libsync)完成的。 Basic Arc 用于共享不可变数据，可变数据有 MutexArc 和 RWArc。与 Arc 共享是免费的。

我举了一个小例子:

extern mod sync;

use std::hashmap::HashMap;

fn main() {
    let arc = sync::RWArc::new(HashMap::<~str, int>::new());
    arc.write(|m| m.insert(~"a", 0));
    for num in range(1, 10) {
        let arc = arc.clone();
        spawn(proc() {
            println!("[task {}] Value before is: {}", num, arc.read(|m| m.get(&~"a").clone()));
            arc.write(|m| { m.insert_or_update_with(~"a", 0, |_, val| *val += 1); });
            println!("[task {}] Value after is: {}", num, arc.read(|m| m.get(&~"a").clone()));
        });
    }
}

对于最新版本的 rust (0.10pre) 使用

extern crate collections;
extern crate sync;
use collections::hashmap::HashMap;
use sync::RWArc;

fn main() {
    let arc = RWArc::new(HashMap::<~str, int>::new());
    arc.write(|m| m.insert(~"a", 0));
    for num in range(1, 10) {
        let arc = arc.clone();
        spawn(proc() {
            println!("[task {}] Value before is: {}", num, arc.read(|m| m.get(&~"a").clone()));
            arc.write(|m| { m.insert_or_update_with(~"a", 0, |_, val| *val += 1); });
            println!("[task {}] Value after is: {}", num, arc.read(|m| m.get(&~"a").clone()));
        });
    }
}