Do not create too many rayon tasks

Cargo.lock

@@ -496,7 +496,7 @@ source = "git+https://github.com/meilisearch/bbqueue#cbb87cc707b5af415ef203bdaf2
 
 [[package]]
 name = "benchmarks"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "anyhow",
  "bumpalo",
@@ -689,7 +689,7 @@ dependencies = [
 
 [[package]]
 name = "build-info"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "anyhow",
  "time",
@@ -1664,7 +1664,7 @@ dependencies = [
 
 [[package]]
 name = "dump"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "anyhow",
  "big_s",
@@ -1876,7 +1876,7 @@ checksum = "486f806e73c5707928240ddc295403b1b93c96a02038563881c4a2fd84b81ac4"
 
 [[package]]
 name = "file-store"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "tempfile",
  "thiserror",
@@ -1898,7 +1898,7 @@ dependencies = [
 
 [[package]]
 name = "filter-parser"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "insta",
  "nom",
@@ -1918,7 +1918,7 @@ dependencies = [
 
 [[package]]
 name = "flatten-serde-json"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "criterion",
  "serde_json",
@@ -2057,7 +2057,7 @@ dependencies = [
 
 [[package]]
 name = "fuzzers"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "arbitrary",
  "bumpalo",
@@ -2624,7 +2624,7 @@ checksum = "206ca75c9c03ba3d4ace2460e57b189f39f43de612c2f85836e65c929701bb2d"
 
 [[package]]
 name = "index-scheduler"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "anyhow",
  "arroy 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -2822,7 +2822,7 @@ dependencies = [
 
 [[package]]
 name = "json-depth-checker"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "criterion",
  "serde_json",
@@ -3441,7 +3441,7 @@ checksum = "490cc448043f947bae3cbee9c203358d62dbee0db12107a74be5c30ccfd09771"
 
 [[package]]
 name = "meili-snap"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "insta",
  "md5",
@@ -3450,7 +3450,7 @@ dependencies = [
 
 [[package]]
 name = "meilisearch"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "actix-cors",
  "actix-http",
@@ -3540,7 +3540,7 @@ dependencies = [
 
 [[package]]
 name = "meilisearch-auth"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "base64 0.22.1",
  "enum-iterator",
@@ -3559,7 +3559,7 @@ dependencies = [
 
 [[package]]
 name = "meilisearch-types"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "actix-web",
  "anyhow",
@@ -3592,7 +3592,7 @@ dependencies = [
 
 [[package]]
 name = "meilitool"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "anyhow",
  "arroy 0.5.0 (git+https://github.com/meilisearch/arroy/?tag=DO-NOT-DELETE-upgrade-v04-to-v05)",
@@ -3627,7 +3627,7 @@ dependencies = [
 
 [[package]]
 name = "milli"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "allocator-api2",
  "arroy 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -4083,7 +4083,7 @@ checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
 
 [[package]]
 name = "permissive-json-pointer"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "big_s",
  "serde_json",
@@ -6486,7 +6486,7 @@ dependencies = [
 
 [[package]]
 name = "xtask"
-version = "1.12.1"
+version = "1.12.0"
 dependencies = [
  "anyhow",
  "build-info",

Cargo.toml

@@ -22,7 +22,7 @@ members = [
 ]
 
 [workspace.package]
-version = "1.12.1"
+version = "1.12.0"
 authors = [
     "Quentin de Quelen <quentin@dequelen.me>",
     "Clément Renault <clement@meilisearch.com>",

crates/index-scheduler/src/batch.rs

@@ -1312,7 +1312,9 @@ impl IndexScheduler {
                     if let DocumentOperation::Add(content_uuid) = operation {
                         let content_file = self.file_store.get_update(*content_uuid)?;
                         let mmap = unsafe { memmap2::Mmap::map(&content_file)? };
-                        content_files.push(mmap);
+                        if !mmap.is_empty() {
+                            content_files.push(mmap);
+                        }
                     }
                 }
 

crates/index-scheduler/src/index_mapper/index_map.rs

@@ -1,4 +1,5 @@
 use std::collections::BTreeMap;
+use std::env::VarError;
 use std::path::Path;
 use std::time::Duration;
 
@@ -300,9 +301,19 @@ fn create_or_open_index(
     enable_mdb_writemap: bool,
     map_size: usize,
 ) -> Result<Index> {
+    use std::str::FromStr;
+
     let mut options = EnvOpenOptions::new();
     options.map_size(clamp_to_page_size(map_size));
-    options.max_readers(1024);
+
+    let max_readers = match std::env::var("MEILI_INDEX_MAX_READERS") {
+        Ok(value) => u32::from_str(&value).unwrap(),
+        Err(VarError::NotPresent) => 100 * 1024,
+        Err(VarError::NotUnicode(value)) => {
+            panic!("Invalid unicode for the `MEILI_INDEX_MAX_READERS` env var: {value:?}")
+        }
+    };
+    options.max_readers(max_readers);
     if enable_mdb_writemap {
         unsafe { options.flags(EnvFlags::WRITE_MAP) };
     }

crates/index-scheduler/src/utils.rs

@@ -291,10 +291,7 @@ impl IndexScheduler {
 
         debug_assert!(old_task != *task);
         debug_assert_eq!(old_task.uid, task.uid);
-        debug_assert!(
-            old_task.batch_uid.is_none() && task.batch_uid.is_some(),
-            "\n==> old: {old_task:?}\n==> new: {task:?}"
-        );
+        debug_assert!(old_task.batch_uid.is_none() && task.batch_uid.is_some());
 
         if old_task.status != task.status {
             self.update_status(wtxn, old_task.status, |bitmap| {

crates/meilisearch/tests/documents/add_documents.rs

@@ -1220,89 +1220,9 @@ async fn replace_document() {
 #[actix_rt::test]
 async fn add_no_documents() {
     let server = Server::new().await;
-    let index = server.index("kefir");
-    let (task, code) = index.add_documents(json!([]), None).await;
+    let index = server.index("test");
+    let (_response, code) = index.add_documents(json!([]), None).await;
     snapshot!(code, @"202 Accepted");
-    let task = server.wait_task(task.uid()).await;
-    let task = task.succeeded();
-    snapshot!(task, @r#"
-    {
-      "uid": "[uid]",
-      "batchUid": "[batch_uid]",
-      "indexUid": "kefir",
-      "status": "succeeded",
-      "type": "documentAdditionOrUpdate",
-      "canceledBy": null,
-      "details": {
-        "receivedDocuments": 0,
-        "indexedDocuments": 0
-      },
-      "error": null,
-      "duration": "[duration]",
-      "enqueuedAt": "[date]",
-      "startedAt": "[date]",
-      "finishedAt": "[date]"
-    }
-    "#);
-
-    let (task, _code) = index.add_documents(json!([]), Some("kefkef")).await;
-    let task = server.wait_task(task.uid()).await;
-    let task = task.succeeded();
-    snapshot!(task, @r#"
-    {
-      "uid": "[uid]",
-      "batchUid": "[batch_uid]",
-      "indexUid": "kefir",
-      "status": "succeeded",
-      "type": "documentAdditionOrUpdate",
-      "canceledBy": null,
-      "details": {
-        "receivedDocuments": 0,
-        "indexedDocuments": 0
-      },
-      "error": null,
-      "duration": "[duration]",
-      "enqueuedAt": "[date]",
-      "startedAt": "[date]",
-      "finishedAt": "[date]"
-    }
-    "#);
-
-    let (task, _code) = index.add_documents(json!([{ "kefkef": 1 }]), None).await;
-    let task = server.wait_task(task.uid()).await;
-    let task = task.succeeded();
-    snapshot!(task, @r#"
-    {
-      "uid": "[uid]",
-      "batchUid": "[batch_uid]",
-      "indexUid": "kefir",
-      "status": "succeeded",
-      "type": "documentAdditionOrUpdate",
-      "canceledBy": null,
-      "details": {
-        "receivedDocuments": 1,
-        "indexedDocuments": 1
-      },
-      "error": null,
-      "duration": "[duration]",
-      "enqueuedAt": "[date]",
-      "startedAt": "[date]",
-      "finishedAt": "[date]"
-    }
-    "#);
-    let (documents, _status) = index.get_all_documents(GetAllDocumentsOptions::default()).await;
-    snapshot!(documents, @r#"
-    {
-      "results": [
-        {
-          "kefkef": 1
-        }
-      ],
-      "offset": 0,
-      "limit": 20,
-      "total": 1
-    }
-    "#);
 }
 
 #[actix_rt::test]

crates/milli/src/thread_pool_no_abort.rs

@@ -1,4 +1,4 @@
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
 
 use rayon::{ThreadPool, ThreadPoolBuilder};
@@ -9,6 +9,8 @@ use thiserror::Error;
 #[derive(Debug)]
 pub struct ThreadPoolNoAbort {
     thread_pool: ThreadPool,
+    /// The number of active operations.
+    active_operations: AtomicUsize,
     /// Set to true if the thread pool catched a panic.
     pool_catched_panic: Arc<AtomicBool>,
 }
@@ -19,7 +21,9 @@ impl ThreadPoolNoAbort {
         OP: FnOnce() -> R + Send,
         R: Send,
     {
+        self.active_operations.fetch_add(1, Ordering::Relaxed);
         let output = self.thread_pool.install(op);
+        self.active_operations.fetch_sub(1, Ordering::Relaxed);
         // While reseting the pool panic catcher we return an error if we catched one.
         if self.pool_catched_panic.swap(false, Ordering::SeqCst) {
             Err(PanicCatched)
@@ -31,6 +35,11 @@ impl ThreadPoolNoAbort {
     pub fn current_num_threads(&self) -> usize {
         self.thread_pool.current_num_threads()
     }
+
+    /// The number of active operations.
+    pub fn active_operations(&self) -> usize {
+        self.active_operations.load(Ordering::Relaxed)
+    }
 }
 
 #[derive(Error, Debug)]
@@ -64,6 +73,10 @@ impl ThreadPoolNoAbortBuilder {
             let catched_panic = pool_catched_panic.clone();
             move |_result| catched_panic.store(true, Ordering::SeqCst)
         });
-        Ok(ThreadPoolNoAbort { thread_pool: self.0.build()?, pool_catched_panic })
+        Ok(ThreadPoolNoAbort {
+            thread_pool: self.0.build()?,
+            active_operations: AtomicUsize::new(0),
+            pool_catched_panic,
+        })
     }
 }

crates/milli/src/update/new/indexer/document_operation.rs

@@ -252,24 +252,6 @@ fn extract_addition_payload_changes<'r, 'pl: 'r>(
         previous_offset = iter.byte_offset();
     }
 
-    if payload.is_empty() {
-        let result = retrieve_or_guess_primary_key(
-            rtxn,
-            index,
-            new_fields_ids_map,
-            primary_key_from_op,
-            None,
-        );
-        match result {
-            Ok(Ok((pk, _))) => {
-                primary_key.get_or_insert(pk);
-            }
-            Ok(Err(UserError::NoPrimaryKeyCandidateFound)) => (),
-            Ok(Err(user_error)) => return Err(Error::UserError(user_error)),
-            Err(error) => return Err(error),
-        };
-    }
-
     Ok(new_docids_version_offsets)
 }
 

crates/milli/src/vector/mod.rs

@@ -637,6 +637,7 @@ impl Embedder {
         }
     }
 
+    #[tracing::instrument(level = "debug", skip_all, target = "indexing::vector")]
     pub fn embed_chunks_ref(
         &self,
         texts: &[&str],

crates/milli/src/vector/ollama.rs

@@ -5,7 +5,7 @@ use rayon::slice::ParallelSlice as _;
 
 use super::error::{EmbedError, EmbedErrorKind, NewEmbedderError, NewEmbedderErrorKind};
 use super::rest::{Embedder as RestEmbedder, EmbedderOptions as RestEmbedderOptions};
-use super::DistributionShift;
+use super::{DistributionShift, REQUEST_PARALLELISM};
 use crate::error::FaultSource;
 use crate::vector::Embedding;
 use crate::ThreadPoolNoAbort;
@@ -113,20 +113,30 @@ impl Embedder {
         texts: &[&str],
         threads: &ThreadPoolNoAbort,
     ) -> Result<Vec<Vec<f32>>, EmbedError> {
-        threads
-            .install(move || {
-                let embeddings: Result<Vec<Vec<Embedding>>, _> = texts
-                    .par_chunks(self.prompt_count_in_chunk_hint())
-                    .map(move |chunk| self.embed(chunk, None))
-                    .collect();
+        if threads.active_operations() >= REQUEST_PARALLELISM {
+            let embeddings: Result<Vec<Vec<Embedding>>, _> = texts
+                .chunks(self.prompt_count_in_chunk_hint())
+                .map(move |chunk| self.embed(chunk, None))
+                .collect();
 
-                let embeddings = embeddings?;
-                Ok(embeddings.into_iter().flatten().collect())
-            })
-            .map_err(|error| EmbedError {
-                kind: EmbedErrorKind::PanicInThreadPool(error),
-                fault: FaultSource::Bug,
-            })?
+            let embeddings = embeddings?;
+            Ok(embeddings.into_iter().flatten().collect())
+        } else {
+            threads
+                .install(move || {
+                    let embeddings: Result<Vec<Vec<Embedding>>, _> = texts
+                        .par_chunks(self.prompt_count_in_chunk_hint())
+                        .map(move |chunk| self.embed(chunk, None))
+                        .collect();
+
+                    let embeddings = embeddings?;
+                    Ok(embeddings.into_iter().flatten().collect())
+                })
+                .map_err(|error| EmbedError {
+                    kind: EmbedErrorKind::PanicInThreadPool(error),
+                    fault: FaultSource::Bug,
+                })?
+        }
     }
 
     pub fn chunk_count_hint(&self) -> usize {

crates/milli/src/vector/openai.rs

@@ -6,7 +6,7 @@ use rayon::slice::ParallelSlice as _;
 
 use super::error::{EmbedError, NewEmbedderError};
 use super::rest::{Embedder as RestEmbedder, EmbedderOptions as RestEmbedderOptions};
-use super::DistributionShift;
+use super::{DistributionShift, REQUEST_PARALLELISM};
 use crate::error::FaultSource;
 use crate::vector::error::EmbedErrorKind;
 use crate::vector::Embedding;
@@ -270,20 +270,29 @@ impl Embedder {
         texts: &[&str],
         threads: &ThreadPoolNoAbort,
     ) -> Result<Vec<Vec<f32>>, EmbedError> {
-        threads
-            .install(move || {
-                let embeddings: Result<Vec<Vec<Embedding>>, _> = texts
-                    .par_chunks(self.prompt_count_in_chunk_hint())
-                    .map(move |chunk| self.embed(chunk, None))
-                    .collect();
+        if threads.active_operations() >= REQUEST_PARALLELISM {
+            let embeddings: Result<Vec<Vec<Embedding>>, _> = texts
+                .chunks(self.prompt_count_in_chunk_hint())
+                .map(move |chunk| self.embed(chunk, None))
+                .collect();
+            let embeddings = embeddings?;
+            Ok(embeddings.into_iter().flatten().collect())
+        } else {
+            threads
+                .install(move || {
+                    let embeddings: Result<Vec<Vec<Embedding>>, _> = texts
+                        .par_chunks(self.prompt_count_in_chunk_hint())
+                        .map(move |chunk| self.embed(chunk, None))
+                        .collect();
 
-                let embeddings = embeddings?;
-                Ok(embeddings.into_iter().flatten().collect())
-            })
-            .map_err(|error| EmbedError {
-                kind: EmbedErrorKind::PanicInThreadPool(error),
-                fault: FaultSource::Bug,
-            })?
+                    let embeddings = embeddings?;
+                    Ok(embeddings.into_iter().flatten().collect())
+                })
+                .map_err(|error| EmbedError {
+                    kind: EmbedErrorKind::PanicInThreadPool(error),
+                    fault: FaultSource::Bug,
+                })?
+        }
     }
 
     pub fn chunk_count_hint(&self) -> usize {

crates/milli/src/vector/rest.rs

@@ -203,20 +203,30 @@ impl Embedder {
         texts: &[&str],
         threads: &ThreadPoolNoAbort,
     ) -> Result<Vec<Embedding>, EmbedError> {
-        threads
-            .install(move || {
-                let embeddings: Result<Vec<Vec<Embedding>>, _> = texts
-                    .par_chunks(self.prompt_count_in_chunk_hint())
-                    .map(move |chunk| self.embed_ref(chunk, None))
-                    .collect();
+        if threads.active_operations() >= REQUEST_PARALLELISM {
+            let embeddings: Result<Vec<Vec<Embedding>>, _> = texts
+                .chunks(self.prompt_count_in_chunk_hint())
+                .map(move |chunk| self.embed_ref(chunk, None))
+                .collect();
 
-                let embeddings = embeddings?;
-                Ok(embeddings.into_iter().flatten().collect())
-            })
-            .map_err(|error| EmbedError {
-                kind: EmbedErrorKind::PanicInThreadPool(error),
-                fault: FaultSource::Bug,
-            })?
+            let embeddings = embeddings?;
+            Ok(embeddings.into_iter().flatten().collect())
+        } else {
+            threads
+                .install(move || {
+                    let embeddings: Result<Vec<Vec<Embedding>>, _> = texts
+                        .par_chunks(self.prompt_count_in_chunk_hint())
+                        .map(move |chunk| self.embed_ref(chunk, None))
+                        .collect();
+
+                    let embeddings = embeddings?;
+                    Ok(embeddings.into_iter().flatten().collect())
+                })
+                .map_err(|error| EmbedError {
+                    kind: EmbedErrorKind::PanicInThreadPool(error),
+                    fault: FaultSource::Bug,
+                })?
+        }
     }
 
     pub fn chunk_count_hint(&self) -> usize {