Почему чтение из потока Rusoto S3 в веб-обработчике Actix вызывает тупик? - PullRequest
Новая
       36

Почему чтение из потока Rusoto S3 в веб-обработчике Actix вызывает тупик?

1 голос
/ 02 июля 2019

Я пишу приложение, используя actix_web и rusoto_s3.

Когда я запускаю команду вне запроса actix непосредственно из main, она работает нормально, а get_object работаеткак и ожидалось.Когда это инкапсулировано в запросе actix_web, поток блокируется навсегда.

У меня есть клиент, который совместно используется для всех запросов, который инкапсулирован в Arc (это происходит во внутренних данных actix).

Полный код: 

fn index(
    _req: HttpRequest,
    path: web::Path<String>,
    s3: web::Data<S3Client>,
) -> impl Future<Item = HttpResponse, Error = actix_web::Error> {
    s3.get_object(GetObjectRequest {
        bucket: "my_bucket".to_owned(),
        key: path.to_owned(),
        ..Default::default()
    })
    .and_then(move |res| {
        info!("Response {:?}", res);
        let mut stream = res.body.unwrap().into_blocking_read();
        let mut body = Vec::new();
        stream.read_to_end(&mut body).unwrap();
        match process_file(body.as_slice()) {
            Ok(result) => Ok(result),
            Err(error) => Err(RusotoError::from(error)),
        }
    })
    .map_err(|e| match e {
        RusotoError::Service(GetObjectError::NoSuchKey(key)) => {
            actix_web::error::ErrorNotFound(format!("{} not found", key))
        }
        error => {
            error!("Error: {:?}", error);
            actix_web::error::ErrorInternalServerError("error")
        }
    })
    .from_err()
    .and_then(move |img| HttpResponse::Ok().body(Body::from(img)))
}

fn health() -> HttpResponse {
    HttpResponse::Ok().finish()
}

fn main() -> std::io::Result<()> {
    let name = "rust_s3_test";
    env::set_var("RUST_LOG", "debug");
    pretty_env_logger::init();
    let sys = actix_rt::System::builder().stop_on_panic(true).build();
    let prometheus = PrometheusMetrics::new(name, "/metrics");
    let s3 = S3Client::new(Region::Custom {
        name: "eu-west-1".to_owned(),
        endpoint: "http://localhost:9000".to_owned(),
    });
    let s3_client_data = web::Data::new(s3);

    Server::build()
        .bind(name, "0.0.0.0:8080", move || {
            HttpService::build().keep_alive(KeepAlive::Os).h1(App::new()
                .register_data(s3_client_data.clone())
                .wrap(prometheus.clone())
                .wrap(actix_web::middleware::Logger::default())
                .service(web::resource("/health").route(web::get().to(health)))
                .service(web::resource("/{file_name}").route(web::get().to_async(index))))
        })?
        .start();
    sys.run()
}

В stream.read_to_end поток блокируется и никогда не разрешается.

Я пытался клонировать клиента для каждого запроса, а также создавать нового клиента для каждого запроса, но у меня был одинаковый результат во всех сценариях.

Я что-то не так делаю?

Это работает, если я не использую его асинхронно ... 

s3.get_object(GetObjectRequest {
    bucket: "my_bucket".to_owned(),
    key: path.to_owned(),
    ..Default::default()
})
.sync()
.unwrap()
.body
.unwrap()
.into_blocking_read();
let mut body = Vec::new();
io::copy(&mut stream, &mut body);

Это проблема с Токио?

1 Ответ

2 голосов
/ 03 июля 2019 
let mut stream = res.body.unwrap().into_blocking_read();

Проверьте реализацию into_blocking_read(): она вызывает .wait(). Вы не должны вызывать код блокировки внутри Future.

Так как body в Rusoto является Stream, существует способ прочитать его асинхронно: 

.and_then(move |res| {
    info!("Response {:?}", res);
    let stream = res.body.unwrap();

    stream.concat2().map(move |file| {
        process_file(&file[..]).unwrap()
    })
    .map_err(|e| RusotoError::from(e)))
})

process_file не должен блокировать ограждение Future. Если ему нужно заблокировать, вы можете запустить его в новом потоке или инкапсулировать с tokio_threadpool's blocking.

Примечание: Вы можете использовать tokio_threadpool blocking в своей реализации, но я рекомендую вам понять, как это работает в первую очередь.

Если вы не хотите загружать весь файл в память, вы можете использовать for_each: 

stream.for_each(|part| {
    //process each part in here 
    //Warning! Do not add blocking code here either.
})

См. Также :

...