Repository: dask/dask
Pull request: Redo `dask.order.order`.  Fix #5584.  Use structural info, not key names
Closes #5584 

This is a rewrite of `dask.order.order`, but the goals remain the same and many previous lessons were taken into consideration.  The new version relies less on the key name by using more metrics and using a different strategy for walking up and down the DAG.

Performance appears to be about the same (often a little faster).

@TomAugspurger also suggested I add some benchmarks to `dask-benchmarks`.

- [x] Tests added / passed
- [x] Passes `black dask` / `flake8 dask`


Problem statement:
Tasks order and scheduling
First of all I apologies because this is more a question than an issue or feature request (though it might turn to one of those in the end).

I am having a really hard time to figure out how to influence task scheduling to get it the way I want it.

Here is my setup. I have two set of tasks, say A and B. Each set has a few thousands tasks in it. A tasks depends on nothing. Each B task depends on a few A tasks output (5 in average).

In my code I delay all A tasks, then I delay B tasks using delayed A objects to model dependencies.

Then I call `client.compute(B_delayed)` and I retrieve results of B on master node on the flow, using `as_completed` (client is from `dask.distributed`).

I would expect the scheduler to figure out that each B task has very few dependencies to A tasks, and therefore to try to complete entire B tasks to be able to clear ressources for A tasks that are no longer needed.

What happens is quite the opposite : the scheduler prioritize A tasks, trying to complete almost all of them while very scarcely computing B tasks (even when a lot of them become ready to run).

This is problematic for us:
- On the flow retrieval is less efficient if all B tasks come at the end (it involves IO operations)
- In some cases, the amount of memory on workers is insufficient to store all A results. This memory could be easily cleared, if only the scheduler decided to complete the corresponding B tasks.

What I tried so far:
- Also submit A tasks with `client.compute()`, with a lower priority (did not change the behavior)
- Submit B tasks in batch with decreasing priority (it helps a bit, but scheduler still wants to complete as many A tasks as possible).
- Mix of both : no noticeable progress

Hints from issue discussion:
This page may interest you:
https://distributed.dask.org/en/latest/priority.html

In general though if you are submitting a single graph with `dask.compute`
then I would expect Dask to choose a prioritization that minimized memory
footprint.  It's heuristics are usually pretty good at this.  If you have
task graphs that are only a few hundred nodes then you can visualize the
prioritization by calling `dask.visualize(..., color="order")`

On Wed, Nov 13, 2019 at 3:21 AM Julien Michel <notifications@github.com>
wrote:

> First of all I apologies because this is more a question than an issue or
> feature request (though it might turn to one of those in the end).
>
> I am having a really hard time to figure out how to influence task
> scheduling to get it the way I want it.
>
> Here is my setup. I have two set of tasks, say A and B. Each set has a few
> thousands tasks in it. A tasks depends on nothing. Each B task depends on a
> few A tasks output (5 in average).
>
> In my code I delay all A tasks, then I delay B tasks using delayed A
> objects to model dependencies.
>
> Then I call client.compute(B_delayed) and I retrieve results of B on
> master node on the flow, using as_completed (client is from
> dask.distributed).
>
> I would expect the scheduler to figure out that each B task has very few
> dependencies to A tasks, and therefore to try to complete entire B tasks to
> be able to clear ressources for A tasks that are no longer needed.
>
> What happens is quite the opposite : the scheduler prioritize A tasks,
> trying to complete almost all of them while very scarcely computing B tasks
> (even when a lot of them become ready to run).
>
> This is problematic for us:
>
>    - On the flow retrieval is less efficient if all B tasks come at the
>    end (it involves IO operations)
>    - In some cases, the amount of memory on workers is insufficient to
>    store all A results. This memory could be easily cleared, if only the
>    scheduler decided to complete the corresponding B tasks.
>
> What I tried so far:
>
>    - Also submit A tasks with client.compute(), with a lower priority
>    (did not change the behavior)
>    - Submit B tasks in batch with decreasing priority (it helps a bit,
>    but scheduler still wants to complete as many A tasks as possible).
>    - Mix of both : no noticeable progress
>
> —
> You are receiving this because you are subscribed to this thread.
> Reply to this email directly, view it on GitHub
> <https://github.com/dask/dask/issues/5584?email_source=notifications&email_token=AACKZTCKOOWLA2SFKHWJYILQTPPL7A5CNFSM4JMZQD7KYY3PNVWWK3TUL52HS4DFUVEXG43VMWVGG33NNVSW45C7NFSM4HY7HRJA>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AACKZTC2N647NRAI5OGCQGDQTPPL7ANCNFSM4JMZQD7A>
> .
>

