Repository: dask/dask
Pull request #11378 (2024-09-06 15:49:39)

Use TaskSpec in local dask execution
This effectively causes all dask/dask schedulers to convert to the new task spec and use it as an execution backend.

Todos
- ~Haven't done a sanity perf check but if there is something slow, it is likely because it is still using tuples as tasks (the parsing has some overhead)~ I believe that these kinds of tests are currently premature. I also haven't seen anything significant happening when running distributed tests and expect performance to improve in follow up PRs either way
- the module structure is a little messed up. For now I put in lazy imports but it would be nice if that wasn't necessary everywhere. Might be not as bad as it looks but I haven't spent the time to clean this up properly.
- (Follow up) Migrate the internals to use TaskSpec. A couple of heavy hitters like Blockwise would already go a long way. The more we migrate, the less overhead the converter has.

Note
- I am removing a non-underscored function from `dask.core`. That's _technically_ a breaking change. However, I would like to propose to make everything in dask.core private. It's effectively a second utils package. We might want to kill the entire module since it's really confusing since there is also dask.base which is *much* more meaningful and contains more valuable user facing stuff.

xref https://github.com/dask/dask/issues/9969

github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

     13 files  ± 0       13 suites  ±0   3h 40m 6s :stopwatch: + 33m 33s
 13 225 tests + 3   12 153 :white_check_mark:  -  13   1 056 :zzz: ±0   16 :x: + 16 
137 949 runs  +39  118 747 :white_check_mark:  - 137  19 024 :zzz:  - 2  178 :x: +178 

