code: def _init_ray_serialization_deserialization():
      _ray_serialize = ray.serialization.SerializationContext.serialize
      _ray_deserialize_object = (
          ray.serialization.SerializationContext._deserialize_object
      )
      serialized_bytes_counter = Metrics.counter(
          "mars.channel_serialized_bytes",
          "The bytes serialized by mars ray channel.",
      )
      deserialized_bytes_counter = Metrics.counter(
          "mars.channel_deserialized_bytes",
          "The bytes deserialized by mars ray channel.",
      )
      serialization_time_mills = Metrics.counter(
          "mars.channel_serialization_time_mills",
          "The time used by mars ray channel serialization.",
      )
      deserialization_time_mills = Metrics.counter(
          "mars.channel_deserialization_time_mills",
          "The time used by mars ray channel deserialization.",
      )
  
      def _serialize(self, value):
          if (
              type(value) is _ArgWrapper
          ):  # pylint: disable=unidiomatic-typecheck
              message = value.message
              with Timer() as timer:
                  serialized_object = _ray_serialize(self, value)
                  bytes_length = serialized_object.total_bytes
                  serialized_bytes_counter.record(bytes_length)
              serialization_time_mills.record(timer.duration * 1000)
              if (
                  bytes_length > 1 * 1024 * 1024 * 1024
              ):  # pragma: no cover
                  logger.warning(
                      "Serialize large message (%s bytes > 1GB) through ray channel, message: %s.",
                      bytes_length,
                      msg_to_simple_str(message),
                  )
              if (
                  timer.duration * 1000 > SERIALIZATION_TIMEOUT_MILLS
              ):  # pragma: no cover
                  report_event(
                      "WARNING",
                      "SERIALIZATION_TIMEOUT",
                      f"Serialization took {timer.duration} seconds for {bytes_length} sized message {msg_to_simple_str(message)}.",
                  )
              try:
                  if message.profiling_context is not None:
                      task_id = message.profiling_context.task_id
                      ProfilingData[task_id, "serialization"].inc(
                          "serialize", timer.duration
                      )
              except AttributeError:  # pragma: no cover
                  logger.info(
                      "Profiling serialization got error, the send "
                      "message %s may not be an instance of message",
                      type(message),
                  )
          else:
              serialized_object = _ray_serialize(self, value)
          return serialized_object
  
      def _deserialize_object(self, data, metadata, object_ref):
          start_time = time.time()
          bytes_length = 0
          if data:
              bytes_length = len(data)
              deserialized_bytes_counter.record(bytes_length)
          value = _ray_deserialize_object(
              self, data, metadata, object_ref
          )
          duration = time.time() - start_time
          deserialization_time_mills.record(duration * 1000)
          if (
              duration * 1000 > DESERIALIZATION_TIMEOUT_MILLS
          ):  # pragma: no cover
              report_event(
                  "WARNING",
                  "DESERIALIZATION_TIMEOUT",
                  f"Deserialization took {duration} seconds for "
                  f"{bytes_length} sized msg {msg_to_simple_str(value)}",
              )
          if (
              type(value) is _ArgWrapper
          ):  # pylint: disable=unidiomatic-typecheck
              message = value.message
              try:
                  if message.profiling_context is not None:
                      task_id = message.profiling_context.task_id
                      ProfilingData[task_id, "serialization"].inc(
                          "deserialize", time.time() - start_time
                      )
              except AttributeError:  # pragma: no cover
                  logger.info(
                      "Profiling serialization got error, the recv "
                      "message %s may not be an instance of message",
                      type(message),
                  )
          return value
  
      ray.serialization.SerializationContext.serialize = _serialize
      ray.serialization.SerializationContext._deserialize_object = (
          _deserialize_object
      )
  
imports: import asyncio
  import concurrent.futures
  import itertools
  import logging
  import time
  from abc import ABC
  from collections import namedtuple
  from dataclasses import dataclass
  from typing import Any, Callable, Coroutine, Dict, List, Set, Tuple, Type
  from urllib.parse import urlparse
  from ....oscar.profiling import ProfilingData
  from ....serialization import serialize, deserialize
  from ....metrics import Metrics
  from ....utils import lazy_import, lazy_import_on_load, implements, classproperty, Timer
  from ...debug import debug_async_timeout
  from ...errors import ServerClosed
  from ..communication.base import Channel, ChannelType, Server, Client
  from ..communication.core import register_client, register_server
  from ..communication.errors import ChannelClosed
  from .utils import report_event
label_desc: Use image node_id 3 for calling a global function with example usage: _init_ray_serialization_deserialization() and returns: serialized_object, value