I don't know what the Travis CI error is about.

@eriknw I have tried these changes locally and I am still seeing memory growth. Could you possibly try the following example (a slightly simplified version of the example from #5864) and see if it works for you? You can examine htop, but my quick and dirty check is to just run with /usr/bin/time and check maxresident.

import dask.array as da
import numpy as np

n_points = 1250*60*60*4
n_scales = 300

scales = da.arange(n_scales, chunks=(1,))
omega = da.fft.fftfreq(n_points, chunks=(1250*60*5,)) * 2 * np.pi

def mul_2d(arr1, arr2):
    return np.atleast_2d(arr1*arr2)

x = da.blockwise(mul_2d, "ij",
                 scales, "i",
                 omega, "j")

y = da.to_zarr(x, 'testdata', overwrite=True)

da.compute(y, scheduler="single-threaded")

On this branch and 2.10.1 this will use ~8.5GB of RAM. With 2.9.1 it only uses 450MB. Very occasionally it behaves itself with 2.10.1/this branch but if you run it a few times the problem should be apparent.

I don't know what the Travis CI error is about.

That can be ignored. Tracking it in #5867

@JSKenyon thanks again! I updated the PR to try a different approach. I think it's generally better than what's on master, and I prefer it over my first attempt in this PR. Can you give this a try on #5859? I don't think it'll be perfect, but I hope and expect it's better.

Regarding your new example, I haven't found any version of order (old, current, new, or variations) that does well. They all use >10GB of memory for me.

@eriknw are you able to write tests for the new behavior?

And the memory usage in #5872 (comment) looks good for me with Dask 2.9.1. Using memory-profiler

Aha! Thanks for checking @TomAugspurger. I need to be more clear about how I tested the above example.

This example runs fine for me in Dask 2.9.1. It also runs fine if I update Dask 2.9.1 to use dask.order.order from this PR. I did not check Dask 2.9.1 with dask.order.order from current master (although I suspect it's fine).

This example does not run fine for me in current master, this PR, or current master using dask.order.order from Dask 2.9.1.

To iterate: any order works fine in Dask 2.9.1, and no order works well in current master. Therefore, I conclude this particular regression is not caused by the change of order.

I'll see what I can do about adding tests. This is kinda hard to isolate.

Ah, makes sense.

I find that a little surprising @eriknw. When I tested for the original issue (here), I observed the change in behaviour only after the ordering changes. I was relatively certain I had isolated it, as I built from source and checked out the relevant commits. That doesn't mean I didn't make a mistake though. :-)

Sorry for the confusion @JSKenyon. The issue in #5859 is indeed from the changes to order. However, the example you posted in #5872 (comment) (adapted from #5864) appears to behave badly due to something other than order.

@eriknw I understand now. Not entirely sure why everything I interact with seems to have a memory problem.

I have tested out the latest version of order.order on my problem. It still underperforms relative to the old order.order. 2.9.1 has a maximum footprint of around 1.3GB on my test data. This PR is around 5.3GB. Unfortunately that is still unacceptably high for the single threaded case, as this ultimately needs to be run in parallel on ~TB sized data sets.

The problem with the example I posted here stems from this commit.

I think it is worthwhile clarifying that there are now two separate problems which have been exposed by this PR and the issues that precede it:

1. The changes to order.order can negatively affect memory usage, particularly for large, complex graphs. This was raised in Ordering changes negatively affecting memory use #5859. This PR is an attempt to improve/rectify this problem.
2. Changes to array metadata creation in to_zarr can lead to massively increased memory footprints. The cause seems to be Delay creating metadata in to_zarr #5797. This emerged from my take on Memory leak on local cluster  #5864. Pinging @chrisroat for his input.

Apologies for the memory issue, if it stems from #5797. I take from this discussion that the problem goes away if a backend other than zarr is substituted in the repro code?

I am not fully familiar with the task execution/graph pieces of dask. Essentially what #5797 does is add one create task upstream of the zarr store tasks. Since a single new task presumably is light-weight, my educated guess is that the extra memory must then come from the extra of graph edges/dependencies from the store tasks back to the extra task (in addition to whatever dependencies they already had).

See this comment for a visual representation.

Thanks again for the engagement @JSKenyon. I've been making progress locally (at least to my eyes), so I should have something new for you to try tomorrow.

@chrisroat I believe your instinct is likely correct. I think wrapping create with delayed creates a common ancestor for all the writes, which in turn causes the scheduler (possibly via order) to hold everything in memory. A colleague of mine pointed me at the following PR for one of his projects. We believe it may be a similar issue.

@eriknw Thank you for continuing to work on this - I know that it isn't exactly an easy problem so I really appreciate your efforts.

I've been making progress locally (at least to my eyes), so I should have something new for you to try tomorrow.

Checking in here. Is there still progress or is this stalled? If it stalls out, do we have a backup plan?

A backup plan would be rolling back the delayed create PR #5797 , if that is the problem. While I believe that PR is the correct way forward, it's a minor improvement that can wait until its downsides can be mitigated.

@chrisroat I think that the delayed creation issue is less pressing than the ordering problems, but @mrocklin likely has a better idea of how many people are affected.

The ordering issue is not resolved as yet. I have been waiting on further progress from @eriknw.

I'll be able to dig into this over the next few days.

@JSKenyon do you have a reproducible example demonstrating the issues you mentioned in #5872 (comment)? It looks to me like the behavior of 2.9.1, master, and this PR are all fairly similar once a27f111 is reverted.

2.9.1 order - a27f111

master order - a27f111

GH-5872 order - a27f111

Hey, sorry for my absence (I choose not to get notifications on the weekend). I've continued to work on this the last week. This is actually a really fun space to think about, and I've explored several variations (too many to enumerate). I'll push a new attempt up today.

To my eyes, this actually looks pretty good (from the example in #5859):

New attempt is up. I think its behavior is pretty reasonable.

@JSKenyon, I would be much obliged if you could give it a try. CC @TomAugspurger

Right now I'm planning to merge this after our release, to give this a bit of time on master. dask/community#38

@mrocklin there's a discussion on performance at https://github.com/dask/dask/pull/5872/files#r394585154 that may interest you. Our benchmarks pick up what should be a worst case scenario for the new partition_keys concept. Do you have thoughts on the relative value of runtime of order vs. getting a good ordering?

Getting a good ordering is important. That's the difference of things working vs them not working. The comparison of FIFO/LIFO vs current approach misses though that our previous version of this broke ties by the key itself, which in practice was typically nicely ordered (because dask dataframe/array tend to name things in a consistent and well-ordered way). I think that reverting a couple of changes in ordering code is probably the thing to compare this to, rather than master.

Ordering costs come up as non-trivial factors in our overhead. This limits Dask's use on very large graphs, such as frequently come up in Pangeo workloads, for example (@rabernat for example will probably not be happy to hear that ordering costs may triple). Keeping these costs low are important. They're not as important as making sure that things work, so this should probably go in, but I recommend that we reassess previous changes that were made that moved order away from using key names. I think that @eriknw was around for that change as well, and may be able to provide a good counter-argument.

The comparison of FIFO/LIFO vs current approach misses though that our previous version of this broke ties by the key itself

I disagree wholeheartedly.

We still break ties based on key name. The structural considerations go deeper than this. I think you are missing the point of FIFO/LIFO/current comparison. Going back to "use keys more often" only addresses how we order dependencies or dependents. This only takes us so far. Behaviorally, it is at least as important to determine what to compute next after a root node (w/o dependents). The old method used a LIFO approach, because dependents remained on the stack. Now we can compare this to FIFO and the current approach. How we use keys is largely irrelevant.

Many of these benchmarks are not meant to capture real workloads, but, instead, to reveal performance characteristics. I am performance-minded, and I worry little over the benchmark that is 3.5x slower. Looking more closely at this benchmark (which has 10,000 layers of two nodes, where each layer is fully connected to the next layer), the time per edge and time per node is still very fast, so this particular performance characteristic is unlikely to dominate in real workloads, and, if by chance it does, then I suspect the overhead of order would be acceptable, because any operation that iterates over the DAG would likely be comparable.

Recall that #5646 generally improved performance across the board. Sometimes, gains of 3x-10x were seen! I'm sorry I don't have these numbers available, but I suspect this PR would compare favorably to pre-#5646.

All previous ordering algorithms have serious known shortcomings. I can't recommend anything to revert, but if somebody wants to dig into this, I'll be available for discussion.

I apologize that my counter-argument is largely dismissive of Matt's argument, but I suppose this is natural given that Matt's was largely dismissive of the rationale for the current approach in this PR. I understand LIFO vs FIFO vs other may be a bit abstract, but these are what appear to dominate the behavior of order on graphs of interest. I'm sorry I haven't been able to demonstrate this more clearly, but I think discussing it even in the abstract will be useful for the next person who chooses to dive into this.

It'd be helpful to know of workloads where tie-breaking by key was important. I'll see if I can run the workload from the original issue on the old order implementation with some metrics for when tie-breaking was used.

I'll also run this order against some more realistic workloads from the pangeo examples. I'd expect the 3x slowdown in order time will be near the theoretical upper bound, not something that's likely in real-world use cases.

Let me discuss this a little more to allow for more informed criticisms and concerns (which are welcome!).

Dependents are sorted by dependents_key, and dependencies are sorted by dependencies_key. Breaking ties based on the key in the DAG is done in these two functions. We actually have a lot of freedom in choosing these functions; in other words, ordering tends to perform well for a wide variety of key functions, and we shouldn't be afraid to tinker with these in the future. The choice of which of the below approaches plays a larger role determining how order succeeds or fails.

In terms of the current node...

LIFO:

• append dependents to stack
• append non-computed dependencies to stack

FIFO:

• if stack is empty, append min dependent to stack
• append non-computed dependencies to stack
• other dependents are added to stack (one at a time) after root node (with no dependents) is computed
  • currently done as FIFO, but could easily switch to LIFO

This PR:

• similar to FIFO, we append non-computed dependencies to stack and, if necessary, the min dependent
• if stack is not empty, sometimes make a new stack from a dependent that is better than current stack (and save previous stack for later)
• other dependents go into dicts of pools, one of which will be processed once all stacks from above are computed
  • there are currently two dicts; one dict is always given priority over the other dict
  • we can sort the keys of the dict, so we are not limited by FIFO or LIFO
• there are some nuances not covered here

Some graphs naturally perform well for LIFO, but fail for others. Likewise, some do better (or worse) with FIFO. This PR acknowledges that neither LIFO or FIFO dominates the other, and both have failings. Moreover, I tried methods that switch between LIFO or FIFO, but none of these experiments fared well either. The current approach is more complicated, but it attempts to address known shortcomings. I agree with @TomAugspurger to give this time to sit in master before release.

I tried both this and master with the old pre-#5646 order out on https://github.com/pangeo-data/pangeo-example-notebooks. No noticeable difference in order time (submitting computations still felt instantaneous) and approximately the same runtimes for the computations.

I'll try to get some statistics on when tie-breaking was important tomorrow.

@TomAugspurger any idea what this test failure is? Looks unrelated.

With the latest change, the example from #5859 (comment) looks like this:

The latest changes may have slightly improved a couple of the benchmarks. Most are unchanged.

That failure can be ignored. We have an open issue about it IIRC.

Short status update: I haven't been able to find realistic workloads where the slowdown in order is noticeable (tried some pangeo ones, and some large dask.dataframe stuff).

I failed to come up with metrics on when we fall back to name as a tie-breaker like I promised in #5872 (comment).

@jcrist is hopefully going to have time to look through this this week.

The error here is unrelated. I think this is good to merge now, thanks @eriknw!

We probably want this to sit in master for a bit before the next release, just in case issues turn up. Since we just released last week, I think now is a fine time. cc-ing @jrbourbeau or @TomAugspurger for a quick 2nd opinion.

The last dask/dask release was April 24, so if we follow the 2 week cadence we'd do a release this Friday. Dunno if we'll follow that this week since we have had some extra distributed releases.

I suspect we'll discuss this on the maintainers call today.

Ah, it was just distributed that was released last week. My mistake.

Since we've now released, I'm going to merge this. Thanks @eriknw!

Thanks Erik!