issue_comments

Honestly we don't have a very clean mechanism for this. Probably you want to look at dask.context._globals['get']. This should either be None, which means "use the collection's default" (dask.threaded.get in your case) or a callable. If you're using the distributed scheduler then this will be a method of a Client object. Again, not a very clean thing to test for. My apologies.

Performance-wise Dask locks will probably add 1-10ms of communication overhead (probably on the lower end of that), plus whatever contention there will be from locking. You can make these locks as fine-grained as you want, for example by defining a lock-per-filename with Lock(filename) with no cost (which would presumably reduce contention).

SerializableLock isn't appropriate here if you want inter process locking. Dask's lock is probably better here if you're running with the distributed scheduler.

On Feb 2, 2018 1:38 PM, "Joe Hamman" notifications@github.com wrote:

The tests failure indicates that the netcdf4/h5netcdf libraries cannot open the file in write/append mode, and it seems that is because the file is already open (by another process).

Two questions:

1. autoclose is False to_netcdf. That generally makes sense to me but I'm concerned that we're not being explicit enough about closing the file after each process is done interacting with it. Do we have a way to lock until the file is closed?
2. The lock we're using is dask's SerializableLock. Is that the correct Lock to be using? There is also the distributed.Lock.

xref: dask/dask#1892 https://github.com/dask/dask/issues/1892

— You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub https://github.com/pydata/xarray/pull/1793#issuecomment-362657475, or mute the thread https://github.com/notifications/unsubscribe-auth/AASszNohUQirqOLvJZ_5dkoQ74icEsCkks5tQ0w4gaJpZM4RHpBe .

For reference, the line

computed = restored.compute()

would have to be replaced with

(computed,) = yield c.compute(restored)

To get the same result. However there were a few more calls to compute hidden in various functions (like to_netcdf) that would be tricky to make asynchronous, so I opted to swich to synchronous style instead.

I've pushed a fix for the future error. We were using a coroutine-style test with synchronous style code. More information here: http://distributed.readthedocs.io/en/latest/develop.html#writing-tests

In the future I suspect that the with cluster style tests will be easier to use for anyone not familiar with async programming. They're a little more opaque (you don't have access to the scheduler or workers), but probably match the API that you expect most people to use in practice.

I can take a look at the future not iterable issue sometime tomorrow.

Has anyone successfully used dask.array.store() with the distributed scheduler?

My guess is that this would be easy with a friendly storage target. I'm not sure though. cc @jakirkham who has been active on this topic recently.

I don't know. I would want to look at the fail case locally. I can try to do this near term, no promises though :/

The zarr test seems a bit different. I think your issue here is that you are trying to use synchronous API with the async test harness. I've changed your test and pushed to your branch (hope you don't mind). Relevant docs are here: http://distributed.readthedocs.io/en/latest/develop.html#writing-tests

Async testing is nicer in many ways, but does require you to be a bit familiar with the async/tornado API. I also suspect that operations like to_zarr really aren't yet async friendly.