Почему моя реализация Future блокируется после однократного опроса и NotReady? - PullRequest
Новая
       15

Почему моя реализация Future блокируется после однократного опроса и NotReady?

0 голосов
/ 14 октября 2019

Я реализовал будущее и сделал запрос на него, но он заблокировал мой curl, и журнал показывает, что poll был вызван только один раз.

Я что-то реализовал неправильно? 

use failure::{format_err, Error};
use futures::{future, Async};
use hyper::rt::Future;
use hyper::service::{service_fn, service_fn_ok};
use hyper::{Body, Method, Request, Response, Server, StatusCode};
use log::{debug, error, info};
use std::{
    sync::{Arc, Mutex},
    task::Waker,
    thread,
};

pub struct TimerFuture {
    shared_state: Arc<Mutex<SharedState>>,
}

struct SharedState {
    completed: bool,
    resp: String,
}

impl Future for TimerFuture {
    type Item = Response<Body>;
    type Error = hyper::Error;
    fn poll(&mut self) -> futures::Poll<Response<Body>, hyper::Error> {
        let mut shared_state = self.shared_state.lock().unwrap();
        if shared_state.completed {
            return Ok(Async::Ready(Response::new(Body::from(
                shared_state.resp.clone(),
            ))));
        } else {
            return Ok(Async::NotReady);
        }
    }
}

impl TimerFuture {
    pub fn new(instance: String) -> Self {
        let shared_state = Arc::new(Mutex::new(SharedState {
            completed: false,
            resp: String::new(),
        }));
        let thread_shared_state = shared_state.clone();
        thread::spawn(move || {
            let res = match request_health(instance) {
                Ok(status) => status.clone(),
                Err(err) => {
                    error!("{:?}", err);
                    format!("{}", err)
                }
            };
            let mut shared_state = thread_shared_state.lock().unwrap();
            shared_state.completed = true;
            shared_state.resp = res;
        });

        TimerFuture { shared_state }
    }
}

fn request_health(instance_name: String) -> Result<String, Error> {
    std::thread::sleep(std::time::Duration::from_secs(1));
    Ok("health".to_string())
}

type BoxFut = Box<dyn Future<Item = Response<Body>, Error = hyper::Error> + Send>;
fn serve_health(req: Request<Body>) -> BoxFut {
    let mut response = Response::new(Body::empty());
    let path = req.uri().path().to_owned();
    match (req.method(), path) {
        (&Method::GET, path) => {
            return Box::new(TimerFuture::new(path.clone()));
        }
        _ => *response.status_mut() = StatusCode::NOT_FOUND,
    }
    Box::new(future::ok(response))
}

fn main() {
    let endpoint_addr = "0.0.0.0:8080";
    match std::thread::spawn(move || {
        let addr = endpoint_addr.parse().unwrap();
        info!("Server is running on {}", addr);
        hyper::rt::run(
            Server::bind(&addr)
                .serve(move || service_fn(serve_health))
                .map_err(|e| eprintln!("server error: {}", e)),
        );
    })
    .join()
    {
        Ok(e) => e,
        Err(e) => println!("{:?}", e),
    }
}

После компиляции и запуска этого кода работает сервер с портом 8080. Позвоните на сервер с curl, и он заблокирует: 

curl 127.0.0.1:8080/my-health-scope

1 Ответ

0 голосов
/ 14 октября 2019

Я что-то не так реализовал?

Да, вы не читали и не следовали документации по методу, который вы реализуете (выделено мной):

Когда будущее еще не готово, будет возвращено значение Async::NotReady. В этой ситуации будущее также зарегистрирует заинтересованность текущей задачи в производимой стоимости. Это делается с помощью вызова task::park для извлечения дескриптора текущего Task. Когда будущее готово к прогрессу (например, оно должно быть снова опрошено), метод unpark вызывается для Task.

As минимальный, воспроизводимый пример , давайте использовать это: 

use futures::{future::Future, Async};
use std::{
    mem,
    sync::{Arc, Mutex},
    thread,
    time::Duration,
};

pub struct Timer {
    data: Arc<Mutex<String>>,
}

impl Timer {
    pub fn new(instance: String) -> Self {
        let data = Arc::new(Mutex::new(String::new()));

        thread::spawn({
            let data = data.clone();
            move || {
                thread::sleep(Duration::from_secs(1));
                *data.lock().unwrap() = instance;
            }
        });

        Timer { data }
    }
}

impl Future for Timer {
    type Item = String;
    type Error = ();

    fn poll(&mut self) -> futures::Poll<Self::Item, Self::Error> {
        let mut data = self.data.lock().unwrap();

        eprintln!("poll was called");

        if data.is_empty() {
            Ok(Async::NotReady)
        } else {
            let data = mem::replace(&mut *data, String::new());
            Ok(Async::Ready(data))
        }
    }
}

fn main() {
    let v = Timer::new("Some text".into()).wait();
    println!("{:?}", v);
}

Он выводит только один раз "опрос был вызван".

Вы можете позвонить task::current (ранее task::park)при реализации Future::poll сохраните полученное значение, а затем используйте значение с Task::notify (ранее Task::unpark) всякий раз, когда в будущем может быть проведен повторный опрос: 

use futures::{
    future::Future,
    task::{self, Task},
    Async,
};
use std::{
    mem,
    sync::{Arc, Mutex},
    thread,
    time::Duration,
};

pub struct Timer {
    data: Arc<Mutex<(String, Option<Task>)>>,
}

impl Timer {
    pub fn new(instance: String) -> Self {
        let data = Arc::new(Mutex::new((String::new(), None)));
        let me = Timer { data };

        thread::spawn({
            let data = me.data.clone();
            move || {
                thread::sleep(Duration::from_secs(1));
                let mut data = data.lock().unwrap();

                data.0 = instance;
                if let Some(task) = data.1.take() {
                    task.notify();
                }
            }
        });

        me
    }
}

impl Future for Timer {
    type Item = String;
    type Error = ();

    fn poll(&mut self) -> futures::Poll<Self::Item, Self::Error> {
        let mut data = self.data.lock().unwrap();

        eprintln!("poll was called");

        if data.0.is_empty() {
            let v = task::current();
            data.1 = Some(v);
            Ok(Async::NotReady)
        } else {
            let data = mem::replace(&mut data.0, String::new());
            Ok(Async::Ready(data))
        }
    }
}

fn main() {
    let v = Timer::new("Some text".into()).wait();
    println!("{:?}", v);
}

См. также:

...