[core] Migrate serialization from _private to _common (#56590)

Moved `ray/_private/serialization.py`  into `ray/_common`.
Moved `test_serialization.py` from file `ray/tests` into
`ray/_common/tests`.

#53478 

Signed-off-by: Vaishnavi Panchavati <[email protected]>
Signed-off-by: elliot-barn <[email protected]>

Author: vaishdho1

python/ray/_common/serialization.py

@@ -0,0 +1,34 @@
+import io
+import logging
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    pass
+
+
+import ray._private.utils
+import ray.cloudpickle as pickle
+import ray.exceptions
+from ray._private import ray_constants
+from ray.util import inspect_serializability
+
+logger = logging.getLogger(__name__)
+ALLOW_OUT_OF_BAND_OBJECT_REF_SERIALIZATION = ray_constants.env_bool(
+    "RAY_allow_out_of_band_object_ref_serialization", True
+)
+
+
+def pickle_dumps(obj: Any, error_msg: str):
+    """Wrap cloudpickle.dumps to provide better error message
+    when the object is not serializable.
+    """
+    try:
+        return pickle.dumps(obj)
+    except (TypeError, ray.exceptions.OufOfBandObjectRefSerializationException) as e:
+        sio = io.StringIO()
+        inspect_serializability(obj, print_file=sio)
+        msg = f"{error_msg}:\n{sio.getvalue()}"
+        if isinstance(e, TypeError):
+            raise TypeError(msg) from e
+        else:
+            raise ray.exceptions.OufOfBandObjectRefSerializationException(msg)

python/ray/_private/function_manager.py

@@ -15,14 +15,14 @@
 import ray
 import ray._private.profiling as profiling
 from ray import cloudpickle as pickle
+from ray._common.serialization import pickle_dumps
 from ray._private import ray_constants
 from ray._private.inspect_util import (
     is_class_method,
     is_function_or_method,
     is_static_method,
 )
 from ray._private.ray_constants import KV_NAMESPACE_FUNCTION_TABLE
-from ray._private.serialization import pickle_dumps
 from ray._private.utils import (
     check_oversized_function,
     ensure_str,

python/ray/_private/serialization.py

@@ -1,4 +1,3 @@
-import io
 import logging
 import threading
 import traceback
@@ -54,7 +53,7 @@
     WorkerCrashedError,
 )
 from ray.experimental.compiled_dag_ref import CompiledDAGRef
-from ray.util import inspect_serializability, serialization_addons
+from ray.util import serialization_addons
 
 logger = logging.getLogger(__name__)
 ALLOW_OUT_OF_BAND_OBJECT_REF_SERIALIZATION = ray_constants.env_bool(
@@ -66,22 +65,6 @@ class DeserializationError(Exception):
     pass
 
 
-def pickle_dumps(obj: Any, error_msg: str):
-    """Wrap cloudpickle.dumps to provide better error message
-    when the object is not serializable.
-    """
-    try:
-        return pickle.dumps(obj)
-    except (TypeError, ray.exceptions.OufOfBandObjectRefSerializationException) as e:
-        sio = io.StringIO()
-        inspect_serializability(obj, print_file=sio)
-        msg = f"{error_msg}:\n{sio.getvalue()}"
-        if isinstance(e, TypeError):
-            raise TypeError(msg) from e
-        else:
-            raise ray.exceptions.OufOfBandObjectRefSerializationException(msg)
-
-
 def _object_ref_deserializer(
     binary, call_site, owner_address, object_status, tensor_transport_val
 ):

python/ray/air/util/object_extensions/arrow.py

@@ -6,8 +6,8 @@
 from packaging.version import parse as parse_version
 
 import ray.air.util.object_extensions.pandas
+from ray._common.serialization import pickle_dumps
 from ray._private.arrow_utils import get_pyarrow_version
-from ray._private.serialization import pickle_dumps
 from ray.util.annotations import PublicAPI
 
 MIN_PYARROW_VERSION_SCALAR_SUBCLASS = parse_version("9.0.0")

python/ray/remote_function.py

@@ -10,12 +10,12 @@
 from ray import Language, cross_language
 from ray._common import ray_option_utils
 from ray._common.ray_option_utils import _warn_if_using_deprecated_placement_group
+from ray._common.serialization import pickle_dumps
 from ray._private.auto_init_hook import wrap_auto_init
 from ray._private.client_mode_hook import (
     client_mode_convert_function,
     client_mode_should_convert,
 )
-from ray._private.serialization import pickle_dumps
 from ray._private.utils import get_runtime_env_info, parse_runtime_env_for_task_or_actor
 from ray._raylet import (
     STREAMING_GENERATOR_RETURN,

python/ray/serve/_private/benchmarks/serialization/serialization_benchmark.py

@@ -16,8 +16,8 @@
     PositiveInt,
     validator,
 )
+from ray._common.serialization import pickle_dumps
 from ray._common.utils import resources_from_ray_options
-from ray._private.serialization import pickle_dumps
 from ray.serve._private.constants import (
     DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_S,
     DEFAULT_GRACEFUL_SHUTDOWN_WAIT_LOOP_S,

python/ray/serve/_private/config.py

@@ -10,7 +10,7 @@
 
 import ray
 from ray import cloudpickle
-from ray._private.serialization import pickle_dumps
+from ray._common.serialization import pickle_dumps
 from ray.serve._private.build_app import build_app
 from ray.serve._private.config import (
     DeploymentConfig,