[serialization] Enable debugging into pickle backend (#23854)

* enable debugging cloudpickle

doc/source/ray-core/objects/serialization.rst

@@ -197,6 +197,10 @@ The resulting output is:
     If you have any suggestions on how to improve this error message, please reach out to the Ray developers on github.com/ray-project/ray/issues/
     =============================================================
 
+For even more detailed information, set environmental variable ``RAY_PICKLE_VERBOSE_DEBUG='2'`` before importing Ray. This enables
+serialization with python-based backend instead of C-Pickle, so you can debug into python code at the middle of serialization.
+However, this would make serialization much slower.
+
 Known Issues
 ------------
 

python/ray/cloudpickle/compat.py

@@ -1,7 +1,21 @@
+import logging
+import os
 import sys
 
+logger = logging.getLogger(__name__)
 
-if sys.version_info < (3, 8):
+RAY_PICKLE_VERBOSE_DEBUG = os.environ.get("RAY_PICKLE_VERBOSE_DEBUG")
+verbose_level = int(RAY_PICKLE_VERBOSE_DEBUG) if RAY_PICKLE_VERBOSE_DEBUG else 0
+
+if verbose_level > 1:
+    logger.warning(
+        "Environmental variable RAY_PICKLE_VERBOSE_DEBUG is set to "
+        f"'{verbose_level}', this enabled python-based serialization backend "
+        f"instead of C-Pickle. Serialization would be very slow."
+    )
+    from ray.cloudpickle import py_pickle as pickle
+    from ray.cloudpickle.py_pickle import Pickler
+elif sys.version_info < (3, 8):
     try:
         import pickle5 as pickle  # noqa: F401
         from pickle5 import Pickler  # noqa: F401
@@ -10,4 +24,4 @@
         from pickle import _Pickler as Pickler  # noqa: F401
 else:
     import pickle  # noqa: F401
-    from _pickle import Pickler  # noqa: F401
+    from pickle import _Pickler as Pickler  # noqa: F401

python/ray/cloudpickle/py_pickle.py

@@ -0,0 +1,30 @@
+from pickle import (
+    _Pickler,
+    _Unpickler as Unpickler,
+    _loads as loads,
+    _load as load,
+    PickleError,
+    PicklingError,
+    UnpicklingError,
+    HIGHEST_PROTOCOL,
+)
+
+__all__ = [
+    "PickleError",
+    "PicklingError",
+    "UnpicklingError",
+    "Pickler",
+    "Unpickler",
+    "load",
+    "loads",
+    "HIGHEST_PROTOCOL",
+]
+
+
+class Pickler(_Pickler):
+    def __init__(self, file, protocol=None, *, fix_imports=True, buffer_callback=None):
+        super().__init__(
+            file, protocol, fix_imports=fix_imports, buffer_callback=buffer_callback
+        )
+        # avoid being overrided by cloudpickle
+        self.dispatch = _Pickler.dispatch.copy()