code: def test_cancel_during_arg_deser_non_reentrant_import(
      ray_start_regular,
  ):
      # This test ensures that task argument deserialization properly defers task
      # cancellation interrupts until after deserialization completes, in order to ensure
      # that non-reentrant imports that happen during both task argument deserialization
      # and during error storage are not interrupted.
  
      # We test this by doing the following:
      #  - register a custom serializer for (a) a task argument that triggers
      #  non-reentrant imports on deserialization, and (b) RayTaskError that triggers
      #  non-reentrant imports on serialization; in our case, we chose pandas it is both
      #  non-reentrant and expensive, with an import time ~0.5 seconds, giving us a wide
      #  cancellation target,
      #  - wait until those serializers are registered on all workers,
      #  - launch the task and wait until we are confident that the cancellation signal
      #  will be received by the workers during task argument deserialization (currently a
      #  200 ms wait).
      #  - check that a graceful task cancellation error is raised, not a
      # WorkerCrashedError.
      def non_reentrant_import():
          # NOTE: Pandas has a non-reentrant import and should take ~0.5 seconds to
          # import, giving us a wide cancellation target.
          import pandas  # noqa
  
      def non_reentrant_import_and_delegate(obj):
          # Custom serializer for task argument and task error resulting in non-reentrant
          # imports being imported on both serialization and deserialization. We use the
          # same custom serializer for both, doing non-reentrant imports on both
          # serialization and deserialization, for the sake of simplicity/reuse.
  
          # Import on serialization.
          non_reentrant_import()
  
          reduced = obj.__reduce__()
          func = reduced[0]
          args = reduced[1]
          others = reduced[2:]
  
          def non_reentrant_import_on_reconstruction(*args, **kwargs):
              # Import on deserialization.
              non_reentrant_import()
  
              return func(*args, **kwargs)
  
          out = (non_reentrant_import_on_reconstruction, args) + others
          return out
  
      # Dummy task argument for which we register a serializer that will trigger
      # non-reentrant imports on deserialization.
      class DummyArg:
          pass
  
      def register_non_reentrant_import_and_delegate_reducer(
          worker_info,
      ):
          from ray.exceptions import RayTaskError
  
          context = (
              ray._private.worker.global_worker.get_serialization_context()
          )
          # Register non-reentrant import serializer for task argument.
          context._register_cloudpickle_reducer(
              DummyArg, non_reentrant_import_and_delegate
          )
          # Register non-reentrant import serializer for RayTaskError.
          context._register_cloudpickle_reducer(
              RayTaskError, non_reentrant_import_and_delegate
          )
  
      ray._private.worker.global_worker.run_function_on_all_workers(
          register_non_reentrant_import_and_delegate_reducer,
      )
  
      # Wait for function to run on all workers.
      time.sleep(3)
  
      @ray.remote
      def run_and_fail(a: DummyArg):
          # Should never be reached.
          assert False
  
      arg = DummyArg()
      obj = run_and_fail.remote(arg)
      # Check that task isn't done.
      # NOTE: This timeout was finely tuned to ensure that task cancellation happens while
      # we are deserializing task arguments (10/10 runs when this comment was added).
      timeout_to_reach_arg_deserialization = 0.2
      assert (
          len(
              ray.wait(
                  [obj], timeout=timeout_to_reach_arg_deserialization
              )[0]
          )
          == 0
      )
  
      # Cancel task.
      use_force = False
      ray.cancel(obj, force=use_force)
  
      # Should raise RayTaskError or TaskCancelledError, NOT WorkerCrashedError.
      with pytest.raises(valid_exceptions(use_force)):
          ray.get(obj)
  
imports: import os
  import random
  import signal
  import sys
  import threading
  import _thread
  import time
  import pytest
  import ray
  from ray.exceptions import TaskCancelledError, RayTaskError, GetTimeoutError, WorkerCrashedError, ObjectLostError
  from ray._private.utils import DeferSigint
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.util.state import list_tasks
label_desc: Use image node_id 7 for calling a global function with example usage: test_cancel_during_arg_deser_non_reentrant_import(ray_start_regular) and returns: out, func