rust - 使用Tokio的mpsc和oneshot会导致死锁

我想编写一个SOCKS服务器，它根据客户机请求的目的地，从多个internet网关中选择一个。一般流程是
执行SOCKS5协商并从客户端获取地址信息
请求内部服务器选择internet网关和目标的IP
连接并进行通信
对于这个内部服务器，将生成一个Tokio任务，该任务将等待mpsc队列。收到的消息应包含SOCKS5地址信息和一个单发频道的发送端，以返回结果。
另一个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可以用于任务之间的传输，而不仅仅是线程之间的传输，因此包含的线程不应该像它那样死锁。
这里怎么了？

最佳答案

在阅读了Aaron's blog之后，期货的概念现在更加清晰了。我的第一个方法不是需求驱动的，因此是不充分的。函数resolve（）实际上应该返回未来而不是结果。
为了恰当地结束这个问题，下面是我修改过的、进一步简化的最小示例，以展示这个概念：

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)
...

关于rust - 使用Tokio的mpsc和oneshot会导致死锁，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/47929646/