Fix regressions in #4651

[crusaderky]

Fix regressions introduced in #4651:

• failing test
• if a worker sends a heartbeat before its SystemMonitor has a chance to run, metrics["memory"] will be None and that will cause a crash in the scheduler.

[pentschev]

I confirmed locally this fixes the problem for me too, thanks so much for the quick fix @crusaderky !

[crusaderky]

Stress test outcome:

• 77 out of 77 times, test_memory was successful
• once, an unrelated test failed
• 22 times, CI fell over with no log whatsoever, very fast and definitely before reaching the test suite

[crusaderky]

I can't see a way to write a unit test for this short of monkey-patching SystemMonitor?

[pentschev]

It seems like it's going to be the only way, another test does exactly that:

distributed/distributed/tests/test_worker.py

Lines 1697 to 1717 in e4b534a

	@pytest.mark.asyncio
	@pytest.mark.parametrize("reconnect", [True, False])
	async def test_heartbeat_comm_closed(cleanup, monkeypatch, reconnect):
	with captured_logger("distributed.worker", level=logging.WARNING) as logger:
	async with await Scheduler() as s:
	def bad_heartbeat_worker(*args, **kwargs):
	raise CommClosedError()
	async with await Worker(s.address, reconnect=reconnect) as w:
	# Trigger CommClosedError during worker heartbeat
	monkeypatch.setattr(
	w.scheduler, "heartbeat_worker", bad_heartbeat_worker
	)
	await w.heartbeat()
	if reconnect:
	assert w.status == Status.running
	else:
	assert w.status == Status.closed
	assert "Heartbeat to scheduler failed" in logger.getvalue()

[fjetter]

There is distributed.admin.system-monitor.interval which controls how often the monitor runs. You could set it to incredibly high values such that it is never executed during the test runtime

---

Another patch version w/ using monkeypatch, you could remove the PC before it is even started. Something like

@pyters.mark.asyncio
async def test_foo():
    s = Scheduler()
    s.periodic_callbacks["monitor"] = None
    w = Worker(s)
    w.periodic_callbacks["monitor"] = None
    await s
    await w
    async with Client(s):
        ....

[crusaderky]

Setting distributed.admin.system-monitor.interval to a very high value before I create the Scheduler has no effect ( I can see data arriving in the heartbeat from the SystemMonitor.update).

Setting

s.periodic_callbacks["monitor"] = None
w.periodic_callbacks["monitor"] = None

fails with

        for pc in self.periodic_callbacks.values():
>           pc.stop()
E           AttributeError: 'NoneType' object has no attribute 'stop'

this has no effect:

del s.periodic_callbacks["monitor"]
del w.periodic_callbacks["monitor"]

this has no effect:

pc = PeriodicCallback(lamba: None, 999999999)
s.periodic_callbacks["monitor"] = pc
w.periodic_callbacks["monitor"] = pc

[pentschev]

Would people be ok with merging this even without a test and leaving a test for a different PR? This would unblock CI in Dask-CUDA and UCX-Py (potentially other RAPIDS projects too).

[fjetter]

I'm fine with postponing tests since our CI is also failing hard. @crusaderky your call

[crusaderky]

I need an extra 2-3 hours to cook a unit test. I'm ok merging without

[jrbourbeau]

Thanks for your work here @crusaderky. I'm going to merge this as is to unblock CI, but left a comment below

[jrbourbeau]

Is there something more direct we can probe here instead of sleeping for 10 seconds?

[crusaderky]

Not really because the unmanaged memory is very volatile, so we don't know how many keys are going to be spilled out exactly. Also, as noted it doesn't slow the test down.