refactor: remove async remote write decode logic (#1631)

## Rationale
Deserialization is a CPU-intensive task, so introducing async does not
provide significant benefits. In addition, using async causes function
contagion issues. Therefore, it is necessary to remove the async decode
logic.


## Detailed Changes
- Remove `decode_async` interface
- Change the back-end object pool dependency from `deadpool` to
`object-pool`
- Change benchmark logic accordingly


## Test Plan
Manual test

Author: BaldDemian

Cargo.lock

@@ -51,11 +51,13 @@ columnar_storage = { path = "src/columnar_storage" }
 common = { path = "src/common" }
 criterion = "0.5"
 datafusion = "43"
-deadpool = "0.10"
 futures = "0.3"
+hotpath = "0.5.2"
 itertools = "0.3"
 lazy_static = "1"
 metric_engine = { path = "src/metric_engine" }
+num_cpus = "1"
+object-pool = "0.6"
 object_store = { version = "0.11" }
 once_cell = "1"
 parquet = { version = "53" }

Cargo.toml

@@ -29,40 +29,38 @@ description.workspace = true
 name = "parser_mem"
 path = "src/bin/parser_mem.rs"
 
-[[bin]]
-name = "pool_stats"
-path = "src/bin/pool_stats.rs"
-
 [features]
+default = ["hotpath", "hotpath-alloc-bytes-total"]
+hotpath = []
+hotpath-alloc-bytes-total = ["hotpath", "hotpath/hotpath-alloc-bytes-total"]
 unsafe-split = ["remote_write/unsafe-split"]
 
 [dependencies]
+anyhow = { workspace = true }
 bytes = { workspace = true }
 columnar_storage = { workspace = true }
 common = { workspace = true }
-deadpool = { workspace = true }
+hotpath = { workspace = true }
+num_cpus = { workspace = true }
 pb_types = { workspace = true }
 prost = { workspace = true }
 protobuf = "3.7"
 quick-protobuf = "0.8"
 remote_write = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
-tikv-jemalloc-ctl = "0.5"
 tokio = { workspace = true }
 toml = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
-tikv-jemallocator = "0.5"
 
 [build-dependencies]
 protobuf-codegen = "3.7"
 
 [dev-dependencies]
 criterion = { workspace = true }
-num_cpus = "1.16"
 
 [[bench]]
 name = "bench"

docs/assets/remote-write-concurrent-performance.png

@@ -64,12 +64,6 @@ fn bench_remote_write(c: &mut Criterion) {
     let concurrent_scales = config.remote_write.concurrent_scales.clone();
     let bench = RefCell::new(RemoteWriteBench::new(config.remote_write));
 
-    let rt = tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(num_cpus::get())
-        .enable_all()
-        .build()
-        .unwrap();
-
     // Sequential parse bench.
     let mut group = c.benchmark_group("remote_write_sequential");
 
@@ -85,10 +79,10 @@ fn bench_remote_write(c: &mut Criterion) {
 
         group.bench_with_input(
             BenchmarkId::new("pooled", n),
-            &(&bench, &rt, n),
-            |b, (bench, rt, scale)| {
+            &(&bench, n),
+            |b, (bench, scale)| {
                 let bench = bench.borrow();
-                b.iter(|| rt.block_on(bench.pooled_parser_sequential(*scale)).unwrap())
+                b.iter(|| bench.pooled_parser_sequential(*scale).unwrap())
             },
         );
 
@@ -118,43 +112,37 @@ fn bench_remote_write(c: &mut Criterion) {
     for &scale in &concurrent_scales {
         group.bench_with_input(
             BenchmarkId::new("prost", scale),
-            &(&bench, &rt, scale),
-            |b, (bench, rt, scale)| {
+            &(&bench, scale),
+            |b, (bench, scale)| {
                 let bench = bench.borrow();
-                b.iter(|| rt.block_on(bench.prost_parser_concurrent(*scale)).unwrap())
+                b.iter(|| bench.prost_parser_concurrent(*scale).unwrap())
             },
         );
 
         group.bench_with_input(
             BenchmarkId::new("pooled", scale),
-            &(&bench, &rt, scale),
-            |b, (bench, rt, scale)| {
+            &(&bench, scale),
+            |b, (bench, scale)| {
                 let bench = bench.borrow();
-                b.iter(|| rt.block_on(bench.pooled_parser_concurrent(*scale)).unwrap())
+                b.iter(|| bench.pooled_parser_concurrent(*scale).unwrap())
             },
         );
 
         group.bench_with_input(
             BenchmarkId::new("quick_protobuf", scale),
-            &(&bench, &rt, scale),
-            |b, (bench, rt, scale)| {
+            &(&bench, scale),
+            |b, (bench, scale)| {
                 let bench = bench.borrow();
-                b.iter(|| {
-                    rt.block_on(bench.quick_protobuf_parser_concurrent(*scale))
-                        .unwrap()
-                })
+                b.iter(|| bench.quick_protobuf_parser_concurrent(*scale).unwrap())
             },
         );
 
         group.bench_with_input(
             BenchmarkId::new("rust_protobuf", scale),
-            &(&bench, &rt, scale),
-            |b, (bench, rt, scale)| {
+            &(&bench, scale),
+            |b, (bench, scale)| {
                 let bench = bench.borrow();
-                b.iter(|| {
-                    rt.block_on(bench.rust_protobuf_parser_concurrent(*scale))
-                        .unwrap()
-                })
+                b.iter(|| bench.rust_protobuf_parser_concurrent(*scale).unwrap())
             },
         );
     }