code: def test_dedupe_serialization(workflow_start_regular_shared):
      @ray.remote(num_cpus=0)
      class Counter:
          def __init__(self):
              self.count = 0
  
          def incr(self):
              self.count += 1
  
          def get_count(self):
              return self.count
  
      counter = Counter.remote()
  
      class CustomClass:
          def __getstate__(self):
              # Count the number of times this class is serialized.
              ray.get(counter.incr.remote())
              return {}
  
      ref = ray.put(CustomClass())
      list_of_refs = [ref for _ in range(2)]
  
      # One for the ray.put
      assert ray.get(counter.get_count.remote()) == 1
  
      single = identity.bind((ref,))
      double = identity.bind(list_of_refs)
  
      workflow.run(gather.bind(single, double))
  
      # One more for hashing the ref, and for uploading.
      assert ray.get(counter.get_count.remote()) == 3
  
file_path: ray/python/ray/workflow/tests/test_serialization.py
imports: import pytest
  import ray
  from ray import workflow
  from ray.tests.conftest import *
  from ray.workflow import serialization
label_desc: Use image node_id 4 for calling a global function with example usage: test_dedupe_serialization(workflow_start_regular_shared) and returns: self, dict