fixup! Remove recursion in task spec

distributed/shuffle/_core.py

@@ -25,7 +25,7 @@
 from tornado.ioloop import IOLoop
 
 import dask.config
-from dask._task_spec import Task, _inline_recursively
+from dask._task_spec import Task
 from dask.core import flatten
 from dask.sizeof import sizeof
 from dask.typing import Key
@@ -609,15 +609,6 @@ def __sizeof__(self) -> int:
     def __repr__(self) -> str:
         return f"P2PBarrierTask({self.key!r})"
 
-    def inline(self, dsk: dict[Key, Any]) -> P2PBarrierTask:
-        self.unpack()
-        new_args = _inline_recursively(self.args, dsk)
-        new_kwargs = _inline_recursively(self.kwargs, dsk)
-        assert self.func is not None
-        return P2PBarrierTask(
-            self.key, self.func, *new_args, spec=self.spec, **new_kwargs
-        )
-
     def __getstate__(self) -> dict[str, Any]:
         state = super().__getstate__()
         state["spec"] = self.spec