code: def test_complex_serialization(ray_start_regular):
      def assert_equal(obj1, obj2):
          module_numpy = (
              type(obj1).__module__ == np.__name__
              or type(obj2).__module__ == np.__name__
          )
          if module_numpy:
              empty_shape = (
                  hasattr(obj1, "shape") and obj1.shape == ()
              ) or (hasattr(obj2, "shape") and obj2.shape == ())
              if empty_shape:
                  # This is a special case because currently
                  # np.testing.assert_equal fails because we do not properly
                  # handle different numerical types.
                  assert (
                      obj1 == obj2
                  ), "Objects {} and {} are different.".format(
                      obj1, obj2
                  )
              else:
                  np.testing.assert_equal(obj1, obj2)
          elif hasattr(obj1, "__dict__") and hasattr(obj2, "__dict__"):
              special_keys = ["_pytype_"]
              assert set(
                  list(obj1.__dict__.keys()) + special_keys
              ) == set(
                  list(obj2.__dict__.keys()) + special_keys
              ), "Objects {} and {} are different.".format(
                  obj1, obj2
              )
              for key in obj1.__dict__.keys():
                  if key not in special_keys:
                      assert_equal(
                          obj1.__dict__[key], obj2.__dict__[key]
                      )
          elif type(obj1) is dict or type(obj2) is dict:
              assert_equal(obj1.keys(), obj2.keys())
              for key in obj1.keys():
                  assert_equal(obj1[key], obj2[key])
          elif type(obj1) is list or type(obj2) is list:
              assert len(obj1) == len(
                  obj2
              ), "Objects {} and {} are lists with different lengths.".format(
                  obj1, obj2
              )
              for i in range(len(obj1)):
                  assert_equal(obj1[i], obj2[i])
          elif type(obj1) is tuple or type(obj2) is tuple:
              assert len(obj1) == len(
                  obj2
              ), "Objects {} and {} are tuples with different lengths.".format(
                  obj1, obj2
              )
              for i in range(len(obj1)):
                  assert_equal(obj1[i], obj2[i])
          elif is_named_tuple(type(obj1)) or is_named_tuple(type(obj2)):
              assert len(obj1) == len(
                  obj2
              ), "Objects {} and {} are named tuples with different lengths.".format(
                  obj1, obj2
              )
              for i in range(len(obj1)):
                  assert_equal(obj1[i], obj2[i])
          else:
              assert (
                  obj1 == obj2
              ), "Objects {} and {} are different.".format(obj1, obj2)
  
      long_extras = [0, np.array([["hi", "hi"], [1.3, 1]])]
  
      PRIMITIVE_OBJECTS = [
          0,
          0.0,
          0.9,
          1 << 62,
          1 << 100,
          1 << 999,
          [1 << 100, [1 << 100]],
          "a",
          string.printable,
          "\u262F",
          "hello world",
          "\xff\xfe\x9c\x001\x000\x00",
          None,
          True,
          False,
          [],
          (),
          {},
          np.int8(3),
          np.int32(4),
          np.int64(5),
          np.uint8(3),
          np.uint32(4),
          np.uint64(5),
          np.float32(1.9),
          np.float64(1.9),
          np.zeros([100, 100]),
          np.random.normal(size=[100, 100]),
          np.array(["hi", 3]),
          np.array(["hi", 3], dtype=object),
      ] + long_extras
  
      COMPLEX_OBJECTS = [
          [[[[[[[[[[[[]]]]]]]]]]]],
          {
              "obj{}".format(i): np.random.normal(size=[100, 100])
              for i in range(10)
          },
          # {(): {(): {(): {(): {(): {(): {(): {(): {(): {(): {
          #      (): {(): {}}}}}}}}}}}}},
          ((((((((((),),),),),),),),),),
          {"a": {"b": {"c": {"d": {}}}}},
      ]
  
      class Foo:
          def __init__(self, value=0):
              self.value = value
  
          def __hash__(self):
              return hash(self.value)
  
          def __eq__(self, other):
              return other.value == self.value
  
      class Bar:
          def __init__(self):
              for i, val in enumerate(
                  PRIMITIVE_OBJECTS + COMPLEX_OBJECTS
              ):
                  setattr(self, "field{}".format(i), val)
  
      class Baz:
          def __init__(self):
              self.foo = Foo()
              self.bar = Bar()
  
          def method(self, arg):
              pass
  
      class Qux:
          def __init__(self):
              self.objs = [Foo(), Bar(), Baz()]
  
      class SubQux(Qux):
          def __init__(self):
              Qux.__init__(self)
  
      class CustomError(Exception):
          pass
  
      Point = collections.namedtuple("Point", ["x", "y"])
      NamedTupleExample = collections.namedtuple(
          "Example", "field1, field2, field3, field4, field5"
      )
  
      CUSTOM_OBJECTS = [
          Exception("Test object."),
          CustomError(),
          Point(11, y=22),
          Foo(),
          Bar(),
          Baz(),  # Qux(), SubQux(),
          NamedTupleExample(1, 1.0, "hi", np.zeros([3, 5]), [1, 2, 3]),
      ]
  
      DataClass0 = make_dataclass("DataClass0", [("number", int)])
  
      CUSTOM_OBJECTS.append(DataClass0(number=3))
  
      class CustomClass:
          def __init__(self, value):
              self.value = value
  
      DataClass1 = make_dataclass(
          "DataClass1", [("custom", CustomClass)]
      )
  
      class DataClass2(DataClass1):
          @classmethod
          def from_custom(cls, data):
              custom = CustomClass(data)
              return cls(custom)
  
          def __reduce__(self):
              return (self.from_custom, (self.custom.value,))
  
      CUSTOM_OBJECTS.append(DataClass2(custom=CustomClass(43)))
  
      BASE_OBJECTS = (
          PRIMITIVE_OBJECTS + COMPLEX_OBJECTS + CUSTOM_OBJECTS
      )
  
      LIST_OBJECTS = [[obj] for obj in BASE_OBJECTS]
      TUPLE_OBJECTS = [(obj,) for obj in BASE_OBJECTS]
      # The check that type(obj).__module__ != "numpy" should be unnecessary, but
      # otherwise this seems to fail on Mac OS X on Travis.
      DICT_OBJECTS = (
          [
              {obj: obj}
              for obj in PRIMITIVE_OBJECTS
              if (
                  obj.__hash__ is not None
                  and type(obj).__module__ != "numpy"
              )
          ]
          + [{0: obj} for obj in BASE_OBJECTS]
          + [{Foo(123): Foo(456)}]
      )
  
      RAY_TEST_OBJECTS = (
          BASE_OBJECTS + LIST_OBJECTS + TUPLE_OBJECTS + DICT_OBJECTS
      )
  
      @ray.remote
      def f(x):
          return x
  
      # Check that we can pass arguments by value to remote functions and
      # that they are uncorrupted.
      for obj in RAY_TEST_OBJECTS:
          assert_equal(obj, ray.get(f.remote(obj)))
          assert_equal(obj, ray.get(ray.put(obj)))
  
      # Test StringIO serialization
      s = io.StringIO("Hello, world!\n")
      s.seek(0)
      line = s.readline()
      s.seek(0)
      assert ray.get(ray.put(s)).readline() == line
  
imports: import collections
  import io
  import logging
  import re
  import string
  import sys
  import weakref
  from dataclasses import make_dataclass
  import numpy
  import pytest
  from numpy import log
  import ray
  import ray.cluster_utils
label_desc: Use image node_id 3 for calling a global function with example usage: test_complex_serialization(ray_start_regular) and returns: x, hash, unknown, cls, self,  self