[BugFix] fix compression context pool slow down after long running (b…

…ackport #53172) (#53232)

Co-authored-by: Yixin Luo <[email protected]>

be/src/util/compression/compression_context_pool.h

@@ -114,14 +114,16 @@ class CompressionContextPool {
 
 private:
     void add(InternalRef ptr) {
+        // Use explicit producer token to avoid the overhead of too many sub-queues
+        static thread_local ::moodycamel::ProducerToken producer_token(_ctx_resources);
         DCHECK(ptr);
         Status status = _resetter(ptr.get());
         // if reset fail, then delete this context
         if (!status.ok()) {
             return;
         }
 
-        _ctx_resources.enqueue(std::move(ptr));
+        _ctx_resources.enqueue(producer_token, std::move(ptr));
     }
 
     Creator _creator;

be/test/util/block_compression_test.cpp

@@ -40,6 +40,7 @@
 #include <thread>
 
 #include "gen_cpp/segment.pb.h"
+#include "util/compression/compression_context_pool_singletons.h"
 #include "util/faststring.h"
 #include "util/random.h"
 #include "util/raw_container.h"
@@ -386,6 +387,22 @@ TEST_F(BlockCompressionTest, LZ4F_compression_LARGE_PAGE_TEST) {
     ASSERT_TRUE(st.ok());
 }
 
+TEST_F(BlockCompressionTest, test_multi_thread_get_ctx) {
+    for (int j = 0; j < 10; j++) {
+        std::vector<std::thread> workers;
+        for (int cnt = 0; cnt < 30; cnt++) {
+            workers.emplace_back([]() {
+                for (uint64_t i = 1; i < 1000; i++) {
+                    StatusOr<compression::LZ4F_CCtx_Pool::Ref> ref = compression::getLZ4F_CCtx();
+                }
+            });
+        }
+        for (auto& worker : workers) {
+            worker.join();
+        }
+    }
+}
+
 //#define LZ4_BENCHMARK
 //#define LZ4F_BENCHMARK
 //#define ZSTD_BENCHMARK