使用Tokio的mpsc和oneshot导致死锁

Question

我想写一个SOCKS服务器，根据客户端的要求，根据目的地选择几个互联网网关中的一个。一般的流程是

1. 执行SOCKS5协商并从客户端获取地址信息
2. 请求内部服务器选择Internet网关和目标的IP
3. 连接并进行通信

对于此内部服务器，将生成一个等待mpsc队列的Tokio任务。收到的消息应包含SOCKS5地址信息和单触发通道的tx侧以返回结果。

另一个Tokio任务只是定期查询内部服务器：

extern crate futures;
extern crate tokio_core;
extern crate tokio_timer;

use std::time;
use std::time::{Duration, Instant};
use std::fmt::Debug;
use tokio_core::reactor::{Core, Interval};
use tokio_timer::wheel;
use futures::{Future, Sink, Stream};
use futures::sync::{mpsc, oneshot};

type MsgRequest<A, E> = oneshot::Sender<Result<A, E>>;
type FutRequest<A, E> = mpsc::Sender<MsgRequest<A, E>>;

#[derive(Debug)]
struct Responder<A, E> {
    fut_tx: FutRequest<A, E>,
}

impl<A: 'static, E: 'static> Responder<A, E>
where
    E: Debug,
{
    fn query(&self) -> Result<A, E> {
        println!("enter query");
        let (res_tx, res_rx) = oneshot::channel::<Result<A, E>>();
        println!("send query");
        let fut_tx = self.fut_tx.clone();
        let res = fut_tx
            .send(res_tx)
            .then(|tx| {
                if let Ok(_tx) = tx {
                    println!("Sink flushed");
                }
                res_rx
            })
            .and_then(|x| Ok(x))
            .wait()
            .unwrap();
        res
    }
}

impl<A: 'static, E: 'static> Clone for Responder<A, E> {
    fn clone(&self) -> Self {
        Responder {
            fut_tx: self.fut_tx.clone(),
        }
    }
}

fn resolve(tx: oneshot::Sender<Result<u8, String>>) -> Result<(), ()> {
    println!("resolve");
    let delay = time::Duration::from_secs(10);
    wheel()
        .build()
        .sleep(delay)
        .then(|_| tx.send(Ok(0)))
        .wait()
        .unwrap();
    println!("resolve answered");
    Ok(())
}

fn main() {
    let mut lp = Core::new().unwrap();
    let handle = lp.handle();

    let (fut_tx, fut_rx) = mpsc::channel::<MsgRequest<u8, String>>(100);
    let resolver = fut_rx.for_each(|msg| resolve(msg));
    handle.spawn(resolver);

    let responder = Responder { fut_tx };

    let server = Interval::new_at(Instant::now(), Duration::new(2, 0), &handle)
        .unwrap()
        .for_each(move |_| {
            println!("Call query for_each");
            let rx = responder.clone();
            let _res = rx.query();
            Ok(())
        })
        .map_err(|_| ());
    handle.spawn(server);

    loop {
        lp.turn(None);
    }
}

使用Cargo.toml依赖项：

[dependencies]
futures = "0.1"
tokio-core = "0.1"
tokio-timer = "0.1"

这段代码死锁。输出是：

Call query for_each
enter query
send query
Sink flushed

预期产出是：

Call query for_each
enter query
send query
Sink flushed
resolve
resolve answered
Call query for_each
enter query
send query
Sink flushed
resolve
resolve answered
....    

这表示tx结束的请求已成功发送到内部服务器，但内部服务器不处理它。根据我的理解，mpsc和oneshot可用于在任务之间进行传输，而不仅仅是线程，因此包含的线程不应该像它一样死锁。

这有什么不对？

Answer 1

在阅读了Aaron's blog之后，期货的概念现在变得更加清晰了。我的第一种方法不是需求驱动的，因而也不充分。函数resolve（）实际上应该返回future而不是结果。

为了恰当地解决这个问题，这里是我修改过的，进一步简化的最小例子来展示这个概念：

extern crate futures;
extern crate tokio_core;
extern crate tokio_timer;

use std::time;
use std::time::{Instant,Duration};
use tokio_core::reactor::{Core, Interval};
use tokio_timer::wheel;
use futures::{Future,Stream,Sink};
use futures::sync::{oneshot,mpsc};

type MsgRequest<A,E> = oneshot::Sender<Result<A,E>>;

fn main() {
    let mut lp = Core::new().unwrap();
    let handle = lp.handle();

    let (fut_tx, fut_rx) = mpsc::channel::<MsgRequest<u8,String>>(100);
    let handle2 = handle.clone();
    let resolver = fut_rx.and_then(move |tx| {
            println!("Got query...wait a bit");
            let delay = time::Duration::from_secs(5);
            handle2.spawn({
                wheel().build().sleep(delay)
                    .then(move |_|{
                        println!("Answer query");
                        tx.send(Ok(0)).unwrap();
                        println!("query answered");
                        Ok(())
                })
            });
            Ok(())
        })
        .for_each(|_| {Ok(())});
    handle.spawn(resolver);

    let server = Interval::new_at(Instant::now(),
                                Duration::new(2,0),&handle).unwrap()
            .then(move |_| {
                let fut_tx = fut_tx.clone();
                let (res_tx, res_rx) = oneshot::channel::<Result<u8,String>>();
                println!("send query");
                fut_tx.send(res_tx)
                    .then( |tx|{
                        if let Ok(_tx) = tx { println!("Sink flushed"); }
                        res_rx
                    })
            })
            .for_each(|res| {
                println!("Received result {:?}",res);
                Ok(())
            }).map_err(|_| ());
    handle.spawn(server);

    loop {
        lp.turn(None);
    }
}

它按预期输出：

send query
Sink flushed
Got query...wait a bit
Answer query
query answered
Received result Ok(0)
send query
Sink flushed
Got query...wait a bit
Answer query
query answered
Received result Ok(0)
...