Блог

У меня есть случай, когда у меня есть http-сервер, axum который получает полезную нагрузку с очень высокой пропускной способностью (может достигать 20 м в секунду). Мне нужно взять эти байты из запроса и выполнить с ними некоторые сложные вычисления. Проблема в том, что объем памяти непредсказуемо высок (может достигать 5 ГБ). Это текущая настройка того, как я пытаюсь ее достичь:

 #[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (tx, mut rx) = mpsc::channel::<WriteRequest>(32);

    tokio::spawn(async move {
      while let Some(payload) = rx.recv().await {
        tokio::task::spawn_blocking(move || {
          // Run heavy computation here...
          heavy_computation(payload)
        }).await;
      }
    });

    // build our application with a route
    let app = Router::new()
        .route("/write", post(move |req: Bytes| async move {
           let data: WriteRequest = Message::decode(req);
           // send data here
           let _ = tx.send(data).await;

           "ok"
         }));

    let addr = ([0, 0, 0, 0], 8080).into();
    let server = axum::Server::bind(amp;addr)
        .serve(app.into_make_service());
    
    if let Err(e) = server.await {
        error!("server error: {}", e);
    }
    
    Ok(())
}
 

Я думаю, что это обратное давление на bounded channel то, что заставляет запросы накапливаться до тех пор, пока они не будут отправлены другому task для обработки, что приводит к увеличению объема памяти.
Потому что даже если бы я попытался заменить heavy_copmutation это на простое sleep , примерно 200ms это закончилось бы теми же результатами.
Если я исключу эту heavy_computation часть, память останется низкой.

Как правильно подойти к такой проблеме? или с такой высокой пропускной способностью здесь ничего нельзя сделать?

Большое спасибо!

Такое ощущение, что пока вычисление heavy_computation занято, накапливаются миллионы ожидающих запросов. Необходимо ограничить количество принятых подключений/запросов, обрабатываемых одновременно. Чтобы использовать 5 ГБ, требуется всего 25 тыс. ожидающих запросов с полезной нагрузкой 200 Кб, а не миллионы.

аксум основан на башне, а башня основана на гипере.

Известно, что у hyper нет параметра max connections, но люди там предлагают использовать промежуточное программное обеспечение ConcurrencyLimit из tower и настроить его на сервере или создать пользовательский цикл приема/обработки.

Возможно, можно было бы передать это промежуточное программное обеспечение в tower и через axum, но в противном случае, если это вариант для вас, вы могли бы попробовать перейти непосредственно в tower или даже использовать hyper и реализовать его с использованием примитивов, доступных там для этой рабочей нагрузки.