queue.Queue - Adding Cancelling Support

[mementum]

Feature or enhancement

Adding methods cancel and cancel_all (akin to notify and notify_all to queue.Queue to support cancelling a get (and potentially a put) operation.

Pitch

Awaiting on get of an asyncio can be cancelled. But waiting on the get of a queue.Queue cannot be cancelled. A usual pattern is sending None to the queue, but this is simply a convention.

Previous discussion

https://discuss.python.org/t/queue-queue-adding-cancelling-support/25123

Linked PRs

• gh-102983: queue.Queue - Adding Cancelling Support #102996

[mementum]

`init(self, maxsize=0)__ would get the following added

        # Support cancelling queues blocking on get
        self.cancelling = threading.Semaphore()

When the not_empty Condition is waited upon and the notified, the semaphore is checked for acquisition.

    def get(self, block=True, timeout=None):
        '''Remove and return an item from the queue.

        If optional args 'block' is true and 'timeout' is None (the default),
        block if necessary until an item is available. If 'timeout' is
        a non-negative number, it blocks at most 'timeout' seconds and raises
        the Empty exception if no item was available within that time.
        Otherwise ('block' is false), return an item if one is immediately
        available, else raise the Empty exception ('timeout' is ignored
        in that case).
        '''
        with self.not_empty:
            if not block:
                if not self._qsize():
                    raise Empty
            elif timeout is None:
                while not self._qsize():
                    self.not_empty.wait()
                    if self.cancelling.acquire(blocking=False):  # cancel?
                        raise Cancelled  # if decrement sem counter: cancelled
            elif timeout < 0:
                raise ValueError("'timeout' must be a non-negative number")
            else:
                endtime = time() + timeout
                while not self._qsize():
                    remaining = endtime - time()
                    if remaining <= 0.0:
                        raise Empty
                    self.not_empty.wait(remaining)
                    if self.cancelling.acquire(blocking=False):  # cancel?
                        raise Cancelled  # if decrement sem counter: cancelled

            item = self._get()
            self.not_full.notify()
            return item

And the cancel and cancel_all methods use the release method of the Semaphore to signal cancellation, notifying the waiting getters.

    def cancel_all(self):
        '''Cancel all get(block=True) calls

        This method cancels alls of the get(block=True) calls waiting for an
        item
        '''
        with self.not_empty:  # access to private len - lock is held: ok
            self.cancelling.release(n=len(self.not_empty._waiters))
            self.not_empty.notify_all()

    def cancel(self, n=1):
        '''Cancel one or more get(block=True) calls

        This method cancels at most n of the get(block=True) calls waiting for
        an item
        '''
        with self.not_empty:
            self.cancelling.release(n=n)
            self.not_empty.notify(n=n)

[mementum]

The get method can actually be improved a lot by unifying the wait calls to the not_empty Condition. Like this.

    def get(self, block=True, timeout=None):
        '''Remove and return an item from the queue.

        If optional args 'block' is true and 'timeout' is None (the default),
        block if necessary until an item is available. If 'timeout' is
        a non-negative number, it blocks at most 'timeout' seconds and raises
        the Empty exception if no item was available within that time.
        Otherwise ('block' is false), return an item if one is immediately
        available, else raise the Empty exception ('timeout' is ignored
        in that case).
        '''
        with self.not_empty:
            if not block:
                if not self._qsize():
                    raise Empty
            elif timeout is not None and timeout < 0:
                raise ValueError("'timeout' must be a non-negative number")

            else:  # timeout is either None or >= 0
                while not self._qsize():
                    if not self.not_empty.wait(timeout=timeout):
                        raise Empty
                    if self.cancelling.acquire(blocking=False):  # cancel?
                        raise Cancelled  # if decrement sem counter: cancelled

            item = self._get()
            self.not_full.notify()
            return item

All the timing calculations are delegated to the wait method which has the same algorithm (this approach could incidentally also be used in the threading module)

[mementum]

And considering that threading.Condition has a wait_for which in addition to timeout checks a predicate, one could even think of this code for the waiting part of get

            else:  # timeout is either None or >= 0
                if not self.not_empty.wait_for(self._qsize, timeout=timeout):
                    raise Empty
                if self.cancelling.acquire(blocking=False):  # cancel?
                    raise Cancelled  # if decrement sem counter: cancelled

[mementum]

The put method of queue.Queue could also be cancelled, either with the same Semaphore or with a separate one, in which case one would have to consider the naming for the cancelling methods, for example:

• cancel_get and cancel_get_all
• cancel_put and cancel_put_all

[mementum]

And the exception that would be raised ...

class Cancelled(Exception):
    'Exception raised by Queue.get(block=True) when cancelled'
    pass

[mementum]

Small test script for the proposed draft PR: #102996

from queue import Empty, Queue
import threading
import time


class Cancelled(Exception):
    'Exception raised by Queue.get(block=True) when cancelled'
    pass


class Queue(Queue):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # Support cancelling queues blocking on get, start with 0 cancellations
        self.cancelling = threading.Semaphore(value=0)

    def cancel_all(self):
        '''Cancel all get(block=True) calls

        This method cancels alls of the get(block=True) calls waiting for an
        item
        '''
        with self.not_empty:  # access to private len - lock is held: ok
            self.cancelling.release(n=len(self.not_empty._waiters))
            self.not_empty.notify_all()

    def cancel(self, n=1):
        '''Cancel one or more get(block=True) calls

        This method cancels at most n of the get(block=True) calls waiting for
        an item
        '''
        with self.not_empty:
            self.cancelling.release(n=n)
            self.not_empty.notify(n=n)

    def get(self, block=True, timeout=None):
        '''Remove and return an item from the queue.

        If optional args 'block' is true and 'timeout' is None (the default),
        block if necessary until an item is available. If 'timeout' is
        a non-negative number, it blocks at most 'timeout' seconds and raises
        the Empty exception if no item was available within that time.
        Otherwise ('block' is false), return an item if one is immediately
        available, else raise the Empty exception ('timeout' is ignored
        in that case).
        '''
        with self.not_empty:
            if not block:
                if not self._qsize():
                    raise Empty
            elif timeout is not None and timeout < 0:
                raise ValueError("'timeout' must be a non-negative number")

            else:  # timeout is either None or >= 0
                while not self._qsize():
                    if not self.not_empty.wait(timeout=timeout):
                        raise Empty
                    if self.cancelling.acquire(blocking=False):  # cancel?
                        raise Cancelled  # if decrement sem counter: cancelled

            item = self._get()
            self.not_full.notify()
            return item


# -----------------------------------------------------------------------------
def wait_for_cancel(i, q):
    try:
        item = q.get()
    except Cancelled:
        pass
    else:
        print(f'[+] {i}: this thread GOT an item: {item}')


COUNT = 100

for N in (0, 5):
    print('-' * 79)
    print(f'Try with {COUNT} threads and {COUNT - N - 1} cancellations')
    q = Queue()

    ths = []
    for i in range(COUNT):
        th = threading.Thread(
            target=wait_for_cancel,
            args=(i, q),
        )
        th.start()
        ths.append(th)

    print(f'[+] {COUNT} threads created and started')
    print('[+] Awaiting all of them blocking on get')
    # check all threads are blocking on get
    while len(q.not_empty._waiters) < COUNT:
        time.sleep(0.1)

    print('[+] All blocking on get. adding one item to check retrieval')
    q.put(None)

    print('[+] All - 1 blocking on get. CANCELLING')
    q.cancel(n=COUNT - N - 1)

    print('[+] Joining all threads')
    for i, th in enumerate(ths):
        th.join(timeout=1.0)
        if th.is_alive():
            print(f'[*] {i}: Failed to join thread')
            print(f'[*] {i}: Adding one cancel')
            q.cancel()

    print('[+] joined all threads -1 successfully')

Output

-------------------------------------------------------------------------------
Try with 100 threads and 99 cancellations
[+] 100 threads created and started
[+] Awaiting all of them blocking on get
[+] All blocking on get. adding one item to check retrieval
[+] All - 1 blocking on get. CANCELLING
[+] 0: this thread GOT an item: None
[+] Joining all threads
[+] joined all threads -1 successfully
-------------------------------------------------------------------------------
Try with 100 threads and 94 cancellations
[+] 100 threads created and started
[+] Awaiting all of them blocking on get
[+] All blocking on get. adding one item to check retrieval
[+] All - 1 blocking on get. CANCELLING
[+] 0: this thread GOT an item: None
[+] Joining all threads
[*] 95: Failed to join thread
[*] 95: Adding one cancel
[*] 96: Failed to join thread
[*] 96: Adding one cancel
[*] 97: Failed to join thread
[*] 97: Adding one cancel
[*] 98: Failed to join thread
[*] 98: Adding one cancel
[*] 99: Failed to join thread
[*] 99: Adding one cancel
[+] joined all threads -1 successfully