diff --git a/CHANGES.md b/CHANGES.md
index a6b0b443..4c12179f 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,6 +1,9 @@
 In development
 ==============
 
+- Add an opt-in shared-folder mode for restoring dynamically compiled Cython
+  functions and classes. ([issue #502](https://github.com/cloudpipe/cloudpickle/issues/502))
+
 - Make pickling of functions depending on globals in notebook more
   deterministic. ([PR#560](https://github.com/cloudpipe/cloudpickle/pull/560))
 
diff --git a/README.md b/README.md
index 12df5903..8521a13d 100644
--- a/README.md
+++ b/README.md
@@ -120,6 +120,44 @@ situations:
 - If a function pickled by reference uses a function pickled by value during its execution.
 
 
+Shared dynamically compiled Cython modules
+-----------------------------------------
+
+Functions and classes created by `pyximport` or the IPython `%%cython`
+magic may belong to modules outside Python's normal import search path. If
+the compiled extension files are accessible at the **same absolute paths**
+in every worker, enable `assume_shared_dynamic_cython_module_folder`:
+
+```python
+payload = cloudpickle.dumps(
+    compiled_function,
+    assume_shared_dynamic_cython_module_folder=True,
+)
+restored_function = pickle.loads(payload)
+```
+
+The option is also available on `cloudpickle.dump` and `cloudpickle.Pickler`.
+It defaults to `False`. Extensions already found on the normal import search
+path continue to use ordinary imports. With the option enabled, an extension
+outside that path is loaded from its recorded filename, and named functions
+and classes are looked up in that module. This also supports module references
+captured by Python functions. It does not serialize native code, compile
+sources on workers, or copy modified module globals.
+
+The compiled extension must be compatible with the worker's Python version,
+ABI, and platform, and its native libraries and parent packages must be
+available. The worker does not need Cython or a compilation hook. Definitions
+must be accessible by their qualified name in the module; locally defined
+Cython closures are outside this feature's scope. Modules explicitly registered
+with `register_pickle_by_value` retain that existing policy.
+
+Repeated loads reuse the module already loaded from the recorded path. If
+the same module name is already loaded from another location, loading raises
+`ImportError` instead of replacing it. Use distinct module names for different
+compiled versions. As with ordinary imports, module initialization runs when
+the extension is first loaded.
+
+
 Running the tests
 -----------------
 
diff --git a/cloudpickle/cloudpickle.py b/cloudpickle/cloudpickle.py
index 08882306..afd92aaa 100644
--- a/cloudpickle/cloudpickle.py
+++ b/cloudpickle/cloudpickle.py
@@ -59,9 +59,13 @@
 import dis
 from enum import Enum
 import io
+import importlib.machinery
+import importlib.util
+from importlib._bootstrap import _ModuleLockManager
 import itertools
 import logging
 import opcode
+import os
 import pickle
 from pickle import _getattribute as _pickle_getattribute
 import platform
@@ -312,6 +316,91 @@ def _lookup_module_and_qualname(obj, name=None):
     return module, name
 
 
+def _extension_module_path(module):
+    """Identify extensions that cannot be imported from the normal search path."""
+    loader = getattr(module, "__loader__", None)
+    path = getattr(module, "__file__", None)
+    if not isinstance(loader, importlib.machinery.ExtensionFileLoader) or not path:
+        return None
+    if _is_registered_pickle_by_value(module):
+        return None
+
+    path = os.path.realpath(path)
+    parent_name = module.__name__.rpartition(".")[0]
+    if parent_name:
+        parent = sys.modules.get(parent_name)
+        search_path = getattr(parent, "__path__", ())
+    else:
+        search_path = None
+    # Unlike importlib.util.find_spec, PathFinder does not return the spec
+    # cached in sys.modules or invoke a pyximport compilation hook.
+    spec = importlib.machinery.PathFinder.find_spec(module.__name__, search_path)
+    if spec is not None and spec.origin is not None:
+        if os.path.normcase(os.path.realpath(spec.origin)) == os.path.normcase(path):
+            return None
+    return path
+
+
+class _SharedExtensionFinder:
+    def __init__(self, module_name, path):
+        self.module_name = module_name
+        self.path = path
+
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname == self.module_name:
+            spec = importlib.util.spec_from_file_location(fullname, self.path)
+            if spec is None or not isinstance(
+                spec.loader, importlib.machinery.ExtensionFileLoader
+            ):
+                raise ImportError(
+                    f"Cannot load shared Cython extension from {self.path!r}"
+                )
+            return spec
+        return None
+
+
+def _check_dynamic_cython_module(module, module_name, path):
+    if module is None:
+        raise ModuleNotFoundError(
+            f"Import of {module_name!r} halted; None in sys.modules",
+            name=module_name,
+        )
+    loaded_path = getattr(module, "__file__", None)
+    if loaded_path is None or os.path.normcase(
+        os.path.realpath(loaded_path)
+    ) != os.path.normcase(os.path.realpath(path)):
+        raise ImportError(
+            f"Module {module_name!r} is already loaded from a different "
+            "location; cannot restore the shared Cython extension."
+        )
+    return module
+
+
+def _load_dynamic_cython_module(module_name, path):
+    """Load an extension from its shared location without changing sys.path."""
+    with _ModuleLockManager(module_name):
+        if module_name in sys.modules:
+            return _check_dynamic_cython_module(
+                sys.modules[module_name], module_name, path
+            )
+
+        # A parent package can import this extension in its own __init__.
+        # Let the normal importer handle that cycle and publish initialization
+        # state to concurrent imports, while supplying only this exact path.
+        finder = _SharedExtensionFinder(module_name, path)
+        sys.meta_path.insert(0, finder)
+        try:
+            module = importlib.import_module(module_name)
+            return _check_dynamic_cython_module(module, module_name, path)
+        finally:
+            sys.meta_path.remove(finder)
+
+
+def _load_dynamic_cython_attribute(module_name, path, name):
+    module = _load_dynamic_cython_module(module_name, path)
+    return _getattribute(module, name)
+
+
 def _extract_code_globals(co):
     """Find all globals names read or written to by codeblock co."""
     out_names = _extract_code_globals_cache.get(co)
@@ -1236,6 +1325,14 @@ def _get_dataclass_field_type_sentinel(name):
 
 
 class Pickler(pickle.Pickler):
+    """Serialize Python objects, including interactive functions and classes.
+
+    Set ``assume_shared_dynamic_cython_module_folder=True`` to restore
+    dynamically compiled Cython modules from their absolute extension paths.
+    The compiled files and their native dependencies must be accessible and
+    compatible on every unpickling process. This option is disabled by default.
+    """
+
     # set of reducers defined and used by cloudpickle (private)
     _dispatch_table = {}
     _dispatch_table[classmethod] = _classmethod_reduce
@@ -1329,7 +1426,50 @@ def dump(self, obj):
             msg = "Could not pickle object as excessively deep recursion required."
             raise pickle.PicklingError(msg) from e
 
-    def __init__(self, file, protocol=None, buffer_callback=None):
+    def _dynamic_cython_reduce(self, obj):
+        if isinstance(obj, types.ModuleType):
+            module, name = obj, None
+        else:
+            obj_type = type(obj)
+            if not (
+                issubclass(obj_type, type)
+                or isinstance(
+                    obj,
+                    (
+                        types.FunctionType,
+                        types.BuiltinFunctionType,
+                        types.MethodDescriptorType,
+                    ),
+                )
+                or obj_type.__name__
+                in ("cython_function_or_method", "fused_cython_function")
+            ):
+                return NotImplemented
+            module_and_name = _lookup_module_and_qualname(obj)
+            if module_and_name is None:
+                return NotImplemented
+            module, name = module_and_name
+
+        entry = self._dynamic_cython_modules.get(id(module))
+        if entry is None:
+            path = _extension_module_path(module)
+            self._dynamic_cython_modules[id(module)] = (module, path)
+        else:
+            _, path = entry
+        if path is None:
+            return NotImplemented
+        if name is None:
+            return _load_dynamic_cython_module, (module.__name__, path)
+        return _load_dynamic_cython_attribute, (module.__name__, path, name)
+
+    def __init__(
+        self,
+        file,
+        protocol=None,
+        buffer_callback=None,
+        *,
+        assume_shared_dynamic_cython_module_folder=False,
+    ):
         if protocol is None:
             protocol = DEFAULT_PROTOCOL
         super().__init__(file, protocol=protocol, buffer_callback=buffer_callback)
@@ -1338,6 +1478,12 @@ def __init__(self, file, protocol=None, buffer_callback=None):
         # their global namespace at unpickling time.
         self.globals_ref = {}
         self.proto = int(protocol)
+        self.assume_shared_dynamic_cython_module_folder = (
+            assume_shared_dynamic_cython_module_folder
+        )
+        self._dynamic_cython_modules = (
+            {} if assume_shared_dynamic_cython_module_folder else None
+        )
 
     if not PYPY:
         # pickle.Pickler is the C implementation of the CPython pickler and
@@ -1397,6 +1543,11 @@ def reducer_override(self, obj):
               reducers, such as Exceptions. See
               https://github.com/cloudpipe/cloudpickle/issues/248
             """
+            if self.assume_shared_dynamic_cython_module_folder:
+                reduction = self._dynamic_cython_reduce(obj)
+                if reduction is not NotImplemented:
+                    return reduction
+
             t = type(obj)
             try:
                 is_anyclass = issubclass(t, type)
@@ -1419,6 +1570,11 @@ def reducer_override(self, obj):
         # hard-coded call to save_global when pickling meta-classes.
         dispatch = pickle.Pickler.dispatch.copy()
 
+        def reducer_override(self, obj):
+            if self.assume_shared_dynamic_cython_module_folder:
+                return self._dynamic_cython_reduce(obj)
+            return NotImplemented
+
         def _save_reduce_pickle5(
             self,
             func,
@@ -1524,7 +1680,14 @@ def save_pypy_builtin_func(self, obj):
 # Shorthands similar to pickle.dump/pickle.dumps
 
 
-def dump(obj, file, protocol=None, buffer_callback=None):
+def dump(
+    obj,
+    file,
+    protocol=None,
+    buffer_callback=None,
+    *,
+    assume_shared_dynamic_cython_module_folder=False,
+):
     """Serialize obj as bytes streamed into file
 
     protocol defaults to cloudpickle.DEFAULT_PROTOCOL which is an alias to
@@ -1536,11 +1699,28 @@ def dump(obj, file, protocol=None, buffer_callback=None):
     guaranteed to work because cloudpickle relies on some internal
     implementation details that can change from one Python version to the
     next).
-    """
-    Pickler(file, protocol=protocol, buffer_callback=buffer_callback).dump(obj)
 
-
-def dumps(obj, protocol=None, buffer_callback=None):
+    ``assume_shared_dynamic_cython_module_folder`` has the same meaning as
+    for Pickler: opt in to loading dynamic Cython definitions from a shared
+    compiled-extension path.
+    """
+    Pickler(
+        file,
+        protocol=protocol,
+        buffer_callback=buffer_callback,
+        assume_shared_dynamic_cython_module_folder=(
+            assume_shared_dynamic_cython_module_folder
+        ),
+    ).dump(obj)
+
+
+def dumps(
+    obj,
+    protocol=None,
+    buffer_callback=None,
+    *,
+    assume_shared_dynamic_cython_module_folder=False,
+):
     """Serialize obj as a string of bytes allocated in memory
 
     protocol defaults to cloudpickle.DEFAULT_PROTOCOL which is an alias to
@@ -1552,9 +1732,20 @@ def dumps(obj, protocol=None, buffer_callback=None):
     guaranteed to work because cloudpickle relies on some internal
     implementation details that can change from one Python version to the
     next).
+
+    ``assume_shared_dynamic_cython_module_folder`` has the same meaning as
+    for Pickler: opt in to loading dynamic Cython definitions from a shared
+    compiled-extension path.
     """
     with io.BytesIO() as file:
-        cp = Pickler(file, protocol=protocol, buffer_callback=buffer_callback)
+        cp = Pickler(
+            file,
+            protocol=protocol,
+            buffer_callback=buffer_callback,
+            assume_shared_dynamic_cython_module_folder=(
+                assume_shared_dynamic_cython_module_folder
+            ),
+        )
         cp.dump(obj)
         return file.getvalue()
 
diff --git a/dev-requirements.txt b/dev-requirements.txt
index 7735fd92..6c85b6f5 100644
--- a/dev-requirements.txt
+++ b/dev-requirements.txt
@@ -19,3 +19,6 @@ coverage
 ./tests/cloudpickle_testpkg
 # Required for setup of the above utility package:
 setuptools; python_version >= '3.12'
+
+# Compile interactive extension modules for the shared-folder tests.
+Cython>=3.0
diff --git a/tests/test_cython.py b/tests/test_cython.py
new file mode 100644
index 00000000..85824648
--- /dev/null
+++ b/tests/test_cython.py
@@ -0,0 +1,498 @@
+import io
+import os
+import pickle
+import subprocess
+import sys
+import textwrap
+import uuid
+from unittest import mock
+
+import pytest
+
+import cloudpickle
+
+pytest.importorskip("Cython")
+
+CYTHON_SOURCE = """
+def plus_two(value):
+    return value + 2
+
+cpdef int triple(int value):
+    return value * 3
+
+class Plain:
+    def __init__(self, value=4):
+        self.value = value
+
+    def run(self, value):
+        return self.value + value
+
+cdef class Native:
+    cdef public int value
+
+    def __init__(self, value=5):
+        self.value = value
+
+    def run(self, value):
+        return self.value + value
+
+alias = plus_two
+counter = 0
+
+def increment():
+    global counter
+    counter += 1
+    return counter
+"""
+
+
+@pytest.fixture(scope="module", params=["pyximport", "cython_magic"])
+def cython_module(request, tmp_path_factory):
+    directory = tmp_path_factory.mktemp(request.param)
+    name = "cloudpickle_cython_" + uuid.uuid4().hex
+    if request.param == "pyximport":
+        import pyximport
+
+        (directory / (name + ".pyx")).write_text(CYTHON_SOURCE)
+        importers = pyximport.install(
+            build_dir=str(directory / "build"), language_level=3
+        )
+        sys.path.insert(0, str(directory))
+        try:
+            module = __import__(name)
+        finally:
+            sys.path.remove(str(directory))
+            pyximport.uninstall(*importers)
+    else:
+        from Cython.Build.IpythonMagic import CythonMagics
+        from IPython.core.interactiveshell import InteractiveShell
+
+        shell = InteractiveShell()
+        magic = CythonMagics(shell)
+        with mock.patch(
+            "Cython.Build.IpythonMagic.get_ipython_cache_dir",
+            return_value=str(directory),
+        ):
+            magic.cython("--name " + name, CYTHON_SOURCE)
+        module = sys.modules[shell.user_ns["plus_two"].__module__]
+    yield module
+    sys.modules.pop(module.__name__, None)
+
+
+def run_in_fresh_process(payload, expression, setup=""):
+    program = textwrap.dedent(
+        """
+        import pickle
+        import sys
+        """
+    )
+    program += setup + "\n"
+    program += "objects = pickle.loads(sys.stdin.buffer.read())\n"
+    program += "result = " + expression + "\n"
+    program += "sys.stdout.buffer.write(pickle.dumps(result))\n"
+    process = subprocess.run(
+        [sys.executable, "-c", program],
+        input=payload,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        timeout=60,
+        check=True,
+    )
+    return pickle.loads(process.stdout)
+
+
+@pytest.mark.parametrize("protocol", [2, 4, 5])
+def test_shared_cython_functions_and_classes(cython_module, protocol):
+    module = cython_module
+    objects = (
+        module.plus_two,
+        module.triple,
+        module.Plain,
+        module.Native,
+        module.alias,
+    )
+    payload = cloudpickle.dumps(
+        objects,
+        protocol=protocol,
+        assume_shared_dynamic_cython_module_folder=True,
+    )
+    assert run_in_fresh_process(
+        payload,
+        "(objects[0](5), objects[1](5), objects[2]().run(5), "
+        "objects[3]().run(5), objects[0] is objects[4], "
+        "'pyximport' in sys.modules, 'Cython' in sys.modules)",
+    ) == (7, 15, 9, 10, True, False, False)
+
+
+def test_shared_cython_module_and_python_wrapper(cython_module):
+    module = cython_module
+
+    def wrapped(value):
+        return module.plus_two(value)
+
+    payload = cloudpickle.dumps(
+        (module, wrapped), assume_shared_dynamic_cython_module_folder=True
+    )
+    assert run_in_fresh_process(payload, "(objects[0].triple(4), objects[1](4))") == (
+        12,
+        6,
+    )
+
+
+def test_shared_cython_instances_and_bound_methods(cython_module):
+    module = cython_module
+    instance = module.Native(9)
+    plain = module.Plain(7)
+    payload = cloudpickle.dumps(
+        (instance, instance.run, module.Native, plain, plain.run, module.Plain),
+        assume_shared_dynamic_cython_module_folder=True,
+    )
+    assert run_in_fresh_process(
+        payload,
+        "(objects[0].value, objects[1](1), type(objects[0]) is objects[2], "
+        "objects[3].value, objects[4](1), type(objects[3]) is objects[5])",
+    ) == (9, 10, True, 7, 8, True)
+
+
+def test_shared_cython_is_opt_in(cython_module):
+    default = cloudpickle.dumps(cython_module.plus_two)
+    disabled = cloudpickle.dumps(
+        cython_module.plus_two, assume_shared_dynamic_cython_module_folder=False
+    )
+    assert default == disabled
+    with pytest.raises(subprocess.CalledProcessError) as raised:
+        run_in_fresh_process(default, "objects(1)")
+    assert b"ModuleNotFoundError" in raised.value.stderr
+
+
+def test_shared_cython_pickler_and_dump(cython_module):
+    outputs = []
+    for constructor in ("pickler", "dump"):
+        stream = io.BytesIO()
+        if constructor == "pickler":
+            pickler = cloudpickle.Pickler(
+                stream, assume_shared_dynamic_cython_module_folder=True
+            )
+            pickler.dump(cython_module.plus_two)
+        else:
+            cloudpickle.dump(
+                cython_module.plus_two,
+                stream,
+                assume_shared_dynamic_cython_module_folder=True,
+            )
+        outputs.append(run_in_fresh_process(stream.getvalue(), "objects(3)"))
+    assert outputs == [5, 5]
+
+
+def test_shared_cython_preserves_existing_module(cython_module):
+    payload = cloudpickle.dumps(
+        (cython_module, cython_module.plus_two),
+        assume_shared_dynamic_cython_module_folder=True,
+    )
+    module, function = pickle.loads(payload)
+    assert module is cython_module
+    assert function is cython_module.plus_two
+
+
+def test_shared_cython_rejects_conflicting_module(cython_module):
+    payload = cloudpickle.dumps(
+        cython_module.plus_two, assume_shared_dynamic_cython_module_folder=True
+    )
+    setup = (
+        "import types\n"
+        f"conflict = types.ModuleType({cython_module.__name__!r})\n"
+        "conflict.__file__ = 'different-extension.so'\n"
+        "sys.modules[conflict.__name__] = conflict\n"
+    )
+    with pytest.raises(subprocess.CalledProcessError) as raised:
+        run_in_fresh_process(payload, "objects(1)", setup)
+    assert b"ImportError" in raised.value.stderr
+    assert b"already loaded" in raised.value.stderr
+
+
+def test_shared_cython_keeps_importable_extensions_by_reference():
+    import math
+
+    objects = (math, math.sin)
+    assert cloudpickle.dumps(
+        objects, assume_shared_dynamic_cython_module_folder=True
+    ) == cloudpickle.dumps(objects)
+
+
+def test_shared_cython_reuses_module_state_across_loads(cython_module):
+    payload = cloudpickle.dumps(
+        cython_module.increment, assume_shared_dynamic_cython_module_folder=True
+    )
+    assert run_in_fresh_process(
+        payload,
+        "(objects(), pickle.loads(payload)(), " "objects is pickle.loads(payload))",
+        f"payload = {payload!r}",
+    ) == (1, 2, True)
+
+
+def test_shared_cython_missing_binary_has_no_import_side_effects(
+    cython_module, tmp_path
+):
+    # Model a shared filename that is absent on the receiver without renaming
+    # a loaded DLL, which Windows does not allow.
+    missing_path = str(tmp_path / os.path.basename(cython_module.__file__))
+    with mock.patch.object(cython_module, "__file__", missing_path):
+        payload = cloudpickle.dumps(
+            cython_module.plus_two, assume_shared_dynamic_cython_module_folder=True
+        )
+    program = (
+        "import pickle,sys\n"
+        "before = tuple(sys.meta_path)\n"
+        "try:\n"
+        "    pickle.loads(sys.stdin.buffer.read())\n"
+        "except ImportError:\n"
+        f"    assert {cython_module.__name__!r} not in sys.modules\n"
+        "    assert tuple(sys.meta_path) == before\n"
+        "else:\n"
+        "    raise AssertionError('missing extension was loaded')\n"
+    )
+    subprocess.run(
+        [sys.executable, "-c", program],
+        input=payload,
+        capture_output=True,
+        timeout=60,
+        check=True,
+    )
+
+
+def test_shared_cython_blocked_module_stays_blocked(cython_module):
+    payload = cloudpickle.dumps(
+        cython_module.plus_two, assume_shared_dynamic_cython_module_folder=True
+    )
+    with pytest.raises(subprocess.CalledProcessError) as raised:
+        run_in_fresh_process(
+            payload,
+            "objects(1)",
+            f"sys.modules[{cython_module.__name__!r}] = None",
+        )
+    assert b"ModuleNotFoundError" in raised.value.stderr
+
+
+def test_shared_cython_respects_module_registration(cython_module):
+    cloudpickle.register_pickle_by_value(cython_module)
+    try:
+        from cloudpickle.cloudpickle import _extension_module_path
+
+        assert _extension_module_path(cython_module) is None
+    finally:
+        cloudpickle.unregister_pickle_by_value(cython_module)
+
+
+def test_shared_cython_unhashable_python_module():
+    import types
+
+    class Module(types.ModuleType):
+        __hash__ = None
+
+    module = Module("cloudpickle_unhashable_python_module")
+    exec("def function(): return 42", module.__dict__)
+    sys.modules[module.__name__] = module
+    try:
+        assert cloudpickle.dumps(
+            module.function, assume_shared_dynamic_cython_module_folder=True
+        ) == cloudpickle.dumps(module.function)
+    finally:
+        del sys.modules[module.__name__]
+
+
+def test_shared_cython_success_cleans_import_hooks(cython_module):
+    payload = cloudpickle.dumps(
+        cython_module.plus_two, assume_shared_dynamic_cython_module_folder=True
+    )
+    assert run_in_fresh_process(
+        payload,
+        "(objects(1), tuple(sys.meta_path) == before)",
+        "before = tuple(sys.meta_path)",
+    ) == (3, True)
+
+
+def test_shared_cython_reexporting_parent(tmp_path):
+    import pyximport
+
+    package_name = "cloudpickle_pkg_" + uuid.uuid4().hex
+    package = tmp_path / package_name
+    package.mkdir()
+    (package / "__init__.py").write_text("from . import native\nbefore = native\n")
+    (package / "native.pyx").write_text(CYTHON_SOURCE)
+    importers = pyximport.install(build_dir=str(tmp_path / "build"), language_level=3)
+    sys.path.insert(0, str(tmp_path))
+    try:
+        parent = __import__(package_name)
+    finally:
+        sys.path.remove(str(tmp_path))
+        pyximport.uninstall(*importers)
+    module = parent.native
+    try:
+        payload = cloudpickle.dumps(
+            (module, module.plus_two, module.Native),
+            assume_shared_dynamic_cython_module_folder=True,
+        )
+        assert run_in_fresh_process(
+            payload,
+            f"(objects[1](1), objects[2]().run(1), "
+            f"sys.modules[{package_name!r}].before is objects[0], "
+            "objects[2] is objects[0].Native)",
+            f"sys.path.insert(0, {str(tmp_path)!r})",
+        ) == (3, 6, True, True)
+    finally:
+        sys.modules.pop(module.__name__, None)
+        sys.modules.pop(package_name, None)
+
+
+def test_shared_cython_initialization_coordinates_imports(tmp_path):
+    import pyximport
+
+    name = "cloudpickle_initializing_" + uuid.uuid4().hex
+    source = """
+import os
+import time
+from pathlib import Path
+started = os.environ.get("CLOUDPICKLE_CYTHON_STARTED")
+if started:
+    Path(started).write_text("started")
+    while not Path(os.environ["CLOUDPICKLE_CYTHON_RELEASE"]).exists():
+        time.sleep(0.005)
+ready = True
+def function():
+    return ready
+"""
+    (tmp_path / (name + ".pyx")).write_text(source)
+    importers = pyximport.install(build_dir=str(tmp_path / "build"), language_level=3)
+    sys.path.insert(0, str(tmp_path))
+    try:
+        module = __import__(name)
+    finally:
+        sys.path.remove(str(tmp_path))
+        pyximport.uninstall(*importers)
+    try:
+        payload = cloudpickle.dumps(
+            module, assume_shared_dynamic_cython_module_folder=True
+        )
+        program = textwrap.dedent(
+            """
+            import importlib
+            import os
+            import pickle
+            from pathlib import Path
+            import sys
+            import threading
+            import time
+            from concurrent.futures import ThreadPoolExecutor
+
+            name, directory = sys.argv[1:]
+            started = Path(directory) / "started"
+            release = Path(directory) / "release"
+            os.environ["CLOUDPICKLE_CYTHON_STARTED"] = str(started)
+            os.environ["CLOUDPICKLE_CYTHON_RELEASE"] = str(release)
+            payload = sys.stdin.buffer.read()
+            before = tuple(sys.meta_path)
+            entered = threading.Event()
+            done = threading.Event()
+
+            def ordinary_import():
+                entered.set()
+                imported = importlib.import_module(name)
+                done.set()
+                return imported
+
+            with ThreadPoolExecutor(max_workers=2) as pool:
+                restored = pool.submit(pickle.loads, payload)
+                deadline = time.monotonic() + 10
+                while not started.exists():
+                    if restored.done():
+                        restored.result()
+                    assert time.monotonic() < deadline
+                    time.sleep(0.005)
+                imported = pool.submit(ordinary_import)
+                try:
+                    assert entered.wait(10)
+                    assert not done.wait(0.1)
+                finally:
+                    release.touch()
+                assert restored.result(timeout=10) is imported.result(timeout=10)
+                assert restored.result().ready
+            assert tuple(sys.meta_path) == before
+            """
+        )
+        subprocess.run(
+            [sys.executable, "-c", program, name, str(tmp_path)],
+            input=payload,
+            capture_output=True,
+            timeout=60,
+            check=True,
+        )
+    finally:
+        sys.modules.pop(name, None)
+
+
+def test_shared_cython_failed_initialization_can_retry(tmp_path):
+    import pyximport
+
+    name = "cloudpickle_init_failure_" + uuid.uuid4().hex
+    source = """
+import os
+if os.environ.get("CLOUDPICKLE_CYTHON_FAIL"):
+    raise RuntimeError("extension initialization failed")
+def function():
+    return 42
+"""
+    (tmp_path / (name + ".pyx")).write_text(source)
+    importers = pyximport.install(build_dir=str(tmp_path / "build"), language_level=3)
+    sys.path.insert(0, str(tmp_path))
+    try:
+        module = __import__(name)
+    finally:
+        sys.path.remove(str(tmp_path))
+        pyximport.uninstall(*importers)
+    try:
+        payload = cloudpickle.dumps(
+            module.function, assume_shared_dynamic_cython_module_folder=True
+        )
+        program = textwrap.dedent(
+            """
+            import os
+            import pickle
+            import sys
+            name = sys.argv[1]
+            payload = sys.stdin.buffer.read()
+            before = tuple(sys.meta_path)
+            os.environ["CLOUDPICKLE_CYTHON_FAIL"] = "1"
+            try:
+                pickle.loads(payload)
+            except RuntimeError as error:
+                assert str(error) == "extension initialization failed"
+            else:
+                raise AssertionError("expected module initialization to fail")
+            assert name not in sys.modules
+            assert tuple(sys.meta_path) == before
+            del os.environ["CLOUDPICKLE_CYTHON_FAIL"]
+            assert pickle.loads(payload)() == 42
+            assert tuple(sys.meta_path) == before
+            """
+        )
+        subprocess.run(
+            [sys.executable, "-c", program, name],
+            input=payload,
+            capture_output=True,
+            timeout=60,
+            check=True,
+        )
+    finally:
+        sys.modules.pop(name, None)
+
+
+def test_shared_cython_importable_compiled_module_uses_reference(cython_module):
+    directory = os.path.dirname(cython_module.__file__)
+    sys.path.insert(0, directory)
+    try:
+        objects = (cython_module, cython_module.plus_two, cython_module.Native)
+        assert cloudpickle.dumps(
+            objects, assume_shared_dynamic_cython_module_folder=True
+        ) == cloudpickle.dumps(objects)
+    finally:
+        sys.path.remove(directory)