For more details on these failures, see [this check](https://github.com/dask/dask/runs/29793186893).

Results for commit c94ea457. ± Comparison against base commit 863a049c.

<details>
  <summary>This pull request <b>removes</b> 5 and <b>adds</b> 8 tests. <i>Note that renamed tests count towards both.</i></summary>

```
dask.tests.test_core ‑ test_has_tasks
dask.tests.test_order ‑ test_base_of_reduce_preferred[abcde]
dask.tests.test_order ‑ test_base_of_reduce_preferred[edcba]
dask.tests.test_task_spec ‑ test_ensure_slots[inst3]
dask.tests.test_task_spec ‑ test_ensure_slots[inst4]
```

```
dask.tests.test_order ‑ test_base_of_reduce_preferred[abcde-False]
dask.tests.test_order ‑ test_base_of_reduce_preferred[abcde-True]
dask.tests.test_order ‑ test_base_of_reduce_preferred[edcba-False]
dask.tests.test_order ‑ test_base_of_reduce_preferred[edcba-True]
dask.tests.test_order ‑ test_handle_out_of_graph_dependencies
dask.tests.test_task_spec ‑ test_convert_tasks_only_ref_lowlevel
dask.tests.test_task_spec ‑ test_execute_tasks_in_graph
dask.tests.test_task_spec ‑ test_task_nested_sequence
```
</details>

[test-results]:data:application/gzip;base64,H4sIAIIt22YC/03Myw6DIBCF4VcxrLsYuQj0ZcyIkJB6aRBWpu9erIJd/t9kzk6cn+xGnk3LHg3Zko93jSlg9OvyawpdpnyN551SUbrfkjEH0lawG1/+fSCI+tg79NNht9gQ1pAJsoS0nNtSc3113W6V5LLita2B8mJlW6oi/9tmnWcfcxCjuUUuJAxonEIAizAYodAyrkQ3oBsRNDDy+QISZYNSHQEAAA==

fjetter: this is ready to merge. There are a few subtle breaking changes. However, they are so deeply burried that I do not believe a proper deprecation cycle is required.

For example:

- The Profiler class (which I never used and only vaguely remember even knowing about) returns the new task spec instead of the tuples regardless of what the user put in
- The function `dask.core.has_tasks` is gone
- Callbacks (I actually didn't know this exists) now also receive the new task spec instead of the tuple

The documentation about our task spec has not been updated yet, see https://docs.dask.org/en/stable/spec.html At this point, this change is entirely internal and we're fully backwards compatible so I don't consider it necessary to have this page updated with this PR (but I'll follow up)
fjetter: The one test failure is https://github.com/dask/dask/issues/11460

Patch:
diff --git a/dask/_task_spec.py b/dask/_task_spec.py
index 3b563aced..e02e6cbb1 100644
--- a/dask/_task_spec.py
+++ b/dask/_task_spec.py
@@ -84,7 +84,7 @@ from typing import Any, TypeVar, cast, overload
 
 from dask.sizeof import sizeof
 from dask.typing import Key as KeyType
-from dask.utils import is_namedtuple_instance
+from dask.utils import funcname, is_namedtuple_instance
 
 try:
     from distributed.collections import LRU
@@ -483,7 +483,7 @@ class Alias(GraphNode):
             return self
 
     def __repr__(self):
-        return f"Alias(key={self.key}, target={self.target})"
+        return f"Alias({self.key}->{self.target})"
 
     def __eq__(self, value: object) -> bool:
         if not isinstance(value, Alias):
@@ -587,6 +587,7 @@ class Task(GraphNode):
     packed_func: None | bytes
     _token: str | None
     _is_coro: bool | None
+    _repr: str | None
 
     __slots__ = tuple(__annotations__)
 
@@ -612,6 +613,7 @@ class Task(GraphNode):
             self._dependencies = _no_deps
         self._is_coro = None
         self._token = None
+        self._repr = None
 
     def copy(self):
         self.unpack()
@@ -651,7 +653,36 @@ class Task(GraphNode):
         )
 
     def __repr__(self) -> str:
-        return f"Task({self.key!r})"
+        if not self._repr:
+            self.unpack()
+            head = funcname(self.func)
+            tail = ")"
+            label_size = 40
+            args = self.args
+            kwargs = self.kwargs
+            if args or kwargs:
+                label_size2 = int(
+                    (label_size - len(head) - len(tail) - len(str(self.key)))
+                    // (len(args) + len(kwargs))
+                )
+            if args:
+                if label_size2 > 5:
+                    args_repr = ", ".join(repr(t) for t in args)
+                else:
+                    args_repr = "..."
+            else:
+                args_repr = ""
+            if kwargs:
+                if label_size2 > 5:
+                    kwargs_repr = ", " + ", ".join(
+                        f"{k}={repr(v)}" for k, v in sorted(kwargs.items())
+                    )
+                else:
+                    kwargs_repr = ", ..."
+            else:
+                kwargs_repr = ""
+            self._repr = f"<Task {self.key!r} {head}({args_repr}{kwargs_repr}{tail}>"
+        return self._repr
 
     def __call__(self, values=()):
         self._verify_values(values)
@@ -871,19 +902,33 @@ def no_function_cache():
         _func_cache, _func_cache_reverse = cache_before
 
 
-def execute_graph(dsk: list[GraphNode] | dict[KeyType, GraphNode]) -> dict:
+def execute_graph(
+    dsk: Iterable[GraphNode] | Mapping[KeyType, GraphNode],
+    cache: MutableMapping[KeyType, object] | None = None,
+    keys: Container[KeyType] | None = None,
+) -> MutableMapping[KeyType, object]:
+    """Execute a given graph.
+
+    The graph is exceuted in topological order as defined by dask.order until
+    all leaf nodes, i.e. nodes without any dependents, are reached. The returned
+    dictionary contains the results of the leaf nodes.
+
+    If keys are required that are not part of the graph, they can be provided in the `cache` argument.
+
+    If `keys` is provided, the result will contain only values that are part of the `keys` set.
+
+    """
     if isinstance(dsk, (list, tuple, set, frozenset)):
         dsk = {t.key: t for t in dsk}
     else:
         assert isinstance(dsk, dict)
-    dependencies = dict(DependenciesMapping(dsk))
 
     refcount: defaultdict[KeyType, int] = defaultdict(int)
-    for vals in dependencies.values():
+    for vals in DependenciesMapping(dsk).values():
         for val in vals:
             refcount[val] += 1
 
-    cache: dict[KeyType, object] = {}
+    cache = cache or {}
     from dask.order import order
 
     priorities = order(dsk)
@@ -892,7 +937,7 @@ def execute_graph(dsk: list[GraphNode] | dict[KeyType, GraphNode]) -> dict:
         cache[key] = node(cache)
         for dep in node.dependencies:
             refcount[dep] -= 1
-            if refcount[dep] == 0:
+            if refcount[dep] == 0 and keys and dep not in keys:
                 del cache[dep]
 
     return cache
diff --git a/dask/core.py b/dask/core.py
index afe799207..3cfd3c3a1 100644
--- a/dask/core.py
+++ b/dask/core.py
@@ -1,10 +1,10 @@
 from __future__ import annotations
 
 from collections import defaultdict
-from collections.abc import Collection, Iterable, Mapping
+from collections.abc import Collection, Iterable, Mapping, MutableMapping
 from typing import Any, Literal, TypeVar, cast, overload
 
-from dask._task_spec import DependenciesMapping
+from dask._task_spec import DependenciesMapping, convert_legacy_graph, execute_graph
 from dask.typing import Graph, Key, NoDefault, no_default
 
 
@@ -53,28 +53,6 @@ def istask(x):
     return type(x) is tuple and x and callable(x[0])
 
 
-def has_tasks(dsk, x):
-    """Whether ``x`` has anything to compute.
-
-    Returns True if:
-    - ``x`` is a task
-    - ``x`` is a key in ``dsk``
-    - ``x`` is a list that contains any tasks or keys
-    """
-    if istask(x):
-        return True
-    try:
-        if x in dsk:
-            return True
-    except Exception:
-        pass
-    if isinstance(x, list):
-        for i in x:
-            if has_tasks(dsk, i):
-                return True
-    return False
-
-
 def preorder_traversal(task):
     """A generator to preorder-traverse a task."""
 
@@ -94,53 +72,13 @@ def lists_to_tuples(res, keys):
     return res
 
 
-def _execute_task(arg, cache, dsk=None):
-    """Do the actual work of collecting data and executing a function
-
-    Examples
-    --------
-
-    >>> inc = lambda x: x + 1
-    >>> add = lambda x, y: x + y
-    >>> cache = {'x': 1, 'y': 2}
-
-    Compute tasks against a cache
-    >>> _execute_task((add, 'x', 1), cache)  # Compute task in naive manner
-    2
-    >>> _execute_task((add, (inc, 'x'), 1), cache)  # Support nested computation
-    3
-
-    Also grab data from cache
-    >>> _execute_task('x', cache)
-    1
-
-    Support nested lists
-    >>> list(_execute_task(['x', 'y'], cache))
-    [1, 2]
-
-    >>> list(map(list, _execute_task([['x', 'y'], ['y', 'x']], cache)))
-    [[1, 2], [2, 1]]
-
-    >>> _execute_task('foo', cache)  # Passes through on non-keys
-    'foo'
-    """
-    if isinstance(arg, list):
-        return [_execute_task(a, cache) for a in arg]
-    elif istask(arg):
-        func, args = arg[0], arg[1:]
-        # Note: Don't assign the subtask results to a variable. numpy detects
-        # temporaries by their reference count and can execute certain
-        # operations in-place.
-        return func(*(_execute_task(a, cache) for a in args))
-    elif not ishashable(arg):
-        return arg
-    elif arg in cache:
-        return cache[arg]
-    else:
-        return arg
+def _pack_result(result: Mapping, keys: list | Key) -> Any:
+    if isinstance(keys, list):
+        return tuple(_pack_result(result, k) for k in keys)
+    return result[keys]
 
 
-def get(dsk, out, cache=None):
+def get(dsk: Mapping, out: list | Key, cache: MutableMapping | None = None) -> Any:
     """Get value from Dask
 
     Examples
@@ -154,19 +92,15 @@ def get(dsk, out, cache=None):
     >>> get(d, 'y')
     2
     """
-    for k in flatten(out) if isinstance(out, list) else [out]:
+    for k in flatten(out):
         if k not in dsk:
             raise KeyError(f"{k} is not a key in the graph")
     if cache is None:
         cache = {}
-    for key in toposort(dsk):
-        task = dsk[key]
-        result = _execute_task(task, cache)
-        cache[key] = result
-    result = _execute_task(out, cache)
-    if isinstance(out, list):
-        result = lists_to_tuples(result, out)
-    return result
+
+    dsk2 = convert_legacy_graph(dsk, all_keys=set(dsk) | set(cache))
+    result = execute_graph(dsk2, cache, keys=set(flatten([out])))
+    return _pack_result(result, out)
 
 
 def keys_in_tasks(keys: Collection[Key], tasks: Iterable[Any], as_list: bool = False):
diff --git a/dask/dask.yaml b/dask/dask.yaml
index ce6b93cfe..8da587e7a 100644
--- a/dask/dask.yaml
+++ b/dask/dask.yaml
@@ -51,6 +51,7 @@ admin:
       - dask[\\\/](base|core|local|multiprocessing|optimization|threaded|utils)\.py
       - dask[\\\/]array[\\\/]core\.py
       - dask[\\\/]dataframe[\\\/](core|methods)\.py
+      - dask[\\\/]_task_spec\.py
       - distributed[\\\/](client|scheduler|utils|worker)\.py
       - tornado[\\\/]gen\.py
       - pandas[\\\/]core[\\\/]
diff --git a/dask/diagnostics/profile_visualize.py b/dask/diagnostics/profile_visualize.py
index 32d3f37cc..98c56e25e 100644
--- a/dask/diagnostics/profile_visualize.py
+++ b/dask/diagnostics/profile_visualize.py
@@ -10,7 +10,7 @@ from tlz import accumulate, groupby, pluck, unique
 
 from dask._compatibility import import_optional_dependency
 from dask.core import istask
-from dask.utils import apply, funcname, import_required
+from dask.utils import import_required
 
 
 def BOKEH_VERSION():
@@ -36,98 +36,6 @@ def unquote(expr):
     return expr
 
 
-def pprint_task(task, keys, label_size=60):
-    """Return a nicely formatted string for a task.
-
-    Parameters
-    ----------
-    task:
-        Value within dask graph to render as text
-    keys: iterable
-        List of keys within dask graph
-    label_size: int (optional)
-        Maximum size of output label, defaults to 60
-
-    Examples
-    --------
-    >>> from operator import add, mul
-    >>> dsk = {'a': 1,
-    ...        'b': 2,
-    ...        'c': (add, 'a', 'b'),
-    ...        'd': (add, (mul, 'a', 'b'), 'c'),
-    ...        'e': (sum, ['a', 'b', 5]),
-    ...        'f': (add,),
-    ...        'g': []}
-
-    >>> pprint_task(dsk['c'], dsk)
-    'add(_, _)'
-    >>> pprint_task(dsk['d'], dsk)
-    'add(mul(_, _), _)'
-    >>> pprint_task(dsk['e'], dsk)
-    'sum([_, _, *])'
-    >>> pprint_task(dsk['f'], dsk)
-    'add()'
-    >>> pprint_task(dsk['g'], dsk)
-    '[]'
-    """
-    if istask(task):
-        func = task[0]
-        if func is apply:
-            head = funcname(task[1])
-            tail = ")"
-            args = unquote(task[2]) if len(task) > 2 else ()
-            kwargs = unquote(task[3]) if len(task) > 3 else {}
-        else:
-            if hasattr(func, "funcs"):
-                head = "(".join(funcname(f) for f in func.funcs)
-                tail = ")" * len(func.funcs)
-            else:
-                head = funcname(task[0])
-                tail = ")"
-            args = task[1:]
-            kwargs = {}
-        if args or kwargs:
-            label_size2 = int(
-                (label_size - len(head) - len(tail)) // (len(args) + len(kwargs))
-            )
-            pprint = lambda t: pprint_task(t, keys, label_size2)
-        if args:
-            if label_size2 > 5:
-                args = ", ".join(pprint(t) for t in args)
-            else:
-                args = "..."
-        else:
-            args = ""
-        if kwargs:
-            if label_size2 > 5:
-                kwargs = ", " + ", ".join(
-                    f"{k}={pprint(v)}" for k, v in sorted(kwargs.items())
-                )
-            else:
-                kwargs = ", ..."
-        else:
-            kwargs = ""
-        return f"{head}({args}{kwargs}{tail}"
-    elif isinstance(task, list):
-        if not task:
-            return "[]"
-        elif len(task) > 3:
-            result = pprint_task(task[:3], keys, label_size)
-            return result[:-1] + ", ...]"
-        else:
-            label_size2 = int((label_size - 2 - 2 * len(task)) // len(task))
-            args = ", ".join(pprint_task(t, keys, label_size2) for t in task)
-            return f"[{args}]"
-    else:
-        try:
-            if task in keys:
-                return "_"
-            else:
-                return "*"
-        except TypeError:
-            return "*"
-
-
 def get_colors(palette, funcs):
     """Get a dict mapping funcs to colors from palette.
 
@@ -297,7 +205,7 @@ def plot_tasks(
         data["width"] = width = [e - s for (s, e) in zip(starts, ends)]
         data["x"] = [w / 2 + s - start_time for (w, s) in zip(width, starts)]
         data["y"] = [id_lk[i] + 1 for i in ids]
-        data["function"] = funcs = [pprint_task(i, dsk, label_size) for i in tasks]
+        data["function"] = funcs = [repr(i) for i in tasks]
         data["color"] = get_colors(palette, funcs)
         data["key"] = [str(i) for i in keys]
 
@@ -491,7 +399,7 @@ def plot_cache(
     if results:
         starts, ends = list(zip(*results))[3:]
         tics = sorted(unique(starts + ends))
-        groups = groupby(lambda d: pprint_task(d[1], dsk, label_size), results)
+        groups = groupby(lambda d: repr(d), results)
         data = {}
         for k, vals in groups.items():
             cnts = dict.fromkeys(tics, 0)
diff --git a/dask/local.py b/dask/local.py
index b00af3ae7..8a8ecaa42 100644
--- a/dask/local.py
+++ b/dask/local.py
@@ -116,8 +116,9 @@ from functools import partial
 from queue import Empty, Queue
 
 from dask import config
+from dask._task_spec import DataNode, DependenciesMapping, convert_legacy_graph
 from dask.callbacks import local_callbacks, unpack_callbacks
-from dask.core import _execute_task, flatten, get_dependencies, has_tasks, reverse_dict
+from dask.core import flatten, get_dependencies, reverse_dict
 from dask.order import order
 from dask.typing import Key
 
@@ -165,17 +166,19 @@ def start_state_from_dask(dsk, cache=None, sortkey=None):
         cache = config.get("cache", None)
     if cache is None:
         cache = dict()
+
+    dsk = convert_legacy_graph(dsk, all_keys=set(dsk) | set(cache))
     data_keys = set()
     for k, v in dsk.items():
-        if not has_tasks(dsk, v):
-            cache[k] = v
+        if isinstance(v, DataNode):
+            cache[k] = v()
             data_keys.add(k)
 
     dsk2 = dsk.copy()
     dsk2.update(cache)
 
-    dependencies = {k: get_dependencies(dsk2, k) for k in dsk}
-    waiting = {k: v.copy() for k, v in dependencies.items() if k not in data_keys}
+    dependencies = DependenciesMapping(dsk)
+    waiting = {k: set(v) for k, v in dependencies.items() if k not in data_keys}
 
     dependents = reverse_dict(dependencies)
     for a in cache:
@@ -223,7 +226,7 @@ def execute_task(key, task_info, dumps, loads, get_id, pack_exception):
     """
     try:
         task, data = loads(task_info)
-        result = _execute_task(task, data)
+        result = task(data)
         id = get_id()
         result = dumps((result, id))
         failed = False
@@ -423,7 +426,7 @@ def get_async(
         result_flat = {result}
     results = set(result_flat)
 
-    dsk = dict(dsk)
+    dsk = dict(convert_legacy_graph(dsk))
     with local_callbacks(callbacks) as callbacks:
         _, _, pretask_cbs, posttask_cbs, _ = unpack_callbacks(callbacks)
         started_cbs = []
@@ -508,7 +511,7 @@ def get_async(
                                 for dep in get_dependencies(dsk, key)
                             }
                             task = dsk[key]
-                            _execute_task(task, data)  # Re-execute locally
+                            task(data)  # Re-execute locally
                         else:
                             raise_exception(exc, tb)
                     res, worker_id = loads(res_info)


Test patch:
diff --git a/dask/diagnostics/tests/test_profiler.py b/dask/diagnostics/tests/test_profiler.py
index b57ebf082..80b83bbc3 100644
--- a/dask/diagnostics/tests/test_profiler.py
+++ b/dask/diagnostics/tests/test_profiler.py
@@ -8,10 +8,11 @@ from timeit import default_timer
 
 import pytest
 
+from dask._task_spec import Task
 from dask.diagnostics import CacheProfiler, Profiler, ResourceProfiler
 from dask.diagnostics.profile_visualize import BOKEH_VERSION
 from dask.threaded import get
-from dask.utils import apply, tmpfile
+from dask.utils import tmpfile
 from dask.utils_test import slowadd
 
 try:
@@ -40,7 +41,8 @@ def test_profiler():
     keys = [i.key for i in prof_data]
     assert keys == ["c", "d", "e"]
     tasks = [i.task for i in prof_data]
-    assert tasks == [(add, "a", "b"), (mul, "a", "b"), (mul, "c", "d")]
+    assert len(tasks) == 3
+    assert all(isinstance(t, Task) for t in tasks)
     prof.clear()
     assert prof.results == []
 
@@ -192,38 +194,6 @@ def test_unquote():
     assert unquote(task_dask) == t
 
 
-@pytest.mark.skipif("not bokeh")
-def test_pprint_task():
-    from dask.diagnostics.profile_visualize import pprint_task
-
-    keys = {"a", "b", "c", "d", "e"}
-    assert pprint_task((add, "a", 1), keys) == "add(_, *)"
-    assert pprint_task((add, (add, "a", 1)), keys) == "add(add(_, *))"
-    res = "sum([*, _, add(_, *)])"
-    assert pprint_task((sum, [1, "b", (add, "a", 1)]), keys) == res
-    assert pprint_task((sum, (1, 2, 3, 4, 5, 6, 7)), keys) == "sum(*)"
-
-    assert len(pprint_task((sum, list(keys) * 100), keys)) < 100
-    assert pprint_task((sum, list(keys) * 100), keys) == "sum([_, _, _, ...])"
-    assert (
-        pprint_task((sum, [1, 2, (sum, ["a", 4]), 5, 6] * 100), keys)
-        == "sum([*, *, sum([_, *]), ...])"
-    )
-    assert (
-        pprint_task((sum, [1, 2, (sum, ["a", (sum, [1, 2, 3])]), 5, 6]), keys)
-        == "sum([*, *, sum([_, sum(...)]), ...])"
-    )
-
-    # With kwargs
-    def foo(w, x, y=(), z=3):
-        return w + x + sum(y) + z
-
-    task = (apply, foo, (tuple, ["a", "b"]), (dict, [["y", ["a", "b"]], ["z", "c"]]))
-    assert pprint_task(task, keys) == "foo(_, _, y=[_, _], z=_)"
-    task = (apply, foo, (tuple, ["a", "b"]), (dict, [["y", ["a", 1]], ["z", 1]]))
-    assert pprint_task(task, keys) == "foo(_, _, y=[_, *], z=*)"
-
-
 @pytest.mark.skipif("not bokeh")
 def test_profiler_plot():
     with prof:
diff --git a/dask/tests/test_callbacks.py b/dask/tests/test_callbacks.py
index 08f5a20c8..dd29121b8 100644
--- a/dask/tests/test_callbacks.py
+++ b/dask/tests/test_callbacks.py
@@ -27,7 +27,7 @@ def test_start_state_callback():
     class MyCallback(Callback):
         def _start_state(self, dsk, state):
             flag[0] = True
-            assert dsk["x"] == 1
+            assert dsk["x"]() == 1
             assert len(state["cache"]) == 1
 
     with MyCallback():
@@ -98,8 +98,8 @@ def test_nested_schedulers():
         get_threaded(outer_dsk, "b")
 
     assert not Callback.active
-    assert outer_callback.dsk == outer_dsk
-    assert inner_callback.dsk == inner_dsk
+    assert outer_callback.dsk.keys() == outer_dsk.keys()
+    assert inner_callback.dsk.keys() == inner_dsk.keys()
     assert not Callback.active
 
 
diff --git a/dask/tests/test_core.py b/dask/tests/test_core.py
index 41b42b44d..67023bd66 100644
--- a/dask/tests/test_core.py
+++ b/dask/tests/test_core.py
@@ -11,7 +11,6 @@ from dask.core import (
     get_dependencies,
     get_deps,
     getcycle,
-    has_tasks,
     ishashable,
     iskey,
     istask,
@@ -88,23 +87,6 @@ def test_istask():
     assert not istask(f(sum, 2))
 
 
-def test_has_tasks():
-    dsk = {
-        "a": [1, 2, 3],
-        "b": "a",
-        "c": [1, (inc, 1)],
-        "d": [(sum, "a")],
-        "e": ["a", "b"],
-        "f": [["a", "b"], 2, 3],
-    }
-    assert not has_tasks(dsk, dsk["a"])
-    assert has_tasks(dsk, dsk["b"])
-    assert has_tasks(dsk, dsk["c"])
-    assert has_tasks(dsk, dsk["d"])
-    assert has_tasks(dsk, dsk["e"])
-    assert has_tasks(dsk, dsk["f"])
-
-
 def test_preorder_traversal():
     t = (add, 1, 2)
     assert list(preorder_traversal(t)) == [add, 1, 2]
diff --git a/dask/tests/test_task_spec.py b/dask/tests/test_task_spec.py
index 46c3e6974..0604d95aa 100644
--- a/dask/tests/test_task_spec.py
+++ b/dask/tests/test_task_spec.py
@@ -62,6 +62,28 @@ def test_convert_legacy_dsk_skip_new():
     assert converted == dsk
 
 
+def test_repr():
+    t = Task("key", func, "a", "b")
+    assert repr(t) == "<Task 'key' func('a', 'b')>"
+
+    t = Task("nested", func2, t, t.ref())
+    assert (
+        repr(t) == "<Task 'nested' func2(<Task 'key' func('a', 'b')>, Alias(key->key))>"
+    )
+
+    def long_function_name_longer_even_longer(a, b):
+        return a + b
+
+    t = Task("long", long_function_name_longer_even_longer, t, t.ref())
+    assert repr(t) == "<Task 'long' long_function_name_longer_even_longer(...)>"
+
+    def use_kwargs(a, kwarg=None):
+        return a + kwarg
+
+    t = Task("kwarg", use_kwargs, "foo", kwarg="kwarg_value")
+    assert repr(t) == "<Task 'kwarg' use_kwargs('foo', kwarg='kwarg_value')>"
+
+
 def _assert_dsk_conversion(new_dsk):
     vals = [(k, v.key) for k, v in new_dsk.items()]
     assert all(v[0] == v[1] for v in vals), vals
@@ -754,7 +776,7 @@ def test_execute_tasks_in_graph():
         t3 := Task("key-3", func, "foo", "bar"),
         Task("key-4", func, t3.ref(), t2.ref()),
     ]
-    res = execute_graph(dsk)
+    res = execute_graph(dsk, keys=["key-4"])
     assert len(res) == 1
     assert res["key-4"] == "foo-bar-a-b=c"