Fix DataArray.__dask_scheduler__ to point to dask.threaded.get

[mrocklin]

Previously this erroneously pointed to an optimize function, likely a
copy-paste error.

For testing this also redirects the .compute methods to use the
dask.compute function directly if dask.version >= '0.16.0'.

Closes #1759

• Closes #xxxx (remove if there is no corresponding issue, which should only be the case for minor changes)
• Tests added (for all bug fixes or enhancements)
• Tests passed (for all non-documentation changes)
• Passes git diff upstream/master **/*py | flake8 --diff (remove if you did not edit any Python files)
• Fully documented, including whats-new.rst for all changes and api.rst for new API (remove if this change should not be visible to users, e.g., if it is an internal clean-up, or if this is part of a larger project that will be documented later)

[mrocklin]

The mock tests still fail for me. I'm not sure how best to generalize them. Recommendations would be welcome here.

[max-sixty]

I took a look at the tests - I don't have too much helpful beyond breaking it down.

For example, this tests that when .compute(...) is called on a Variable backed by a dask Array, that .compute gets called on the dask Array.

Has that changed? Does dask.compute(x) call x.compute()?

[mrocklin]

Has that changed? Does dask.compute(x) call x.compute()?

No, we no longer call Array.compute at all

[max-sixty]

Right, though xarray used to call .load, which I think must have called through to obj.compute? If that's the case, potentially the tests need changing

xarray/tests/test_backends.py:252: AssertionError
______________________ test_dask_kwargs_variable[compute] ______________________
method = 'compute'
    @pytest.mark.parametrize("method", ['load', 'compute'])
    def test_dask_kwargs_variable(method):
        x = Variable('y', da.from_array(np.arange(3), chunks=(2,)))
        # args should be passed on to da.Array.compute()
        with mock.patch.object(da.Array, 'compute',
                               return_value=np.arange(3)) as mock_compute:
            getattr(x, method)(foo='bar')
>       mock_compute.assert_called_with(foo='bar')
xarray/tests/test_dask.py:724: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
_mock_self = <MagicMock name='compute' id='139838059011600'>, args = ()
kwargs = {'foo': 'bar'}, expected = "compute(foo='bar')"
    def assert_called_with(_mock_self, *args, **kwargs):
        """assert that the mock was called with the specified arguments.
    
            Raises an AssertionError if the args and keyword args passed in are
            different to the last call to the mock."""
        self = _mock_self
        if self.call_args is None:
            expected = self._format_mock_call_signature(args, kwargs)
>           raise AssertionError('Expected call: %s\nNot called' % (expected,))
E           AssertionError: Expected call: compute(foo='bar')
E           Not called

[mrocklin]

@shoyer how would you like me to proceed here? Should I drop the current modifications to the .compute methods and just leave in the __dask_scheduler__ fix?

[shoyer]

Should I drop the current modifications to the .compute methods and just leave in the dask_scheduler fix?

That's probably easiest for now -- updating our mock tests to handle either calling dask.compute() or the Array.compute() method would be a little tricky.

It would be nice to add a test case that uses the default scheduler, e.g., that does a basic compute on each xarray type (Dataset, DataArray and Variable) inside a with dask.set_options(get=None) block. That would give us test coverage for __dask_scheduler__, which is currently missing.

(It would probably be better to explicitly set get explicitly in a function when computing xarray objects containing dask arrays, but that's a larger project.)

[mrocklin]

Done. Tests pass.

[shoyer]

Thanks!