Only spawn one search queue in actix-web

This commit is contained in:
Tamo 2024-08-27 17:19:25 +02:00
parent 36d8684dc8
commit 99fdccdc7e
5 changed files with 49 additions and 10 deletions

View File

@ -13,11 +13,10 @@ pub mod search_queue;
use std::fs::File;
use std::io::{BufReader, BufWriter};
use std::num::NonZeroUsize;
use std::path::Path;
use std::str::FromStr;
use std::sync::Arc;
use std::thread::{self, available_parallelism};
use std::thread;
use std::time::Duration;
use actix_cors::Cors;
@ -118,6 +117,7 @@ pub type LogStderrType = tracing_subscriber::filter::Filtered<
pub fn create_app(
index_scheduler: Data<IndexScheduler>,
auth_controller: Data<AuthController>,
search_queue: Data<SearchQueue>,
opt: Opt,
logs: (LogRouteHandle, LogStderrHandle),
analytics: Arc<dyn Analytics>,
@ -137,6 +137,7 @@ pub fn create_app(
s,
index_scheduler.clone(),
auth_controller.clone(),
search_queue.clone(),
&opt,
logs,
analytics.clone(),
@ -469,19 +470,16 @@ pub fn configure_data(
config: &mut web::ServiceConfig,
index_scheduler: Data<IndexScheduler>,
auth: Data<AuthController>,
search_queue: Data<SearchQueue>,
opt: &Opt,
(logs_route, logs_stderr): (LogRouteHandle, LogStderrHandle),
analytics: Arc<dyn Analytics>,
) {
let search_queue = SearchQueue::new(
opt.experimental_search_queue_size,
available_parallelism().unwrap_or(NonZeroUsize::new(2).unwrap()),
);
let http_payload_size_limit = opt.http_payload_size_limit.as_u64() as usize;
config
.app_data(index_scheduler)
.app_data(auth)
.app_data(web::Data::new(search_queue))
.app_data(search_queue)
.app_data(web::Data::from(analytics))
.app_data(web::Data::new(logs_route))
.app_data(web::Data::new(logs_stderr))

View File

@ -1,8 +1,10 @@
use std::env;
use std::io::{stderr, LineWriter, Write};
use std::num::NonZeroUsize;
use std::path::PathBuf;
use std::str::FromStr;
use std::sync::Arc;
use std::thread::available_parallelism;
use actix_web::http::KeepAlive;
use actix_web::web::Data;
@ -11,6 +13,7 @@ use index_scheduler::IndexScheduler;
use is_terminal::IsTerminal;
use meilisearch::analytics::Analytics;
use meilisearch::option::LogMode;
use meilisearch::search_queue::SearchQueue;
use meilisearch::{
analytics, create_app, setup_meilisearch, LogRouteHandle, LogRouteType, LogStderrHandle,
LogStderrType, Opt, SubscriberForSecondLayer,
@ -148,11 +151,17 @@ async fn run_http(
let opt_clone = opt.clone();
let index_scheduler = Data::from(index_scheduler);
let auth_controller = Data::from(auth_controller);
let search_queue = SearchQueue::new(
opt.experimental_search_queue_size,
available_parallelism().unwrap_or(NonZeroUsize::new(2).unwrap()),
);
let search_queue = Data::new(search_queue);
let http_server = HttpServer::new(move || {
create_app(
index_scheduler.clone(),
auth_controller.clone(),
search_queue.clone(),
opt.clone(),
logs.clone(),
analytics.clone(),

View File

@ -1,7 +1,9 @@
#![allow(dead_code)]
use std::num::NonZeroUsize;
use std::path::Path;
use std::str::FromStr as _;
use std::sync::Arc;
use std::time::Duration;
use actix_http::body::MessageBody;
@ -10,6 +12,7 @@ use actix_web::http::StatusCode;
use byte_unit::{Byte, Unit};
use clap::Parser;
use meilisearch::option::{IndexerOpts, MaxMemory, MaxThreads, Opt};
use meilisearch::search_queue::SearchQueue;
use meilisearch::{analytics, create_app, setup_meilisearch, SubscriberForSecondLayer};
use once_cell::sync::Lazy;
use tempfile::TempDir;
@ -32,6 +35,12 @@ pub struct Server {
pub static TEST_TEMP_DIR: Lazy<TempDir> = Lazy::new(|| TempDir::new().unwrap());
impl Server {
fn new_search_queue(options: &Opt) -> Arc<SearchQueue> {
let search_queue =
SearchQueue::new(options.experimental_search_queue_size, NonZeroUsize::new(1).unwrap());
Arc::new(search_queue)
}
pub async fn new() -> Self {
let dir = TempDir::new().unwrap();
@ -44,7 +53,13 @@ impl Server {
let options = default_settings(dir.path());
let (index_scheduler, auth) = setup_meilisearch(&options).unwrap();
let service = Service { index_scheduler, auth, options, api_key: None };
let service = Service {
index_scheduler,
auth,
search_queue: Self::new_search_queue(&options),
options,
api_key: None,
};
Server { service, _dir: Some(dir) }
}
@ -59,7 +74,13 @@ impl Server {
options.master_key = Some("MASTER_KEY".to_string());
let (index_scheduler, auth) = setup_meilisearch(&options).unwrap();
let service = Service { index_scheduler, auth, options, api_key: None };
let service = Service {
index_scheduler,
auth,
search_queue: Self::new_search_queue(&options),
options,
api_key: None,
};
Server { service, _dir: Some(dir) }
}
@ -72,7 +93,13 @@ impl Server {
pub async fn new_with_options(options: Opt) -> Result<Self, anyhow::Error> {
let (index_scheduler, auth) = setup_meilisearch(&options)?;
let service = Service { index_scheduler, auth, options, api_key: None };
let service = Service {
index_scheduler,
auth,
search_queue: Self::new_search_queue(&options),
options,
api_key: None,
};
Ok(Server { service, _dir: None })
}
@ -100,6 +127,7 @@ impl Server {
actix_web::test::init_service(create_app(
self.service.index_scheduler.clone().into(),
self.service.auth.clone().into(),
self.service.search_queue.clone().into(),
self.service.options.clone(),
(route_layer_handle, stderr_layer_handle),
analytics::MockAnalytics::new(&self.service.options),

View File

@ -5,6 +5,7 @@ use actix_web::http::StatusCode;
use actix_web::test;
use actix_web::test::TestRequest;
use index_scheduler::IndexScheduler;
use meilisearch::search_queue::SearchQueue;
use meilisearch::{analytics, create_app, Opt, SubscriberForSecondLayer};
use meilisearch_auth::AuthController;
use tracing::level_filters::LevelFilter;
@ -16,6 +17,7 @@ use crate::common::Value;
pub struct Service {
pub index_scheduler: Arc<IndexScheduler>,
pub auth: Arc<AuthController>,
pub search_queue: Arc<SearchQueue>,
pub options: Opt,
pub api_key: Option<String>,
}
@ -123,6 +125,7 @@ impl Service {
let app = test::init_service(create_app(
self.index_scheduler.clone().into(),
self.auth.clone().into(),
self.search_queue.clone().into(),
self.options.clone(),
(route_layer_handle, stderr_layer_handle),
analytics::MockAnalytics::new(&self.options),

View File

@ -44,6 +44,7 @@ async fn basic_test_log_stream_route() {
let app = actix_web::test::init_service(create_app(
server.service.index_scheduler.clone().into(),
server.service.auth.clone().into(),
server.service.search_queue.clone().into(),
server.service.options.clone(),
(route_layer_handle, stderr_layer_handle),
analytics::MockAnalytics::new(&server.service.options),