Repository navigation
subinterpreters.Queue interface is incompatible with logging.handlers.QueueListener #138253
Description
Bug report
Bug description:
Common interface for getting an item from queue (queue.Queue, multiprocessing.Queue, etc.) is
def get(self, block=True, timeout=None): ...but interpreters.Queue has a different one:
def get(self, timeout=None, *, _delay=10 / 1000): ...This leads to incorrect behavior when used in logging.handlers.QueueListener. QueueListener thread exits ~1second after the queue becomes empty.
class QueueListener(object):
def dequeue(self, block):
return self.queue.get(block)
def _monitor(self):
...
record = self.dequeue(True) # instead of blocking flag it sets the timeout for interpreters.Queue
...Here's the test script
import sys
import time
import logging
from logging.handlers import QueueHandler, QueueListener
from concurrent.futures import InterpreterPoolExecutor, wait
from concurrent.interpreters import create_queue
def initworker(queue):
loghandler = QueueHandler(queue)
loghandler.setLevel(logging.DEBUG)
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
logger.addHandler(loghandler)
def worker(num: int, delay: float):
time.sleep(delay) # emulate highload
logging.debug('finish %s', str(num))
class CustomQueueListener(QueueListener):
def dequeue(self, block: bool) -> logging.LogRecord:
# return self.queue.get(block) # original
return self.queue.get()
def check(listener: QueueListener, bounds: tuple[int, int], delay: float):
tasks = []
with (
InterpreterPoolExecutor(initializer=initworker, initargs=(queue, )) as pool,
listener,
):
for i in range(*bounds):
tasks.append(pool.submit(worker, i, delay))
wait(tasks)
if __name__ == '__main__':
loghandler = logging.StreamHandler(sys.stdout)
queue = create_queue()
check(QueueListener(queue, loghandler), (0, 10), delay=0.5) # everything is ok
queue = create_queue()
check(QueueListener(queue, loghandler), (10, 20), delay=1.25) # no output, because listener thread is stopped
queue = create_queue()
check(CustomQueueListener(queue, loghandler), (20, 30), delay=1.25) # everything is okCPython versions tested on:
3.14
Operating systems tested on:
Windows, Linux
Activity
picnixz commentedon Aug 29, 2025
@ericsnowcurrently does it make sense to allow
blockfor such queues?if not, we could explicitly document the queue interface for logging. I explicitly avoided that in a previous PR by checking if Q is an instance of a known queue (e.g., multiprocessing.Queue), but maybe it's time to do it? cc @vsajipif not, this is not a bug as it doesn't implement the Queue API as documented.EDIT: I did document it in a way that references Queue itself (https://docs.python.org/3/library/logging.config.html#configuring-queuehandler-and-queuelistener).
serhiy-storchaka commentedon Aug 30, 2025
If adding the
blockparameter is not possible, makingtimeouta keyword-only parameter would prevent such kind of errors.pythongh-138253: Fix compatibility of sub-interpreters queues with qu…
serhiy-storchaka commentedon Aug 30, 2025
On other hand, it is very easy to add the
blockparameter since the*_nowait()methods already exist. See #138256. But we should do this before the 3.14 release to avoid breaking user code that can be written between 3.14.0 and 3.14.1.ericsnowcurrently commentedon Sep 1, 2025
I think adding a "block" parameter would be fine.
gh-138253: Fix compatibility of sub-interpreters queues with queue.Qu…
pythongh-138253: Fix compatibility of sub-interpreters queues with qu…
[3.14] gh-138253: Fix compatibility of sub-interpreters queues with q…
hugovk commentedon Sep 8, 2025
Thanks for the report and fix!
[3.14] pythongh-138253: Fix compatibility of sub-interpreters queues …
pythongh-138253: Fix compatibility of sub-interpreters queues with qu…