code: def test_workflow_queuing_2(shutdown_only, tmp_path):
      ray.init(storage=str(tmp_path))
      workflow.init(max_running_workflows=2, max_pending_workflows=2)
  
      @ray.remote
      def short_running(x):
          return x
  
      wfs = [short_running.bind(i) for i in range(5)]
      refs = [
          workflow.run_async(wfs[i], workflow_id=f"workflow_{i}")
          for i in range(4)
      ]
      for i in range(4):
          assert workflow.get_output(f"workflow_{i}") == i
      assert ray.get(refs) == [0, 1, 2, 3]
      assert workflow.run(wfs[4], workflow_id="workflow_4") == 4
      assert sorted(
          x[0] for x in workflow.list_all({workflow.SUCCESSFUL})
      ) == [
          "workflow_0",
          "workflow_1",
          "workflow_2",
          "workflow_3",
          "workflow_4",
      ]
  
file_path: ray/python/ray/workflow/tests/test_workflow_queuing.py
imports: import pytest
  import ray
  from ray import workflow
  from ray.tests.conftest import *
label_desc: Use image node_id 4 for calling a global function with example usage: test_workflow_queuing_2(shutdown_only, tmp_path) and returns: x