[SYCL] Don't use zstd context across threads. (#19747)

`ZSTDCompressor` holds `zstd` context as its only data members. The idea
behind `GetSingletonInstance()` method was to re-use these contexts for
subsequent compression and decompressions. Re-using context across
(de)compression reduces system memory usage.

However, `zstd` contexts are not meant to be used concurrently,
therefore, this PR makes `ZSTDCompressor` object thread-local, instead
of static.
Relevant excerpt from zstd doc
(https://facebook.github.io/zstd/zstd_manual.html):

>   When decompressing many times,
>   it is recommended to allocate a context only once,
>   and re-use it for each successive compression operation.
>   This will make workload friendlier for system's memory.
>   Use one context per thread for parallel execution.

Author: uditagarwal97

sycl/source/detail/compression.hpp

@@ -33,7 +33,9 @@ class ZSTDCompressor {
 
   // Get the singleton instance of the ZSTDCompressor class.
   static ZSTDCompressor &GetSingletonInstance() {
-    static ZSTDCompressor instance;
+    // Use thread_local to ensure that each thread has its own instance.
+    // This avoids issues with concurrent access to the ZSTD contexts.
+    thread_local ZSTDCompressor instance;
     return instance;
   }
 

sycl/unittests/compression/CompressionTests.cpp

@@ -6,6 +6,7 @@
 //
 //===----------------------------------------------------------------------===//
 
+#include "../thread_safety/ThreadUtils.h"
 #include <detail/compression.hpp>
 #include <sycl/sycl.hpp>
 
@@ -79,3 +80,36 @@ TEST(CompressionTest, EmptyInputTest) {
   std::string decompressedStr((char *)decompressedData.get(), decompressedSize);
   ASSERT_EQ(input, decompressedStr);
 }
+
+// Test to check for concurrent compression and decompression.
+TEST(CompressionTest, ConcurrentCompressionDecompression) {
+  std::string data = "Concurrent compression and decompression test!";
+
+  constexpr size_t ThreadCount = 20;
+
+  Barrier b(ThreadCount);
+  {
+    auto testCompressDecompress = [&](size_t threadId) {
+      b.wait();
+      size_t compressedDataSize = 0;
+      auto compressedData = ZSTDCompressor::CompressBlob(
+          data.c_str(), data.size(), compressedDataSize, 3);
+
+      ASSERT_NE(compressedData, nullptr);
+      ASSERT_GT(compressedDataSize, (size_t)0);
+
+      size_t decompressedSize = 0;
+      auto decompressedData = ZSTDCompressor::DecompressBlob(
+          compressedData.get(), compressedDataSize, decompressedSize);
+
+      ASSERT_NE(decompressedData, nullptr);
+      ASSERT_GT(decompressedSize, (size_t)0);
+
+      std::string decompressedStr((char *)decompressedData.get(),
+                                  decompressedSize);
+      ASSERT_EQ(data, decompressedStr);
+    };
+
+    ::ThreadPool MPool(ThreadCount, testCompressDecompress);
+  }
+}