code: def test_dedupe_indirect(workflow_start_regular_shared, tmp_path):
      counter = Path(tmp_path) / "counter.txt"
      lock = Path(tmp_path) / "lock.txt"
      counter.write_text("0")
  
      @ray.remote
      def incr():
          with FileLock(str(lock)):
              c = int(counter.read_text())
              c += 1
              counter.write_text(f"{c}")
  
      @ray.remote
      def identity(a):
          return a
  
      @ray.remote
      def join(*a):
          return counter.read_text()
  
      # Here a is passed to two tasks and we need to ensure
      # it's only executed once
      a = incr.bind()
      i1 = identity.bind(a)
      i2 = identity.bind(a)
      assert "1" == workflow.run(join.bind(i1, i2))
      assert "2" == workflow.run(join.bind(i1, i2))
      # pass a multiple times
      assert "3" == workflow.run(join.bind(a, a, a, a))
      assert "4" == workflow.run(join.bind(a, a, a, a))
  
file_path: ray/python/ray/workflow/tests/test_basic_workflows_3.py
imports: import pytest
  from filelock import FileLock
  from pathlib import Path
  import ray
  from ray import workflow
  from ray.tests.conftest import *
label_desc: Use image node_id 2 for calling a global function with example usage: test_dedupe_indirect(workflow_start_regular_shared, tmp_path) and returns: a, counter