SharedMemoryDataCatalog

def set_manager_datasets(self, manager: SyncManager) -> None:
    """
    Associate a multiprocessing manager with all shared memory datasets in the catalog.

    This method iterates through all datasets in the catalog and sets the provided
    multiprocessing manager for datasets of type `SharedMemoryDataset`. This ensures
    that these datasets are properly synchronized across threads or processes.

    Args:
        manager: A multiprocessing manager to be associated with
            shared memory datasets.

    Example:
    ```python
        from multiprocessing.managers import SyncManager
        from kedro.io.data_catalog import SharedMemoryDataCatalog
        catalog = SharedMemoryDataCatalog(datasets={"shared_data": MemoryDataset(data=[1, 2, 3])})
        manager = SyncManager()
        manager.start()
        catalog.set_manager_datasets(manager)
        print(catalog)
        # {'shared_data': kedro.io.memory_dataset.MemoryDataset(data='<list>')}
    ```
    """
    for _, ds in self._datasets.items():
        if isinstance(ds, SharedMemoryDataset):
            ds.set_manager(manager)

def validate_catalog(self) -> None:
    """
    Validate the catalog to ensure all datasets are serializable and compatible with multiprocessing.

    This method checks that all datasets in the catalog are serializable and do not
    include non-proxied memory datasets as outputs. Non-serializable datasets or
    datasets that rely on single-process memory cannot be used in a multiprocessing
    context. If any such datasets are found, an exception is raised with details.

    If a non-serializable dataset appears to be backed by remote/cloud storage
    (detected via its `_protocol` attribute), a `UserWarning` is also issued for
    that dataset suggesting `ThreadRunner` or `SequentialRunner` as an
    alternative, since cloud-based filesystem clients (e.g. via `fsspec`)
    commonly cannot be pickled.

    Raises:
        AttributeError: If any datasets are found to be non-serializable or incompatible
            with multiprocessing.

    Example:
    ```python
        from kedro.io.data_catalog import SharedMemoryDataCatalog

        catalog = SharedMemoryDataCatalog(datasets={"shared_data": MemoryDataset(data=[1, 2, 3])})
        try:
            catalog.validate_catalog()
        except AttributeError as e:
            print(f"Validation failed: {e}")
        # No error
    ```
    """
    unserialisable = []
    for name, dataset in self._datasets.items():
        if getattr(dataset, "_SINGLE_PROCESS", False):  # SKIP_IF_NO_SPARK
            unserialisable.append(name)
            continue
        try:
            ForkingPickler.dumps(dataset)
        except (AttributeError, PicklingError):
            unserialisable.append(name)
            protocol = getattr(dataset, "_protocol", None)
            if protocol and protocol != "file":
                warnings.warn(
                    f"Dataset '{name}' appears to be backed by remote/cloud "
                    f"storage (protocol: '{protocol}'). Its filesystem client "
                    f"could not be pickled, which is a common issue with "
                    f"cloud-based datasets under `ParallelRunner`. Consider "
                    f"using `ThreadRunner` or `SequentialRunner` instead.",
                    stacklevel=2,
                )

    if unserialisable:
        raise AttributeError(
            f"The following datasets cannot be used with multiprocessing: "
            f"{sorted(unserialisable)}\nIn order to utilize multiprocessing you "
            f"need to make sure all datasets are serialisable, i.e. datasets "
            f"should not make use of lambda functions, nested functions, closures "
            f"etc.\nIf you are using custom decorators ensure they are correctly "
            f"decorated using functools.wraps()."
        )