code: def test_pretask_posttask_shared_state_multi(ray_start_1_cpu):
      """
      Test that pretask return values are passed to the correct corresponding
      posttask callbacks when multiple callbacks are given.
      """
  
      class PretaskPosttaskCallback(RayDaskCallback):
          def __init__(self, suffix):
              self.suffix = suffix
  
          def _ray_pretask(self, key, object_refs):
              return key + self.suffix
  
          def _ray_posttask(self, key, result, pre_state):
              assert pre_state == key + self.suffix
  
      class PretaskOnlyCallback(RayDaskCallback):
          def _ray_pretask(self, key, object_refs):
              return "baz"
  
      class PosttaskOnlyCallback(RayDaskCallback):
          def _ray_posttask(self, key, result, pre_state):
              assert pre_state is None
  
      cb1 = PretaskPosttaskCallback("foo")
      cb2 = PretaskOnlyCallback()
      cb3 = PosttaskOnlyCallback()
      cb4 = PretaskPosttaskCallback("bar")
      with cb1, cb2, cb3, cb4:
          z = add(2, 3)
          result = z.compute(scheduler=ray_dask_get)
  
      assert result == 5
  
file_path: ray/python/ray/util/dask/tests/test_dask_callback.py
imports: import dask
  import pytest
  import ray
  from ray.util.dask import ray_dask_get, RayDaskCallback
label_desc: Use image node_id 10 for calling a global function with example usage: test_pretask_posttask_shared_state_multi(ray_start_1_cpu) and returns: unknown, str