issue_comments

Any final comments on this? If not, I'll probably merge this in the next day or two.

All the test are passing here. I would appreciate another round of reviews.

@shoyer - all of your previous comments have been addressed.

I've added some additional tests and cleaned up the implementation a bit. I'd like to get reviews from a few folks and hopefully get this merged later this week.

Thanks all for the comments. I will clean this up a bit and request a full review later this week.

A few things to note:

1. I have not tested save_mfdataset yet. In theory, it should work now but it will require some testing. I'll save that for another PR.
2. I will raise an informative error message when either h5netcdf or scipy are used to write files along with distributed.

I would echo @rabernat in the sense that if distributed writes work for some backends, I would love to see it merged as soon as possible. Thanks for working on this, I am very excited about this feature.

I don't totally understand the scipy constraints on incremental writes but could that be playing a factor here?

I'm pretty sure SciPy supports incremental reads but not incremental writes. In general the entire netCDF file needs to get written at once. Certainly it's not possible to update only part of an array -- scipy needs it in memory as a NumPy array to copy its raw data to the netCDF file.

Given the fact that distributed writes never actually worked, I would be happy at this point if they worked for some backends (e.g. netCDF4 and zarr) but not all (e.g. scipy, h5netcdf). We could document this and make sure to raise an appropriate NotImplementedError if the user tries a distributed write with a non-supported backend.

I make this suggestion because I know Joe has been slogging through this for a long time and it may be better to get it merged and live to fight another day. On the other hand, if Joe is optimistic about resolving the remaining backends, then by all means carry on!

The battle of inches continues. Turning off HDF5's file locking fixes all the tests for netCDF4 (🎉 ). Scipy is not working and h5netcdf doesn't support autoclose so it isn't expected to work.

@shoyer - I don't totally understand the scipy constraints on incremental writes but could that be playing a factor here?

I've this down to 4 test failures:

test_dask_distributed_netcdf_integration_test[NETCDF3_CLASSIC-True-scipy] test_dask_distributed_netcdf_integration_test[NETCDF3_CLASSIC-False-scipy] test_dask_distributed_netcdf_integration_test[NETCDF4_CLASSIC-False-netcdf4] test_dask_distributed_netcdf_integration_test[NETCDF4-False-netcdf4]

I think I'm ready for an initial review. I've made some changes to autoclose and sync so I'd like to get feedback on my approach before I spend too much time sorting out the last few failures.

This is mostly working now. I'm getting a test failure from open_dataset + distributed + autoclose so there is something to sort out there.

@shoyer - I have this working with the netcdf4 backend for with the NETCDF3_CLASSIC file format. I'm still having some locking issues with the HDF library and I'm not sure why.

I think we're getting close. We're currently failing during the sync step and I'm hypothesizing that it is due to the file not being closed after the setup steps. That said, I wasn't able to pinpoint why/where we're missing a close. I think this traceback is pretty informative:

Traceback (most recent call last): File "/Users/jhamman/anaconda/envs/xarray36/lib/python3.6/site-packages/distributed/worker.py", line 1255, in add_task self.tasks[key] = _deserialize(function, args, kwargs, task) File "/Users/jhamman/anaconda/envs/xarray36/lib/python3.6/site-packages/distributed/worker.py", line 641, in _deserialize args = pickle.loads(args) File "/Users/jhamman/anaconda/envs/xarray36/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 59, in loads return pickle.loads(x) File "/Users/jhamman/Dropbox/src/xarray/xarray/backends/common.py", line 445, in __setstate__ self.ds = self._opener(mode=self._mode) File "/Users/jhamman/Dropbox/src/xarray/xarray/backends/netCDF4_.py", line 204, in _open_netcdf4_group ds = nc4.Dataset(filename, mode=mode, **kwargs) File "netCDF4/_netCDF4.pyx", line 2015, in netCDF4._netCDF4.Dataset.__init__ File "netCDF4/_netCDF4.pyx", line 1636, in netCDF4._netCDF4._ensure_nc_success OSError: [Errno -101] NetCDF: HDF error: b'/var/folders/v0/qnh7jvgx5gnglpxfztxdlhk00000gn/T/tmpn_mo662_/temp-0.nc' distributed.scheduler - ERROR - error from worker tcp://127.0.0.1:63248: [Errno -101] NetCDF: HDF error: b'/var/folders/v0/qnh7jvgx5gnglpxfztxdlhk00000gn/T/tmpn_mo662_/temp-0.nc' HDF5-DIAG: Error detected in HDF5 (1.10.1) thread 140736093991744: #000: H5F.c line 586 in H5Fopen(): unable to open file major: File accessibilty minor: Unable to open file #001: H5Fint.c line 1305 in H5F_open(): unable to lock the file major: File accessibilty minor: Unable to open file #002: H5FD.c line 1839 in H5FD_lock(): driver lock request failed major: Virtual File Layer minor: Can't update object #003: H5FDsec2.c line 940 in H5FD_sec2_lock(): unable to lock file, errno = 35, error message = 'Resource temporarily unavailable' major: File accessibilty minor: Bad file ID accessed

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.

