def distributed_pack(self) -> Optional[Any]:
        """Pack the layer for scheduler communication in Distributed

        This method should pack its current state and is called by the Client when
        communicating with the Scheduler.
        The Scheduler will then use .distributed_unpack(data, ...) to unpack the
        state, materialize the layer, and merge it into the global task graph.

        The returned state must be compatible with Distributed's scheduler, which
        means it must obey the following:
          - Serializable by msgpack
          - All remote data must be unpacked (see unpack_remotedata())
          - All keys must be converted to strings now or when unpacking
          - All tasks must be serialized (see dumps_task())

        Alternatively, the method can return None, which will make Distributed
        materialize the layer and use a default packing method.

        Returns
        -------
        state: Object serializable by msgpack
            Scheduler compatible state of the layer
        """
        return None

    @classmethod
    def distributed_unpack(
        cls, state: Any, dsk: Dict[str, Any], dependencies: Mapping[Hashable, Set]
    ) -> None:
        """Unpack the state of a layer previously packed by .distributed_pack()

        This method is called by the scheduler in Distributed in order to unpack
        the state of a layer and merge it into its global task graph. The method
        should update `dsk` and `dependencies`, which are the already materialized
        state of the preceding layers in the high level graph. The layers of the
        high level graph are unpacked in topological order.

        See Layer.distributed_pack() for packing detail.

        Parameters
        ----------
        state: Any
            The state returned by Layer.distributed_pack()
        dsk: dict
            The materialized low level graph of the already unpacked layers
        dependencies: Mapping
            The dependencies of each key in `dsk`
        """
        raise NotImplementedError(f"{type(cls)} doesn't implement distributed_unpack()")