code: def test_large_messages(compression):
      x = "x" * 200_000_000
  
      msg = {
          "x": [to_serialize(x), b"small_bytes"],
          "y": {"a": to_serialize(x), "b": b"small_bytes"},
      }
  
      b = dumps(msg, context={"compression": compression})
      if compression is not None:
          assert sum(len(frame) for frame in b) < len(x)
  
      msg2 = loads(b)
      assert msg["x"][1] == msg2["x"][1]
      assert msg["y"]["b"] == msg2["y"]["b"]
      assert msg["x"][0].data == msg2["x"][0]
      assert msg["y"]["a"].data == msg2["y"]["a"]
  
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 15 for calling a global function with example usage: test_large_messages(compression) without return types