[ci] [dask] CI jobs failing with Dask 2022.7.1

[jameslamb]

Description

Created from #5388 (comment).

All CUDA CI jobs and several Linux jobs are failing with the following.

FAILED ../tests/python_package_test/test_dask.py::test_machines_should_be_used_if_provided[binary-classification]
FAILED ../tests/python_package_test/test_dask.py::test_machines_should_be_used_if_provided[multiclass-classification]
FAILED ../tests/python_package_test/test_dask.py::test_machines_should_be_used_if_provided[regression]
FAILED ../tests/python_package_test/test_dask.py::test_machines_should_be_used_if_provided[ranking]
= 4 failed, 700 passed, 10 skipped, 2 xfailed, 395 warnings in 655.51s (0:10:55) =

client.restart() calls in that test are resulting in the following:

raise TimeoutError(f"{len(bad_nannies)}/{len(nannies)} nanny worker(s) did not shut down within {timeout}s")
E asyncio.exceptions.TimeoutError: 1/2 nanny worker(s) did not shut down within 120s

traceback (click me)

/root/miniforge/envs/test-env/lib/python3.9/site-packages/distributed/client.py:3360: in restart
    return self.sync(
/root/miniforge/envs/test-env/lib/python3.9/site-packages/distributed/utils.py:338: in sync
    return sync(
/root/miniforge/envs/test-env/lib/python3.9/site-packages/distributed/utils.py:405: in sync
    raise exc.with_traceback(tb)
/root/miniforge/envs/test-env/lib/python3.9/site-packages/distributed/utils.py:378: in f
    result = yield future
/root/miniforge/envs/test-env/lib/python3.9/site-packages/tornado/gen.py:762: in run
    value = future.result()
/root/miniforge/envs/test-env/lib/python3.9/site-packages/distributed/client.py:3329: in _restart
    await self.scheduler.restart(timeout=timeout, wait_for_workers=wait_for_workers)
/root/miniforge/envs/test-env/lib/python3.9/site-packages/distributed/core.py:1153: in send_recv_from_rpc
    return await send_recv(comm=comm, op=key, **kwargs)
/root/miniforge/envs/test-env/lib/python3.9/site-packages/distributed/core.py:943: in send_recv
    raise exc.with_traceback(tb)
/root/miniforge/envs/test-env/lib/python3.9/site-packages/distributed/core.py:769: in _handle_comm
    result = await result
/root/miniforge/envs/test-env/lib/python3.9/site-packages/distributed/utils.py:778: in wrapper
    return await func(*args, **kwargs)

It looks like those jobs are getting dask and distributed 2022.7.1

    dask-2022.7.1              |     pyhd8ed1ab_0           5 KB  conda-forge
    dask-core-2022.7.1         |     pyhd8ed1ab_0         840 KB  conda-forge
    dbus-1.13.6                |       h5008d03_3         604 KB  conda-forge
    distributed-2022.7.1       |     pyhd8ed1ab_0         735 KB  conda-forge

which hit conda-forge 3 days ago.

Reproducible example

Here's an example: https://github.com/microsoft/LightGBM/runs/7522939980?check_suite_focus=true

I don't believe the failure is related to anything specific on the PR that that failed build came from.

Additional Comments

Note that this should not be a concern for jobs using Python < 3.8, as dask / distributed have dropped support for those Python versions.

Logs from an example build on #5388 where I tried to pin to exactly dask==2022.7.0 (build link):

UnsatisfiableError: The following specifications were found
to be incompatible with the existing python installation in your environment:

Specifications:

  - dask==2022.7.0 -> python[version='>=3.8']
  - distributed==2022.7.0 -> python[version='>=3.8']

[jmoralez]

Linking possible source: dask/distributed#6714

[jameslamb]

Great find @jmoralez !!!

I just read through that PR, totally agree with you that it looks very relevant and is probably the root cause for this.

I wonder if the way that the failing test leaves the cluster with an error on one worker results in a situation where the worker can't shut down cleanly.

I believe the breaking change in that PR you linked is that restart() will now, by default, block until all workers shut down and restart successfully.

LightGBM/tests/python_package_test/test_dask.py

Lines 1643 to 1652 in f94050a

	# test that "machines" is actually respected by creating a socket that uses
	# one of the ports mentioned in "machines"
	error_msg = f"Binding port {open_ports[0]} failed"
	with pytest.raises(lgb.basic.LightGBMError, match=error_msg):
	with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
	s.bind((workers_hostname, open_ports[0]))
	dask_model.fit(dX, dy, group=dg)
	# The above error leaves a worker waiting
	client.restart()

Based on https://docs.dask.org/en/stable/_modules/distributed/client.html#Client.restart, I guess we could try something like the following to recover the old behavior?

client.restart(wait_for_workers=False)
client.wait_for_workers(n_workers=1)

Or, alternatively, we can try to figure out why raising an error the way that this test intentionally does creates issues for Client.restart().

@jmoralez do you have time to look into this this week? I'd be happy to back off and let you take it if you're interested.

---

@gjoseph92 , also tagging you since you were the author of dask/distributed#6714, and maybe whatever we're doing here in LightGBM is weird/unexpected and might be informative to you as you continue working on distributed.

[gjoseph92]

Sorry about that. Yeah, client.restart is now stricter and verifies that the restart was actually successful.

I don't understand what you're doing with the socket, but presumably you're doing something that makes the worker block during Worker.close. I'm kinda curious what the motivation for that is / what you're trying to test (and if this is something LightGBM users would encounter and have issues with like this test does)? Another important question: what happens to the stuck worker after the test? Does it ever actually shut down, or is the process leaked and just keeps running?

Regardless, what @jameslamb said is what you'll want to match the previous behavior here:

client.restart(wait_for_workers=False)
client.wait_for_workers(n_workers=1)

[gjoseph92]

Also, at some point soon-ish, I'm hoping to finish up and merge dask/distributed#6427. This might break your test yet again.

Currently, the Nanny.kill process goes:

1. Call Worker.close
2. If that doesn't shut it down, send a SIGTERM to the worker. I think with your current test setup this will actually kill the worker process. But if you were running dask-worker (like users would be), it would just trigger Worker.close, so effectively a no-op.
3. If it's still not closed after the timeout... 🤷

After that PR:

1. Call Worker.close
2. If that doesn't shut it down, send a SIGKILL to the worker. SIGKILL cannot be caught so the process will shut down nearly immediately (modulo it being suspended or currently waiting on a system call).
3. If it's still not closed after the timeout, raise an error.

The key difference is that maybe what you have right now is managing to block/ignore the SIGTERM, preventing the stuck worker from shutting down. Since you can't block a SIGKILL, after this PR is merged I would expect (and hope) that client.restart would result in both workers restarting successfully, not just 1. If you're counting on the stuck worker to survive the restart, I don't think it would (it's a bug that it's able to right now).

[jameslamb]

Thanks very much for that @gjoseph92 ! And no need to apologize... my @ wasn't a passive-aggressive "hey why did you break our tests", I promise 😂

I'm kinda curious what the motivation for that is / what you're trying to test

LightGBM has its own distributed computing setup (e.g. to perform collective operations across multiple processes) that lightgbm.dask just wraps. In that setup, worker processes communicate over TCP sockets. For users in highly-restricted environments (e.g. behind a strict firewall), we provide the option to explicitly list the ports used for those sockets.

This test checks that the expected error is raised in the situation where someone specifies a port that is already in use at the time that training starts.

what happens to the stuck worker after the test? Does it ever actually shut down, or is the process leaked and just keeps running?

I suspect (and have suspected for a while) that the way we are raising errors on lightgbm.dask could be improved, and might result in leaving workers in a bad state.

LightGBM/python-package/lightgbm/dask.py

Lines 664 to 671 in f94050a

	# Start computation in the background
	parts = list(map(delayed, parts))
	parts = client.compute(parts)
	wait(parts)
	for part in parts:
	if part.status == 'error': # type: ignore
	return part # trigger error locally

I wouldn't be surprised to learn that the process is leaked and keeps running. Since our only use of Client.restart() is in a test and since that happens only on remote machines during CI, it would be easy for us to miss a leaked process.

If you're counting on the stuck worker to survive the restart

We definitely are not, and the new behavior you've described sounds like a nice improvement to Client.restart() 😁

[jmoralez]

Or, alternatively, we can try to figure out why raising an error the way that this test intentionally does creates issues for Client.restart()

It's not the error itself, it's actually the other worker that gets stuck. Since the distributed training expects all machines to be ready, when one fails to bind the port the other ones just wait forever. That's what I thought restart fixed but I'm not sure anymore haha. I think one better fix would be to add a timeout to the distributed training, so that if one of the machines fails to connect, the process is interrupted in the rest of them. I'll give that a shot and let you know.

[jameslamb]

add a timeout to the distributed training, so that if one of the machines fails to connect, the process is interrupted in the rest of them

oh interesting! LightGBM does support that by the way --> https://lightgbm.readthedocs.io/en/v3.3.2/Parameters.html#time_out

but it defaults to 120 minutes 😬

[jmoralez]

Nice! That should work, I'll try it.

[jmoralez]

It seems that timeout is for individual sockets, we need to have one for the whole network setup. This is the relevant part:

LightGBM/src/network/linkers_socket.cpp

Lines 196 to 216 in f94050a

	for (auto it = need_connect.begin(); it != need_connect.end(); ++it) {
	int out_rank = it->first;
	// let smaller rank connect to larger rank
	if (out_rank > rank_) {
	int connect_fail_delay_time = connect_fail_retry_first_delay_interval;
	for (int i = 0; i < connect_fail_retry_cnt; ++i) {
	TcpSocket cur_socket;
	if (cur_socket.Connect(client_ips_[out_rank].c_str(), client_ports_[out_rank])) {
	// send local rank
	cur_socket.Send(reinterpret_cast<const char*>(&rank_), sizeof(rank_));
	SetLinker(out_rank, cur_socket);
	break;
	} else {
	Log::Warning("Connecting to rank %d failed, waiting for %d milliseconds", out_rank, connect_fail_delay_time);
	cur_socket.Close();
	std::this_thread::sleep_for(std::chrono::milliseconds(connect_fail_delay_time));
	connect_fail_delay_time = static_cast<int>(connect_fail_delay_time * connect_fail_retry_delay_factor);
	}
	}
	}
	}

So it doesn't wait forever, it does have a finite amount of retries and each time it waits longer. I added a log there to see how many retries it would make and there were 20, however after these 20 tries instead of failing the setup it assumed everything was ok and then segfaulted.

I'll keep investigating this