Time-based coordination in unit test may not guarantee thread orderings
Dear Loguru maintainers.
I wanted to report a potential issue in the test test_safe_adding_while_logging. It looks like the test currently relies on time-based coordination between threads:
def thread_2():
barrier.wait()
time.sleep(0.5)
logger.add(sink_2, format="{message}", catch=False)
logger.info("ccc{}ddd", next(counter))The intent seems to be to delay thread_2 long enough for thread_1 to execute:
def thread_1():
barrier.wait()
logger.info("aaa{}bbb", next(counter))However, time.sleep(0.5) does not actually guarantee that thread_1 will have completed its logger.info(...) call before thread_2 resumes. It only delays thread_2 for at least that duration. The assumed ordering may not hold under higher system load, or with increased thread parallelism (eg: Python 3.14 free-threaded mode)
May I suggest to replace the sleep with an explicit threading.Event, so that thread_2 only proceeds once thread_1 has definitely completed its logging step? For example:
def test_safe_adding_while_logging2(writer):
barrier = Barrier(2)
counter = itertools.count()
thread_1_logging_completed = Event()
sink_1 = NonSafeSink(1)
sink_2 = NonSafeSink(1)
logger.add(sink_1, format="{message}", catch=False)
def thread_1():
barrier.wait()
logger.info("aaa{}bbb", next(counter))
thread_1_logging_completed.set()
def thread_2():
barrier.wait()
thread_1_logging_completed.wait()
logger.add(sink_2, format="{message}", catch=False)
logger.info("ccc{}ddd", next(counter))
threads = [Thread(target=thread_1), Thread(target=thread_2)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
logger.remove()
assert sink_1.written == "aaa0bbb\nccc1ddd\n"
assert sink_2.written == "ccc1ddd\n"This preserves the intended ordering while making the test depend on an explicit synchronisation signal instead of a fixed sleep duration.
If you prefer, I would be happy to submit this as a PR. Thanks again for your time!
Source: Delgan/loguru