Thanks for your answer. I submit work with a single compute on a list of delayed B. One thing to note is that A code calls C/C++ code under the hood, so dask may not be aware of all the memory used. 

I have been able to visualize the graph on a subset my problem, and I can confirm that dask mostly prioritize A tasks over B tasks (I can share the graph plot if that helps).
Sure

On Thu, Nov 14, 2019 at 12:25 AM Julien Michel <notifications@github.com>
wrote:

> Thanks for your answer. I submit work with a single compute on a list of
> delayed B. One thing to note is that A code calls C/C++ code under the
> hood, so dask may not be aware of all the memory used.
>
> I have been able to visualize the graph on a subset my problem, and I can
> confirm that dask mostly prioritize A tasks over B tasks (I can share the
> graph plot if that helps).
>
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub
> <https://github.com/dask/dask/issues/5584?email_source=notifications&email_token=AACKZTGWU7L3HPV2O4UQIOTQTUDOHA5CNFSM4JMZQD7KYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEEA77KQ#issuecomment-553779114>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AACKZTCUCMG3RWYH5OIQLSLQTUDOHANCNFSM4JMZQD7A>
> .
>

So this is a code sample simulating what I am actually doing:

```python
import xarray as xr
import numpy as np
import dask
from dask.distributed import as_completed
from time import sleep

task_length = 2
a_size = 1000

# First task in graph
def A(value):
    coord = np.array(range(0,a_size))
    buffer = np.ndarray(shape=(a_size,a_size),dtype=float)
    buffer.fill(value)

    output = xr.Dataset({'a' : (['c1','c2'] , buffer),
                         'b' : (['c1','c2'] , buffer),
                         'c' : (['c1','c2'] , buffer)},
                        coords = {'c1' : coord, 'c2' : coord})
    sleep(task_length)
    return output

b_size = 500

# second task in graph
def B(inputs,value):
    coord = np.array(range(0,b_size))
    buffer = np.ndarray(shape=(b_size,b_size),dtype=float)
    buffer.fill(value)

    output = xr.Dataset({'a' : (['c1','c2'] , buffer),
                         'b' : (['c1','c2'] , buffer),
                         'c' : (['c1','c2'] , buffer)},
                        coords = {'c1' : coord, 'c2' : coord})

    sleep(task_length)
    return output

a_delayed = []

nb_tasks = 20
nb_deps = 3

# Submit all A tasks as delayed
for i in range(0,nb_tasks):
    a_delayed.append(dask.delayed(A)(i))

# Submit all B tasks as delayed, each B task depend on 3 A tasks
b_delayed = []
for i in range(0,nb_tasks):
    b_delayed.append(dask.delayed(B)(a_delayed[i:min(i+nb_deps,nb_tasks)],i))

# Display graph
dask.visualize(b_delayed,filename='graph.png',color='order', cmap='autumn')
```

I tried to return `xarray` with approximate size matching what is done in the actual code. I am not sure this information is used by dask but I also use this code to simulate actual execution on a distributed cluster.

You can find attached the displayed graph.

