code: def test_compression_thread_safety(compression, dtype):
      barrier = threading.Barrier(4)
      expect = b"0" * 10000
      payload = dtype(expect)
  
      def compress_decompress():
          barrier.wait()
          for _ in range(2000):
              rc, rd = maybe_compress(payload, compression=compression)
              assert rc == compression
              assert compressions[rc].decompress(rd) == expect
  
      with ThreadPoolExecutor(4) as ex:
          futures = [ex.submit(compress_decompress) for _ in range(4)]
          for future in futures:
              future.result()
  
imports: from __future__ import annotations
  import threading
  import zlib
  from concurrent.futures import ThreadPoolExecutor
  import pytest
  import dask.config
  from distributed import Worker, wait
  from distributed.compatibility import LINUX
  from distributed.protocol import dumps, loads, maybe_compress, msgpack, to_serialize
  from distributed.protocol.compression import Compression, compressions, get_compression_settings
  from distributed.utils import nbytes
  from distributed.utils_test import gen_cluster
label_desc: Use image node_id 14 for calling a global function with example usage: test_compression_thread_safety(compression, dtype) without return types