Repository: dask/dask
File: dask/_task_spec.py (Python)

Code before:
            raise RuntimeError(
                f"Error during serialization of function of task {self.key}"
            ) from exc
        try:
            _func_cache_reverse[self.packed_func], _func_cache[self.func] = (
                self.func,
                self.packed_func,
            )
        except TypeError:
            pass
        self.func = None
        return self

    def lazy_pack(self):
        # TODO: This could dispatch to a TPE and de/serialize the arguments
        # lazily such that pack only waits for the result. This way we can use
        # spare CPU cycles and parallelize on multiple CPUs.
        # Thread safety should not be an issue if we guarantee idempotency and
        # that the non-lazy sync also uses this
        # This may block the GIL pretty aggressively
        raise NotImplementedError("Not implemented")

    def lazy_unpack(self):
        raise NotImplementedError("Not implemented")

    def inline(self, dsk) -> Task:
        self.unpack()
        new_args = _inline_recursively(self.args, dsk)
        new_kwargs = _inline_recursively(self.kwargs, dsk)
        assert self.func is not None
        return Task(self.key, self.func, *new_args, **new_kwargs)

    def unpack(self):
        if not self.is_packed():
            return
        assert self.packed_func is not None

        try:
            from distributed.protocol.pickle import loads
        except ImportError:
            from cloudpickle import loads
        try:
            self.func = _func_cache_reverse[self.packed_func]
            self.packed_func = None
            return self
        except KeyError:
            # We're not handling the below in this except to simplify traceback
            # and cause
            pass
        try:
            self.func = loads(self.packed_func)

Review comment (suggestion):
Inline isn't used anywhere anymore. I thought about killing it off entirely but it's also an interesting feature. I'm fine with keeping it for a little while (I assume it is useful if we actually migrated low level fusion) but we should kill it if it is being unused

Code after:
            ) from exc
        assert self.func is not None
        new_argspec = _call_recursively(self.args, values)
        if self.kwargs:
            kwargs = _call_recursively(self.kwargs, values)
            return self.func(*new_argspec, **kwargs)
        return self.func(*new_argspec)

    def pack(self):
        # TODO: pack args and kwargs as well. Probably with a sizeof threshold
        if self.is_packed():
            return self
        try:
            from distributed.protocol.pickle import dumps
        except ImportError:
            from cloudpickle import dumps

        try:
            self.packed_func = _func_cache[self.func]
            self.func = None
            return self
        except (KeyError, TypeError):
            # We're not handling the below in this except to simplify traceback
            # and cause
            pass
        try:
            self.packed_func = dumps(self.func)
        except Exception as exc:
            raise RuntimeError(
                f"Error during serialization of function of task {self.key}"
            ) from exc
        try:
            _func_cache_reverse[self.packed_func], _func_cache[self.func] = (
                self.func,
                self.packed_func,
            )
        except TypeError:
            pass
        self.func = None
        return self

    def lazy_pack(self):
        # TODO: This could dispatch to a TPE and de/serialize the arguments
        # lazily such that pack only waits for the result. This way we can use
        # spare CPU cycles and parallelize on multiple CPUs.
        # Thread safety should not be an issue if we guarantee idempotency and
        # that the non-lazy sync also uses this
        # This may block the GIL pretty aggressively
        raise NotImplementedError("Not implemented")

    def lazy_unpack(self):