Как мне реализовать механизм блокировки очереди с помощью futures :: sync :: mpsc :: channel? - PullRequest
Новая
       18

Как мне реализовать механизм блокировки очереди с помощью futures :: sync :: mpsc :: channel?

0 голосов
/ 12 ноября 2018

Я пытаюсь понять, как работает futures::sync::mpsc::Receiver.В приведенном ниже примере поток получателя спит в течение двух секунд, а отправитель отправляет каждую секунду.

Я ожидаю, что отправитель должен быть заблокирован из-за ожидания и отправки после освобождения буфера.

Вместо этого я вижу, что через некоторое время он заблокирован.Увеличение буфера канала только увеличивает время, пока он не будет заблокирован.

Что я должен сделать, чтобы отправитель отправлял данные, когда буфер был доступен, и оказывал некоторое противодавление на отправителя в таких случаях?futures::sync::mpsc::channel имеет собственную документацию, но я не понимаю, как правильно ее использовать. 

extern crate futures;
extern crate tokio_core;

use std::{thread, time};

use futures::sync::mpsc;
use futures::{Future, Sink, Stream};

use tokio_core::reactor::Core;

#[derive(Debug)]
struct Stats {
    pub success: usize,
    pub failure: usize,
}

fn main() {
    let mut core = Core::new().expect("Failed to create core");
    let remote = core.remote();

    let (tx, rx) = mpsc::channel(1);

    thread::spawn(move || loop {
        let tx = tx.clone();

        let delay = time::Duration::from_secs(1);
        thread::sleep(delay);
        let f = ::futures::done::<(), ()>(Ok(()));

        remote.spawn(|_| {
            f.then(|res| {
                println!("Sending");
                tx.send(res).wait();
                println!("Sent");
                Ok(())
            })
        });
    });

    let mut stats = Stats {
        success: 0,
        failure: 0,
    };

    let f2 = rx.for_each(|res| {
        println!("Received");
        let delay = time::Duration::from_secs(2);
        thread::sleep(delay);

        match res {
            Ok(_) => stats.success += 1,
            Err(_) => stats.failure += 1,
        }
        println!("stats = {:?}", stats);

        Ok(())
    });

    core.run(f2).expect("Core failed to run");
}

1 Ответ

0 голосов
/ 13 ноября 2018

  1. Никогда не звоните wait внутри будущего.Это блокировка, и блокировка никогда не должна выполняться внутри будущего.

  2. Никогда не вызывайте sleep внутри будущего.Это блокировка, и блокирование никогда не должно выполняться внутри будущего.

  3. Противодавление канала реализуется тем фактом, что send потребляет Sender и возвращаетбудущее.Будущее возвращает Sender вам , когда в очереди есть место . 

extern crate futures; // 0.1.25
extern crate tokio; // 0.1.11

use futures::{future, sync::mpsc, Future, Sink, Stream};
use std::time::Duration;
use tokio::timer::Interval;

#[derive(Debug)]
struct Stats {
    pub success: usize,
    pub failure: usize,
}

fn main() {
    tokio::run(future::lazy(|| {
        let (tx, rx) = mpsc::channel::<Result<(), ()>>(1);

        tokio::spawn({
            Interval::new_interval(Duration::from_millis(10))
                .map_err(|e| panic!("Interval error: {}", e))
                .fold(tx, |tx, _| {
                    tx.send(Ok(())).map_err(|e| panic!("Send error: {}", e))
                })
                .map(drop) // discard the tx
        });

        let mut stats = Stats {
            success: 0,
            failure: 0,
        };

        let i = Interval::new_interval(Duration::from_millis(20))
            .map_err(|e| panic!("Interval error: {}", e));

        rx.zip(i).for_each(move |(res, _)| {
            println!("Received");
            match res {
                Ok(_) => stats.success += 1,
                Err(_) => stats.failure += 1,
            }
            println!("stats = {:?}", stats);

            Ok(())
        })
    }));
}
...