Repository: dask/dask
Pull request #3271 (2018-03-11 22:24:31)

Consistent use of ordering by task name
![dask](https://user-images.githubusercontent.com/306380/37259149-2d3bf412-2559-11e8-8f01-563dac607a40.png)

We break ties in task ordering by using the task key itself.  Previously we did this inconsistently due to using `sorted(reverse=True/False)` in different situations.  This cause inconsistent ordrering in general and pathological ordering in some situations (like nearest neighbor computations).

We now use two different comparators for string comparison.

- [ ] Tests added / passed
- [ ] Passes `flake8 dask`
- [ ] Fully documented, including `docs/source/changelog.rst` for all changes
      and one of the `docs/source/*-api.rst` files for new API


mrocklin: It also looks like we need to drop the initial call to `sorted` on root nodes.  https://github.com/pangeo-data/pangeo/issues/150#issuecomment-373066066
mrocklin: Merging this soon if there are no comments

Patch:
diff --git a/dask/order.py b/dask/order.py
index 6ae4bd684..3b6d14666 100644
--- a/dask/order.py
+++ b/dask/order.py
@@ -99,16 +99,17 @@ def order(dsk, dependencies=None):
     waiting = {k: set(v) for k, v in dependencies.items()}
 
     def dependencies_key(x):
-        return total_dependencies.get(x, 0), StrComparable(x)
+        return total_dependencies.get(x, 0), ReverseStrComparable(x)
 
     def dependents_key(x):
-        return total_dependents.get(x, 0), StrComparable(x)
+        return (total_dependents.get(x, 0),
+                StrComparable(x))
 
     result = dict()
     i = 0
 
     stack = [k for k, v in dependents.items() if not v]
-    stack = sorted(stack, key=total_dependencies.get)
+    stack = sorted(stack, key=dependencies_key)
 
     while stack:
         item = stack.pop()
@@ -214,3 +215,21 @@ class StrComparable(object):
             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/docs/source/changelog.rst b/docs/source/changelog.rst
index c65294bf8..e81a61aa7 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -29,6 +29,7 @@ Core
 ++++
 
 - Fixed bug when using unexpected but hashable types for keys (:pr:`3238`) `Daniel Collins`_
+- Fix bug in task ordering so that we break ties consistently with the key name (:pr:`3271`) `Matthew Rocklin`_
 
 
 0.17.1 / 2018-02-22


Test patch:
diff --git a/dask/tests/test_local.py b/dask/tests/test_local.py
index 401717b2b..a7bc42f86 100644
--- a/dask/tests/test_local.py
+++ b/dask/tests/test_local.py
@@ -176,4 +176,4 @@ def test_ordering():
 
     get_sync(dsk, 'y')
 
-    assert L == sorted(L, reverse=True)
+    assert L == sorted(L)
diff --git a/dask/tests/test_order.py b/dask/tests/test_order.py
index 38dd98c2c..b9c9d8514 100644
--- a/dask/tests/test_order.py
+++ b/dask/tests/test_order.py
@@ -146,8 +146,10 @@ def test_avoid_upwards_branching_complex(abcde):
            (a, 2): (f, (a, 3)),
            (a, 3): (f, (b, 1), (c, 1)),
            (b, 1): (f, (b, 2)),
+           (b, 2): (f,),
            (c, 1): (f, (c, 2)),
            (c, 2): (f, (c, 3)),
+           (c, 3): (f,),
            (d, 1): (f, (c, 1)),
            (d, 2): (f, (d, 1)),
            (d, 3): (f, (d, 1)),
@@ -239,7 +241,8 @@ def test_gh_3055():
     dsk = dict(w.__dask_graph__())
     o = order(dsk)
     L = [o[k] for k in w.__dask_keys__()]
-    assert sorted(L) == L
+    assert sum(x < len(o) / 2 for x in L) > len(L) / 3  # some complete quickly
+    assert sorted(L) == L  # operate in order
 
 
 def test_type_comparisions_ok(abcde):
@@ -351,3 +354,58 @@ def test_local_parents_of_reduction(abcde):
     dask.get(dsk, [a1, b1, c1])  # trigger computation
 
     assert log == expected
+
+
+def test_nearest_neighbor(abcde):
+    """
+
+    a1  a2  a3  a4  a5  a6  a7 a8  a9
+     \  |  /  \ |  /  \ |  / \ |  /
+        b1      b2      b3     b4
+
+    Want to finish off a local group before moving on.
+    This is difficult because all groups are connected.
+    """
+    a, b, c, _, _ = abcde
+    a1, a2, a3, a4, a5, a6, a7, a8, a9 = [a + i for i in '123456789']
+    b1, b2, b3, b4 = [b + i for i in '1234']
+
+    dsk = {b1: (f,),
+           b2: (f,),
+           b3: (f,),
+           b4: (f,),
+           a1: (f, b1),
+           a2: (f, b1),
+           a3: (f, b1, b2),
+           a4: (f, b2),
+           a5: (f, b2, b3),
+           a6: (f, b3),
+           a7: (f, b3, b4),
+           a8: (f, b4),
+           a9: (f, b4)}
+
+    o = order(dsk)
+
+    assert 3 < sum(o[a + i] < len(o) / 2 for i in '123456789') < 7
+    assert 1 < sum(o[b + i] < len(o) / 2 for i in '1234') < 4
+    assert o[min([b1, b2, b3, b4])] == 0
+
+
+def test_string_ordering():
+    """ Prefer ordering tasks by name first """
+    dsk = {('a', 1): (f,), ('a', 2): (f,), ('a', 3): (f,)}
+    o = order(dsk)
+    assert o == {('a', 1): 0,
+                 ('a', 2): 1,
+                 ('a', 3): 2}
+
+
+def test_string_ordering_dependents():
+    """ Prefer ordering tasks by name first even when in dependencies """
+    dsk = {('a', 1): (f, 'b'), ('a', 2): (f, 'b'), ('a', 3): (f, 'b'),
+           'b': (f,)}
+    o = order(dsk)
+    assert o == {'b': 0,
+                 ('a', 1): 1,
+                 ('a', 2): 2,
+                 ('a', 3): 3}