CI: Test when running with pytest-xdist, `pytest -n 8` has duplicated event emitted.

[eddiebergman]

While tests run fine if running in just a single worker of with -n 4, it seems that when boosting this further to 8 cores, the scheduler emitted an extra event of EMPTY.

Looking at the async task the monitors for empty, it is possible there's a race condition causing this to trigger twice. This should be handled properly although it's hard to create a reproducing example. In general, how can we deterministically create race conditions for testing?

https://github.com/automl/amltk/blob/0f4e1af869ace60fa658ca759ae04ff8fc9f10c8/src/amltk/scheduling/scheduler.py#L1148-L1165

---

Error, traceback of test

test_queue_empty_status[loky]

    def test_queue_empty_status(scheduler: Scheduler) -> None:
        task = scheduler.task(sleep_and_return)

        # Reload on the first empty
        @scheduler.on_empty(when=lambda: scheduler.event_counts[scheduler.EMPTY] == 1)
        def launch_first() -> None:
            task.submit(sleep_time=0.1)

        # Stop on the second empty
        @scheduler.on_empty(when=lambda: scheduler.event_counts[scheduler.EMPTY] == 2)
        def stop_scheduler() -> None:
            scheduler.stop()

        end_status = scheduler.run(timeout=3, end_on_empty=False)

        assert task.event_counts == Counter(
            {task.SUBMITTED: 1, task.DONE: 1, task.RESULT: 1},
        )

>       assert scheduler.event_counts == Counter(
            {
                scheduler.STARTED: 1,
                scheduler.FINISHING: 1,
                scheduler.FINISHED: 1,
                scheduler.EMPTY: 2,
                scheduler.STOP: 1,
                scheduler.FUTURE_SUBMITTED: 1,
                scheduler.FUTURE_DONE: 1,
                scheduler.FUTURE_RESULT: 1,
            },
        )
E       AssertionError: assert Counter({Event(name='on_empty'): 1, Event(name='on_start'): 1, Event(name='on_future_submitted'): 1, Event(name='on_timeout'): 1, Event(name='on_finishing'): 1, Event(name='on_finished'): 1, Event(name='on_future_done'): 1, Event(name='on_future_result'): 1}) == Counter({Event(name='on_empty'): 2, Event(name='on_start'): 1, Event(name='on_finishing'): 1, Event(name='on_finished'): 1, Event(name='on_stop'): 1, Event(name='on_future_submitted'): 1, Event(name='on_future_done'): 1, Event(name='on_future_result'): 1})
E         Common items:
E         {Event(name='on_start'): 1,
E          Event(name='on_finishing'): 1,
E          Event(name='on_finished'): 1,
E          Event(name='on_future_submitted'): 1,
E          Event(name='on_future_done'): 1,
E          Event(name='on_future_result'): 1}
E         Differing items:
E         {Event(name='on_empty'): 1} != {Event(name='on_empty'): 2}
E         Left contains 1 more item:
E         {Event(name='on_timeout'): 1}
E         Right contains 1 more item:
E         {Event(name='on_stop'): 1}
E         Full diff:
E         - Counter({Event(name='on_empty'): 2,
E         ?                                  ^
E         + Counter({Event(name='on_empty'): 1,
E         ?                                  ^
E                    Event(name='on_start'): 1,
E         +          Event(name='on_future_submitted'): 1,
E         +          Event(name='on_timeout'): 1,
E                    Event(name='on_finishing'): 1,
E                    Event(name='on_finished'): 1,
E         -          Event(name='on_stop'): 1,
E         -          Event(name='on_future_submitted'): 1,
E                    Event(name='on_future_done'): 1,
E                    Event(name='on_future_result'): 1},
E           )

tests/scheduling/test_scheduler.py:241: AssertionError