code: def test_task_resources(workflow_start_regular, tmp_path):
      lock_path = str(tmp_path / "lock")
      # We use signal actor here because we can't guarantee the order of tasks
      # sent from worker to raylet.
      signal_actor = SignalActor.remote()
  
      @ray.remote
      def task_run():
          ray.wait([signal_actor.send.remote()])
          with FileLock(lock_path):
              return None
  
      @ray.remote(num_cpus=1)
      def remote_run():
          return None
  
      lock = FileLock(lock_path)
      lock.acquire()
      ret = workflow.run_async(task_run.options(num_cpus=2).bind())
      ray.wait([signal_actor.wait.remote()])
      obj = remote_run.remote()
      with pytest.raises(ray.exceptions.GetTimeoutError):
          ray.get(obj, timeout=2)
      lock.release()
      assert ray.get(ret) is None
      assert ray.get(obj) is None
  
file_path: ray/python/ray/workflow/tests/test_basic_workflows_2.py
imports: import pytest
  import ray
  from filelock import FileLock
  from ray._private.test_utils import SignalActor
  from ray import workflow
  from ray.tests.conftest import *
  from ray._private.test_utils import skip_flaky_core_test_premerge
label_desc: Use image node_id 1 for calling a global function with example usage: test_task_resources(workflow_start_regular, tmp_path) and returns: None, None