subinterpreters.Queue interface is incompatible with logging.handlers.QueueListener

[j2cry]

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 ok

CPython versions tested on:

3.14

Operating systems tested on:

Windows, Linux

Linked PRs

• gh-138253: Fix compatibility of sub-interpreters queues with queue.Queue #138256
• [3.14] gh-138253: Fix compatibility of sub-interpreters queues with queue.Queue (GH-138256) #138367

[picnixz]

@ericsnowcurrently does it make sense to allow block for 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 @vsajip if 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]

If adding the block parameter is not possible, making timeout a keyword-only parameter would prevent such kind of errors.

[serhiy-storchaka]

On other hand, it is very easy to add the block parameter 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.