code: def test_duplicated_arg(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=1)
      ray.init(address=cluster.address)
  
      @ray.remote
      def task_with_dup_arg(*args):
          return sum(args)
  
      # Basic verification.
      arr = np.ones(1 * 1024 * 1024, dtype=np.uint8)  # 1MB
      ref = ray.put(arr)
      assert np.array_equal(
          ray.get(task_with_dup_arg.remote(ref, ref, ref)),
          sum([arr, arr, arr]),
      )
  
      # Make sure it works when it is mixed with other args.
      ref2 = ray.put(arr)
      assert np.array_equal(
          ray.get(task_with_dup_arg.remote(ref, ref2, ref)),
          sum([arr, arr, arr]),
      )
  
      # Test complicated scenario with multi nodes.
      cluster.add_node(num_cpus=1, resources={"worker_1": 1})
      cluster.add_node(num_cpus=1, resources={"worker_2": 1})
      cluster.wait_for_nodes()
  
      @ray.remote
      def create_remote_ref(arr):
          return ray.put(arr)
  
      @ray.remote
      def task_with_dup_arg_ref(*args):
          args = ray.get(list(args))
          return sum(args)
  
      ref1 = create_remote_ref.options(
          resources={"worker_1": 1}
      ).remote(arr)
      ref2 = create_remote_ref.options(
          resources={"worker_2": 1}
      ).remote(arr)
      ref3 = create_remote_ref.remote(arr)
      np.array_equal(
          ray.get(
              task_with_dup_arg_ref.remote(
                  ref1, ref2, ref3, ref1, ref2, ref3
              )
          ),
          sum([arr] * 6),
      )
  
imports: import glob
  import logging
  import multiprocessing
  import os
  import sys
  import tempfile
  import time
  from unittest import mock
  import numpy
  import psutil
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray._private.ray_constants
  import ray._private.utils
  import ray.cluster_utils
  import ray.util.accelerators
  from ray._private.test_utils import wait_for_condition
  from ray.dashboard import k8s_utils
  from ray.runtime_env import RuntimeEnv
label_desc: Use image node_id 13 for calling a global function with example usage: test_duplicated_arg(ray_start_cluster) and returns: sum, ray, sum