![graph](https://user-images.githubusercontent.com/10595679/69046277-de0aa380-09f8-11ea-8c58-b36270a78cc7.png)
You can see that task 39 is the last one that will be computed, which means dask will have to keep 19, 26 and 28 until the end. With this setup I would expect an order such as 

A20->B20->A19->B19->A18->B18 (at this point A20 can be released)
->A17 -> B17 (at this point A19 can be released) 
and so on.

Now if instead of 2 x 20 tasks we have 2 x 10 000 this effect is emphasised and I end up with workers memory problem.

Just chiming in to ask if this may be related to https://github.com/dask/distributed/issues/2602? A simpler vision of the problem?
It's possibly similar, though there are a few issues at play. It's important to separate

1. `dask.order.order` getting a bad linear ordering of tasks
2. The distributed scheduler not understanding later dependency structure when assigning a batch of tasks to idle workers (see https://github.com/dask/distributed/issues/2602#issuecomment-518297064)

I'm not sure which one this is yet.

FYI @eriknw is looking into this set of issues for the next little while. Hopefully he'll have more luck at a solution than I did :)
So I have been trying to trick dask into consuming tasks the way I want, using my knowledge of how tasks are organized (which dask tries to discover by itself during graph optimization if I understand well).

I tried options listed on this documentation page (as pointed out by @mrocklin). If I read it correctly, `priority` and `fifo_timeout` should allow to control priority between several calls to `client.compute()`.

So I formed groups of B tasks which I know have a lot of A tasks in common, and submitted each group with a new call to `client.compute()`, using `fifo_timout='0ms'` parameter (as in the doc) to enforce the first in first out rule even if calls are close in time, and also using `priority=` parameters with a decreasing priority from one group to the other.

I must say that neither option seems to have any impact at all on how the tasks are dispatched by the scheduler. I still see it consume most of A tasks over B tasks. Did I mis-use something ?

How is the scheduler using this priority parameter ? Is there something else it takes into account (task average duration, number of tasks, number of dependencies of task ...) that is taken into which would explain that A tasks always get priority ?

On this topic, should I cancel futures as I consume them in `as_completed()` using `client.cancel()` ?

Commit message: Clean up: update docstrings, code comments, and some performance tweaks

Patch to review:
diff --git a/dask/order.py b/dask/order.py
index 15a8fe52ba1..574a59e571b 100644
--- a/dask/order.py
+++ b/dask/order.py
@@ -74,6 +74,7 @@
     This relies on the regularity of graph constructors like dask.array to be a
     good proxy for ordering.  This is usually a good idea and a sane default.
 """
+from math import log
 from .core import get_dependencies, reverse_dict, get_deps  # noqa: F401
 from .utils_test import add, inc  # noqa: F401
 
@@ -88,12 +89,12 @@ def order(dsk, dependencies=None):
     This currently traverses the graph as a single-threaded scheduler would
     traverse it.  It breaks ties in the following ways:
 
-    1.  Start from roots nodes that have the largest subgraphs
-    2.  When a node has dependencies that are not yet computed prefer
-        dependencies with large subtrees  (start hard things first)
-    2.  When we reach a node that can be computed we then traverse up and
-        prefer dependents that have small super-trees (few total dependents)
-        (finish existing work quickly)
+    1.  Begin at a leaf node that is a dependency of a root node that has the
+        largest subgraph (start hard things first)
+    2.  Prefer tall branches with few dependents (start hard things first and
+        try to avoid memory usage)
+    3.  Prefer dependents that are dependencies of root nodes that have
+        the smallest subgraph (do small goals that can terminate quickly)
 
     Examples
     --------
@@ -108,108 +109,265 @@ def order(dsk, dependencies=None):
         deps.discard(k)
 
     dependents = reverse_dict(dependencies)
-
-    total_dependencies = ndependencies(dependencies, dependents)
-    total_dependents, min_dependencies = ndependents(
-        dependencies, dependents, total_dependencies
-    )
-
-    waiting = {k: set(v) for k, v in dependencies.items()}
-
-    def dependencies_key(x):
-        return total_dependencies.get(x, 0), ReverseStrComparable(x)
+    num_needed, total_dependencies = ndependencies(dependencies, dependents)
+    metrics = graph_metrics(dependencies, dependents, total_dependencies)
+    del total_dependencies
+
+    def initial_stack_key(x):
+        """ Choose which task to run at the very beginning
+
+        First prioritize large, tall groups, then prioritize the same as ``dependents_key``.
+        """
+        num_dependents = len(dependents[x])
+        total_dependents, min_dependencies, max_dependencies, min_heights, max_heights = metrics[
+            x
+        ]
+        return (
+            # at a high-level, work towards a large goal (and prefer tall and narrow)
+            -max_dependencies,
+            num_dependents - max_heights,
+            # tactically, finish small connected jobs first
+            min_dependencies,
+            num_dependents - min_heights,  # prefer tall and narrow
+            -total_dependents,  # take a big step
+            # try to be memory efficient
+            num_dependents,
+            # tie-breaker
+            StrComparable(x),
+        )
 
     def dependents_key(x):
-        return (min_dependencies[x], -(total_dependents.get(x, 0)), StrComparable(x))
-
-    result = dict()
-    seen = set()  # tasks that should not be added again to the stack
-    i = 0
-
-    stack = [k for k, v in dependents.items() if not v]
-    if len(stack) < 10000:
-        stack = sorted(stack, key=dependencies_key)
-    else:
-        stack = stack[::-1]
-
-    while stack:
-        item = stack.pop()
+        """ Choose a path from our starting task to our tactical goal
+
+        This path is connected to a large goal, but focuses on completing a small goal.
+        """
+        num_dependents = len(dependents[x])
+        total_dependents, min_dependencies, _, min_heights, _ = metrics[x]
+        return (
+            # tactically, finish small connected jobs first
+            min_dependencies,
+            num_dependents - min_heights,  # prefer tall and narrow
+            -total_dependents,  # take a big step
+            # try to be memory efficient
+            num_dependents - len(dependencies[x]) + num_needed[x],
+            num_dependents,
+            # tie-breaker
+            StrComparable(x),
+        )
 
-        if item in result:
-            continue
-
-        deps = waiting[item]
-
-        if deps:
-            stack.append(item)
-            seen.add(item)
-            if len(deps) < 1000:
-                deps = sorted(deps, key=dependencies_key)
-            stack.extend(deps)
-            continue
-
-        result[item] = i
-        i += 1
-
-        for dep in dependents[item]:
-            waiting[dep].discard(item)
+    def dependencies_key(x):
+        """ Choose which dependency to run as part of a reverse DFS
 
-        deps = [
-            d
-            for d in dependents[item]
-            if d not in result and not (d in seen and len(waiting[d]) > 1)
+        This is very similar to both ``initial_stack_key`` and ``dependents_key``.
+        """
+        num_dependents = len(dependents[x])
+        total_dependents, min_dependencies, max_dependencies, min_heights, max_heights = metrics[
+            x
         ]
-        if len(deps) < 1000:
-            deps = sorted(deps, key=dependents_key, reverse=True)
+        return (
+            # at a high-level, work towards a large goal (and prefer tall and narrow)
+            -max_dependencies,
+            num_dependents - max_heights,
+            # tactically, finish small connected jobs first
+            min_dependencies,
+            num_dependents - min_heights,  # prefer tall and narrow
+            -total_dependents,  # stay where the work is
+            # try to be memory efficient
+            num_dependents - len(dependencies[x]) + num_needed[x],
+            num_dependents,
+            # tie-breaker
+            StrComparable(x),
+        )
+
+    result = {}
+    i = 0
 
-        stack.extend(deps)
+    # Leaf nodes.  We choose one--the initial node--for each weakly connected subgraph
+    init_stack = {k for k, v in dependencies.items() if not v}
+
+    # Nodes in the current weakly connected subgraph that are not part of the current
+    # DFS along dependencies.  The next DFS will begin from these nodes.  `outer_stack`
+    # is populated from `init_stack` and the dependencies of `completed_list`.
+    outer_stack = []
+
+    # Nodes get added to this as they complete.  We will populate `outer_stack` from
+    # dependents of these nodes.  We consider nodes using a FIFO policy (yes, we could
+    # have used collections.deque, but a list with the current index is faster).
+    # Note that this does not include nodes with no dependents!
+    completed_list = []
+    completed_list_index = 0
+
+    # Used to perform a DFS along dependencies.  Once emptied (when traversing dependencies),
+    # this continues (along dependents) along a path from the initial (leaf) node down to a
+    # root node (a tactical goal).  This ensures this root node will get calculated before
+    # we consider any other dependents from `completed_list`.
+    inner_stack = []
+
+    # aliases for speed
+    outer_stack_append = outer_stack.append
+    outer_stack_pop = outer_stack.pop
+    outer_stack_extend = outer_stack.extend
+    completed_list_append = completed_list.append
+    inner_stack_append = inner_stack.append
+    inner_stack_pop = inner_stack.pop
+    inner_stack_extend = inner_stack.extend
+
+    is_init_sorted = False
+    item = min(init_stack, key=initial_stack_key)
+    while True:
+        outer_stack_append(item)
+        while outer_stack:
+            item = outer_stack_pop()
+            if item not in result:
+                inner_stack_append(item)
+
+                while inner_stack:
+                    # Perform a DFS along dependencies until we complete our tactical goal
+                    item = inner_stack_pop()
+                    if item in result:
+                        continue
+
+                    if num_needed[item]:
+                        inner_stack_append(item)
+                        deps = dependencies[item].difference(result)
+                        if len(deps) < 1000:
+                            inner_stack_extend(
+                                sorted(deps, key=dependencies_key, reverse=True)
+                            )
+                        else:
+                            inner_stack_extend(deps)
+                            inner_stack_append(
+                                min(deps, key=dependencies_key)  # one good one
+                            )
+                        continue
+
+                    result[item] = i
+                    i += 1
+                    deps = dependents[item]
+                    if deps:
+                        for dep in deps:
+                            num_needed[dep] -= 1
+                        if not inner_stack:
+                            # Continue towards our tactical goal (an easy-to-compute root)
+                            if len(deps) == 1:
+                                inner_stack_extend(deps)
+                            else:
+                                inner_stack_append(min(deps, key=dependents_key))
+
+                        # Our next starting point will be "seeded" by a completed task in a FIFO
+                        # manner.  When our DFS with `inner_stack` is finished--which means we
+                        # computed our tactical goal--we will choose our next starting point from
+                        # the dependents of completed tasks.  However, it is too expensive to
+                        # consider all dependents of all completed tasks, so we consider the
+                        # dependents of tasks in the order they complete.
+                        completed_list_append(item)
+
+            while not outer_stack and completed_list_index < len(completed_list):
+                # We wait for as long as possible to consider dependents of completed nodes, because:
+                # 1. some may have already completed, so waiting lets us sort fewer items, and
+                # 2. sorting via `dependents_key` depends on `num_needed`, which is dynamic.
+                deps = dependents[completed_list[completed_list_index]].difference(
+                    result
+                )
+                if deps:
+                    if len(deps) == 1:
+                        outer_stack_extend(deps)
+                    else:
+                        outer_stack_extend(
+                            sorted(deps, key=dependents_key, reverse=True)
+                        )
+                completed_list_index += 1
+
+        if len(dependencies) == len(result):
+            break  # all done!
+
+        # We just finished computing a connected group.
+        # Let's choose the first `item` in the next group to compute.
+        # If we have few large groups left, then it's best to find `item` by taking a minimum.
+        # If we have many small groups left, then it's best to sort.
+        # If we have many tiny groups left, then it's best to simply iterate.
+        if not is_init_sorted:
+            prev_len = len(init_stack)
+            init_stack = init_stack.difference(result)
+            N = len(init_stack)
+            m = prev_len - N
+            # is `min` likely better than `sort`?
+            if m >= N or N + (N - m) * log(N - m) < N * log(N):
+                item = min(init_stack, key=initial_stack_key)
+                continue
+
+            if len(init_stack) < 10000:
+                init_stack = sorted(init_stack, key=initial_stack_key, reverse=True)
+            else:
+                init_stack = list(init_stack)
+            init_stack_pop = init_stack.pop
+            is_init_sorted = True
+
+        item = init_stack_pop()
+        while item in result:
+            item = init_stack_pop()
 
     return result
 
 
-def ndependents(dependencies, dependents, total_dependencies):
-    """ Number of total data elements that depend on key
-
-    For each key we return the number of keys that can only be run after this
-    key is run.  The root nodes have value 1 while deep child nodes will have
-    larger values.
+def graph_metrics(dependencies, dependents, total_dependencies):
+    """ Useful measures of a graph used by ``dask.order.order``
+Number of total data elements that depend on key
 
-    We also return the minimum value of the maximum number of dependencies of
-    all final dependencies (see module-level comment for more)
+    For each key we return:
+    1.  The number of keys that can only be run after this key is run.  The
+        root nodes have value 1 while deep child nodes will have larger values.
+    2.  The minimum value of the maximum number of dependencies of
+        all final dependencies (see module-level comment for more)
+    3.  The maximum value of the maximum number of dependencies of
+        all final dependencies (see module-level comment for more)
+    4.  The minimum height from a root node
+    5.  The maximum height from a root node
 
     Examples
     --------
     >>> dsk = {'a': 1, 'b': (inc, 'a'), 'c': (inc, 'b')}
     >>> dependencies, dependents = get_deps(dsk)
 
-    >>> total_dependencies = ndependencies(dependencies, dependents)
-    >>> total_dependents, min_dependencies = ndependents(dependencies,
-    ...                                                  dependents,
-    ...                                                  total_dependencies)
+    >>> _, total_dependencies = ndependencies(dependencies, dependents)
+    >>> metrics = graph_metrics(dependencies, dependents, total_dependencies)
+    >>> total_dependents = {key: val[0] for key, val in metrics.items()}
     >>> sorted(total_dependents.items())
     [('a', 3), ('b', 2), ('c', 1)]
+    >>> sorted(metrics.items())
+    [('a', (3, 3, 3, 3, 3)), ('b', (2, 3, 3, 2, 2)), ('c', (1, 3, 3, 1, 1))]
 
     Returns
     -------
-    total_dependendents: Dict[key, int]
-    min_dependencies: Dict[key, int]
+    metrics: Dict[key, Tuple[int, int, int, int, int]]
     """
-    result = dict()
-    min_result = dict()
+    result = {}
     num_needed = {k: len(v) for k, v in dependents.items()}
     current = {k for k, v in num_needed.items() if v == 0}
+    current_pop = current.pop
+    current_add = current.add
     while current:
-        key = current.pop()
-        result[key] = 1 + sum(result[parent] for parent in dependents[key])
-        try:
-            min_result[key] = min(min_result[parent] for parent in dependents[key])
-        except ValueError:
-            min_result[key] = total_dependencies[key]
+        key = current_pop()
+        parents = dependents[key]
+        if parents:
+            total_dependents, min_dependencies, max_dependencies, min_heights, max_heights = zip(
+                *(result[parent] for parent in parents)
+            )
+            result[key] = (
+                1 + sum(total_dependents),
+                min(min_dependencies),
+                max(max_dependencies),
+                1 + min(min_heights),
+                1 + max(max_heights),
+            )
+        else:
+            val = total_dependencies[key]
+            result[key] = (1, val, val, 1, 1)
         for child in dependencies[key]:
             num_needed[child] -= 1
             if num_needed[child] == 0:
-                current.add(child)
-    return result, min_result
+                current_add(child)
+    return result
 
 
 def ndependencies(dependencies, dependents):
@@ -222,20 +380,29 @@ def ndependencies(dependencies, dependents):
     --------
     >>> dsk = {'a': 1, 'b': (inc, 'a'), 'c': (inc, 'b')}
     >>> dependencies, dependents = get_deps(dsk)
-    >>> sorted(ndependencies(dependencies, dependents).items())
+    >>> num_dependencies, total_dependencies = ndependencies(dependencies, dependents).items())
+    >>> sorted(total_dependencies.items())
     [('a', 1), ('b', 2), ('c', 3)]
+
+    Returns
+    -------
+    num_dependencies: Dict[key, int]
+    total_dependencies: Dict[key, int]
     """
-    result = dict()
+    result = {}
     num_needed = {k: len(v) for k, v in dependencies.items()}
+    num_dependencies = num_needed.copy()
     current = {k for k, v in num_needed.items() if v == 0}
+    current_pop = current.pop
+    current_add = current.add
     while current:
-        key = current.pop()
+        key = current_pop()
         result[key] = 1 + sum(result[child] for child in dependencies[key])
         for parent in dependents[key]:
             num_needed[parent] -= 1
             if num_needed[parent] == 0:
-                current.add(parent)
-    return result
+                current_add(parent)
+    return num_dependencies, result
 
 
 class StrComparable(object):
@@ -265,22 +432,3 @@ def __lt__(self, other):
             return self.obj < other.obj
         except Exception:
             return str(self.obj) < str(other.obj)
-
-
-class ReverseStrComparable(object):
-    """ Wrap object so that it defaults to string comparison
-
-    Used when sorting in reverse direction.  See StrComparable for normal
-    documentation.
-    """
-
-    __slots__ = ("obj",)
-
-    def __init__(self, obj):
-        self.obj = obj
-
-    def __lt__(self, other):
-        try:
-            return self.obj > other.obj
-        except Exception:
-            return str(self.obj) > str(other.obj)
diff --git a/dask/tests/test_order.py b/dask/tests/test_order.py
index cef23e0f135..2bf8a053602 100644
--- a/dask/tests/test_order.py
+++ b/dask/tests/test_order.py
@@ -53,9 +53,7 @@ def test_avoid_broker_nodes(abcde):
         (b, 1): (f, (a, 1)),
         (b, 2): (f, (a, 1)),
     }
-
     o = order(dsk)
-
     assert o[(a, 0)] < o[(a, 1)]
 
     # Switch name of 0, 1 to ensure that this isn't due to string comparison
@@ -66,11 +64,20 @@ def test_avoid_broker_nodes(abcde):
         (b, 1): (f, (a, 0)),
         (b, 2): (f, (a, 0)),
     }
-
     o = order(dsk)
-
     assert o[(a, 0)] > o[(a, 1)]
 
+    # Switch name of 0, 1 for "b"s too
+    dsk = {
+        (a, 0): (f,),
+        (a, 1): (f,),
+        (b, 1): (f, (a, 0)),
+        (b, 0): (f, (a, 1)),
+        (b, 2): (f, (a, 1)),
+    }
+    o = order(dsk)
+    assert o[(a, 0)] < o[(a, 1)]
+
 
 def test_base_of_reduce_preferred(abcde):
     r"""
@@ -168,11 +175,10 @@ def test_avoid_upwards_branching_complex(abcde):
     }
 
     o = order(dsk)
-
     assert o[(c, 1)] < o[(b, 1)]
 
 
-@pytest.mark.xfail(reason="this case is ambiguous", strict=False)
+# @pytest.mark.xfail(reason="this case is ambiguous", strict=False)  # this now passes
 def test_deep_bases_win_over_dependents(abcde):
     r"""
     It's not clear who should run first, e or d
@@ -219,7 +225,6 @@ def test_stacklimit(abcde):
     ndependencies(dependencies, dependents)
 
 
-@pytest.mark.xfail(reason="Can't please 'em all")
 def test_break_ties_by_str(abcde):
     a, b, c, d, e = abcde
     dsk = {("x", i): (inc, i) for i in range(10)}
@@ -227,8 +232,8 @@ def test_break_ties_by_str(abcde):
     dsk["y"] = list(x_keys)
 
     o = order(dsk)
-    expected = {"y": 0}
-    expected.update({k: i + 1 for i, k in enumerate(x_keys)})
+    expected = {"y": 10}
+    expected.update({k: i for i, k in enumerate(x_keys)})
 
     assert o == expected
 
@@ -494,6 +499,7 @@ def test_prefer_short_ancestor(abcde):
     }
     o = order(dsk)
 
+    assert o[(a, 0)] < o[(a, 1)]
     assert o[(b, 0)] < o[(b, 1)]
     assert o[(b, 0)] < o[(c, 2)]
     assert o[(c, 1)] < o[(c, 2)]
@@ -535,3 +541,43 @@ def test_map_overlap(abcde):
     o = order(dsk)
 
     assert o[(b, 1)] < o[(e, 5)] or o[(b, 5)] < o[(e, 1)]
+
+
+def test_use_structure_not_keys(abcde):
+    """ See https://github.com/dask/dask/issues/5584#issuecomment-554963958
+
+    We were using key names to infer structure, which could result in funny behavior.
+    """
+    a, b, _, _, _ = abcde
+    dsk = {
+        (a, 0): (f,),
+        (a, 1): (f,),
+        (a, 2): (f,),
+        (a, 3): (f,),
+        (a, 4): (f,),
+        (a, 5): (f,),
+        (a, 6): (f,),
+        (a, 7): (f,),
+        (a, 8): (f,),
+        (a, 9): (f,),
+        (b, 5): (f, (a, 2)),
+        (b, 7): (f, (a, 0), (a, 2)),
+        (b, 9): (f, (a, 7), (a, 0), (a, 2)),
+        (b, 1): (f, (a, 4), (a, 7), (a, 0)),
+        (b, 2): (f, (a, 9), (a, 4), (a, 7)),
+        (b, 4): (f, (a, 6), (a, 9), (a, 4)),
+        (b, 3): (f, (a, 5), (a, 6), (a, 9)),
+        (b, 8): (f, (a, 1), (a, 5), (a, 6)),
+        (b, 6): (f, (a, 8), (a, 1), (a, 5)),
+        (b, 0): (f, (a, 3), (a, 8), (a, 1)),
+    }
+    o = order(dsk)
+    As = sorted(val for (letter, _), val in o.items() if letter == a)
+    Bs = sorted(val for (letter, _), val in o.items() if letter == b)
+    assert Bs[0] in {1, 3}
+    if Bs[0] == 3:
+        assert As == [0, 1, 2, 4, 6, 8, 10, 12, 14, 16]
+        assert Bs == [3, 5, 7, 9, 11, 13, 15, 17, 18, 19]
+    else:
+        assert As == [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
+        assert Bs == [1, 3, 5, 7, 9, 11, 13, 15, 17, 19]


Code review comments:
diff_hunk: @@ -168,11 +175,10 @@ def test_avoid_upwards_branching_complex(abcde):
     }
 
     o = order(dsk)
-
     assert o[(c, 1)] < o[(b, 1)]
 
 
-@pytest.mark.xfail(reason="this case is ambiguous", strict=False)
+# @pytest.mark.xfail(reason="this case is ambiguous", strict=False)  # this now passes
path: dask/tests/test_order.py
text: @user1:
Can probably remove this comment.

Can you update the docstring to reflect the behavior? I see that we run `e` before `d`. Can we say that `b` runs before both `b` and `c`? Or does `b` run first?