@mrocklin - What is the preferred method for determining which scheduler is being used?

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).

A simpler way to handle locking for now (but with possibly subpar performance) would be to use a single global distributed lock.

As for autoclose, perhaps we should make the default autoclose=None, which becomes True if using dask-distributed (or maybe using dask in general?) and otherwise is False.

We might always need to use autoclose=True with distributed. The problem is that in xarray's default mode of operation, we open a netCDF file (without using dask) to create variables, dimensions and attributes, keeping the file open. Then we write the data using dask (via AbstractWritableDataStore.sync()), but the original file is still open.

As for the lock, we need locking both: - Per process: only one thread can use HDF5 for reading/writing at the same time. - Per file: only one worker can read/write a file at the same time.

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 .

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: https://github.com/dask/dask/issues/1892

Thanks @mrocklin for taking a look here. I reworked the tests a bit more to put the to_netcdf inside the distributed cluster section.

Bad news is that the tests are failing again. The good news is we have a semi-informative error message that indicates we're missing a lock somewhere.

Link to most descriptive failing test: https://travis-ci.org/pydata/xarray/jobs/336643000#L5076

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.

Looking into this a little bit, this looks like a dask-distributed bug to me. Somehow Client.get() is returning a tornado.concurrent.Future object, even though sync=True.

xref: https://github.com/pydata/xarray/issues/798 and https://github.com/dask/dask/issues/2488 which are both seem to be relevant to this discussion.

I'm also remembering @pwolfram was quite involved with the original distributed integration so pinging him to see if he is interested in this.

I have definitely used the distributed scheduler with dask.array.store both via Zarr and via a custom store class I wrote: https://gist.github.com/rabernat/e54755e7de4eb5a93cc4e7f9f903e3cc

But I cannot recall if I ever got it to work with netCDF.

Yes, the zarr backend here in xarray is also using dask.array.store and seems to work with distributed just fine.

Yep, using dask.array.store regularly with the distributed scheduler both on our cluster and in a local Docker image for testing. Am using Zarr Arrays as the targets for store to write to. Basically rechunk the data to match the chunking selected for the Zarr Array and then write out in parallel lock-free.

Our cluster uses NFS for things like one's home directory. So these are accessible across nodes. Also there are other types of storage available that are a bit faster and still remain accessible across nodes. So these work pretty well.

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.

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

Kudos for pushing this forward. I don't have much help to offer, but I wanted to recognize your effort...this is hard stuff!.

I've just taken another swing at this and come up empty. I open to ideas in the following areas:

1. scipy backend is failing to roundtrip a length 1 datetime array: https://travis-ci.org/pydata/xarray/jobs/333068098#L4504
2. scipy, netcdf4, and h5netcdf backends are all failing inside dask-distributed: https://travis-ci.org/pydata/xarray/jobs/333068098#L4919

The good news here is that only 8 tests are failing after applying the array wrapper so I suspect we're quite close. I'm hoping @shoyer may have some ideas on (1) since I think he had implemented some scipy workarounds in the past. @mrocklin, I'm hoping you can point me in the right direction.

All of these tests are reproducible locally.

(BTW, I have a use case that is going to need this functionality so I'm personally motivated to see it across the finish line)

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 :/

@mrocklin -

I have a test failing here with a familiar message.

E       TypeError: 'Future' object is not iterable

We saw this last week when debugging some pangeo things. Can you remind me what our solution was?

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.

yes, see https://github.com/pydata/xarray/issues/1464#issuecomment-341329662