Compression cpu thread pool

quickwit/Cargo.toml

@@ -187,7 +187,7 @@ pulsar = { git = "https://github.com/quickwit-oss/pulsar-rs.git", rev = "f9eff04
 quote = "1.0.23"
 rand = "0.8"
 rand_distr = "0.4"
-rayon = "1"
+rayon = "1.10"
 rdkafka = { version = "0.33", default-features = false, features = [
   "cmake-build",
   "libz",
@@ -269,16 +269,12 @@ wiremock = "0.5"
 zstd = "0.13.0"
 
 aws-config = "1.2"
-aws-credential-types = { version = "1.2", features = [
-  "hardcoded-credentials",
-] }
+aws-credential-types = { version = "1.2", features = ["hardcoded-credentials"] }
 aws-sdk-kinesis = "1.21"
 aws-sdk-s3 = "1.24"
 aws-smithy-async = "1.2"
 aws-smithy-runtime = "1.3"
-aws-smithy-types = { version = "1.1", features = [
-  "byte-stream-poll-next"
-] }
+aws-smithy-types = { version = "1.1", features = ["byte-stream-poll-next"] }
 aws-types = "1.2"
 
 azure_core = { version = "0.13.0", features = ["enable_reqwest_rustls"] }
@@ -321,7 +317,7 @@ quickwit-serve = { path = "quickwit-serve" }
 quickwit-storage = { path = "quickwit-storage" }
 quickwit-telemetry = { path = "quickwit-telemetry" }
 
-tantivy = { git = "https://github.com/quickwit-oss/tantivy/", rev = "1ee5f907", default-features = false, features = [
+tantivy = { git = "https://github.com/quickwit-oss/tantivy/", rev = "6181c1e", default-features = false, features = [
   "lz4-compression",
   "mmap",
   "quickwit",

quickwit/quickwit-common/Cargo.toml

@@ -30,6 +30,7 @@ pin-project = { workspace = true }
 pnet = { workspace = true }
 prometheus = { workspace = true }
 rand = { workspace = true }
+rayon = { workspace = true }
 regex = { workspace = true }
 serde = { workspace = true }
 siphasher = { workspace = true }

quickwit/quickwit-common/src/lib.rs

@@ -43,6 +43,7 @@ pub mod stream_utils;
 pub mod temp_dir;
 #[cfg(any(test, feature = "testsuite"))]
 pub mod test_utils;
+pub mod thread_pool;
 pub mod tower;
 pub mod type_map;
 pub mod uri;

quickwit/quickwit-common/src/metrics.rs

@@ -179,19 +179,19 @@ pub fn new_histogram_vec<const N: usize>(
     HistogramVec { underlying }
 }
 
-pub struct GaugeGuard {
-    gauge: &'static IntGauge,
+pub struct GaugeGuard<'a> {
+    gauge: &'a IntGauge,
     delta: i64,
 }
 
-impl std::fmt::Debug for GaugeGuard {
+impl<'a> std::fmt::Debug for GaugeGuard<'a> {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         self.delta.fmt(f)
     }
 }
 
-impl GaugeGuard {
-    pub fn from_gauge(gauge: &'static IntGauge) -> Self {
+impl<'a> GaugeGuard<'a> {
+    pub fn from_gauge(gauge: &'a IntGauge) -> Self {
         Self { gauge, delta: 0i64 }
     }
 
@@ -210,7 +210,44 @@ impl GaugeGuard {
     }
 }
 
-impl Drop for GaugeGuard {
+impl<'a> Drop for GaugeGuard<'a> {
+    fn drop(&mut self) {
+        self.gauge.sub(self.delta)
+    }
+}
+
+pub struct OwnedGaugeGuard {
+    gauge: IntGauge,
+    delta: i64,
+}
+
+impl std::fmt::Debug for OwnedGaugeGuard {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        self.delta.fmt(f)
+    }
+}
+
+impl OwnedGaugeGuard {
+    pub fn from_gauge(gauge: IntGauge) -> Self {
+        Self { gauge, delta: 0i64 }
+    }
+
+    pub fn get(&self) -> i64 {
+        self.delta
+    }
+
+    pub fn add(&mut self, delta: i64) {
+        self.gauge.add(delta);
+        self.delta += delta;
+    }
+
+    pub fn sub(&mut self, delta: i64) {
+        self.gauge.sub(delta);
+        self.delta -= delta;
+    }
+}
+
+impl Drop for OwnedGaugeGuard {
     fn drop(&mut self) {
         self.gauge.sub(self.delta)
     }

quickwit/quickwit-common/src/stream_utils.rs

@@ -233,7 +233,7 @@ where T: RpcName
     }
 }
 
-pub struct InFlightValue<T>(T, #[allow(dead_code)] GaugeGuard);
+pub struct InFlightValue<T>(T, #[allow(dead_code)] GaugeGuard<'static>);
 
 impl<T> fmt::Debug for InFlightValue<T>
 where T: fmt::Debug