Fix bug in sort_values in distributed setting

[phofl]

Our await logic doesn't play nicely with the intermediate computes, hence the different test logic here. dask/dask suffers the same issue

[rjzamora]

Any way to leverage gen_cluster here?

[phofl]

No, it's async

[rjzamora]

Can you explain more? Also, do we really want to allow a test to use all the available cores on our machine like this?

[mrocklin]

You maybe want these fixtures from distributed.utils_test

@pytest.fixture
def cluster_fixture(loop):
    with cluster() as (scheduler, workers):
        yield (scheduler, workers)


@pytest.fixture
def s(cluster_fixture):
    scheduler, workers = cluster_fixture
    return scheduler


@pytest.fixture
def a(cluster_fixture):
    scheduler, workers = cluster_fixture
    return workers[0]


@pytest.fixture
def b(cluster_fixture):
    scheduler, workers = cluster_fixture
    return workers[1]


@pytest.fixture
def client(loop, cluster_fixture):
    scheduler, workers = cluster_fixture
    with Client(scheduler["address"], loop=loop) as client:
        yield client

They handle cleanup and various sanity checks.

Alternatively, if you want things to be very fast, you might also want with LocalCluster(processes=False)

[phofl]

the gen_cluster decorator creates a async cluster and async tests will fail for sort_values/set_index because of intermediate computes. That's not a dask-expr problem, dask/dask has the same issue.

Sure, we can restrict the number of workers

[phofl]

I think I prefer the fast version, this is only supposed to test that the function serialisation works as expected

[rjzamora]

Sure, we can restrict the number of workers

Sorry for mentioning that - I care less about that detail (for now).

[phofl]

See dask/distributed#8167 This one has the same issue in dask/dask