0

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

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

对于这个内部服务器，会生成一个等待 mpsc 队列的 Tokio 任务。接收到的消息应包含 SOCKS5 地址信息和 oneshot 通道的 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 可用于在任务之间传输，而不仅仅是线程，因此包含线程不应像它那样死锁。

这里有什么问题？

4

1 回答 1

0

读过Aaron 的博客后，期货的概念现在更加清晰了。我的第一种方法不是需求驱动的，因此是不够的。函数 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)
...
于 2017-12-23T09:59:51.333 回答