diff --git a/README.md b/README.md index 514f2b2..f7c13f9 100644 --- a/README.md +++ b/README.md @@ -78,7 +78,36 @@ q.get(task.message_id) - Timing metrics. As long as tasks are still in the queue or not pruned, you can see how long they have been there or how long they took to finish. - Easy to extend using SQL -## Queue capacity +## Queue size and capacity + +`qsize()`, `empty()`, `full()`, and `maxsize` all use the ready backlog: messages +that are waiting to be claimed. A message stops contributing to that backlog as +soon as `pop()` locks it for a worker. + +Use the explicit count methods when you need a different view: + +- `ready_count()` counts messages waiting to be claimed. +- `locked_count()` counts messages currently claimed by workers. +- `done_count()` counts completed messages that have not been pruned. +- `failed_count()` counts failed messages that have not been pruned. +- `active_count()` counts ready plus locked messages. +- `stored_count()` counts every stored row regardless of status. + +For example, active locked work does not make the ready backlog non-empty: + +```python +queue = LiteQueue(filename="tasks.sqlite3", maxsize=1) +queue.put("resize image") +task = queue.pop() + +assert task is not None +assert queue.qsize() == 0 +assert queue.empty() +assert not queue.full() +assert queue.locked_count() == 1 +assert queue.active_count() == 1 +assert queue.stored_count() == 1 +``` `maxsize` limits the number of ready messages and is stored as an immutable property of the queue. Omit `maxsize` when reopening a queue to use its stored @@ -194,6 +223,10 @@ distributions, and uploads them with uv. ## Important changes +- After version 0.12: + - `qsize()` now reports only the ready backlog instead of ready plus locked work. + - `empty()`, `full()`, and `maxsize` use the same ready-backlog definition. + - Explicit ready, locked, done, failed, active, and stored count methods are available. - In version 0.10: - Each SQLite database can contain only one LiteQueue queue. - You must migrate version 0.9 queues before you use version 0.10 or later. diff --git a/src/litequeue/__init__.py b/src/litequeue/__init__.py index 0319229..0596ecf 100644 --- a/src/litequeue/__init__.py +++ b/src/litequeue/__init__.py @@ -179,7 +179,6 @@ def __init__( compiled in serialized mode, as in standard CPython builds. LiteQueue serializes writes and explicit transactions while a read-only pool serves committed data to other threads. - """ filename_is_supported = isinstance(filename, (str, Path)) if not filename_is_supported: @@ -214,9 +213,7 @@ def __init__( cached_statements=0, **kwargs, ) - self.conn.row_factory = sqlite3.Row - self.table = f'"{_QUEUE_TABLE_NAME}"' with self.transaction(mode="IMMEDIATE"): @@ -266,7 +263,6 @@ def __init__( f'CREATE UNIQUE INDEX IF NOT EXISTS "Queue_message_id_unique_idx" ' f"ON {self.table}(message_id)" ) - self.conn.execute( f'CREATE INDEX IF NOT EXISTS "Queue_status_message_id_idx" ' f"ON {self.table}(status, message_id)" @@ -282,7 +278,6 @@ def __init__( f"maxsize {validated_maxsize} conflicts with stored maxsize " f"{stored_maxsize} for queue '{_QUEUE_TABLE_NAME}'" ) - effective_maxsize = stored_maxsize else: effective_maxsize = validated_maxsize @@ -382,10 +377,7 @@ def _select_pop_func(self) -> PopFunction: return self._pop_transaction def put(self, data: str) -> Message: - """ - Insert a new message - """ - # timeout: int = None + """Insert a new message.""" message_id = str(uuid7()) now = time_ns() @@ -498,7 +490,7 @@ def _read_connection(self) -> Iterator[sqlite3.Connection]: read_connections.put(read_connection) def peek(self) -> Message | None: - "Show next message to be popped, if any." + """Show next message to be popped, if any.""" with self._read_connection() as connection: value = connection.execute( @@ -508,7 +500,7 @@ def peek(self) -> Message | None: return _message_from_row(value) if value is not None else None def get(self, message_id: str) -> Message | None: - "Get a message by its `message_id`" + """Get a message by its `message_id`.""" with self._read_connection() as connection: value = connection.execute( @@ -576,7 +568,7 @@ def list_locked(self, threshold_seconds: int) -> Iterator[Message]: SELECT * FROM {self.table} WHERE status = {MessageStatus.LOCKED.value} - AND lock_time < :time_value + AND lock_time < :time_value """.strip(), {"time_value": time_ns() - threshold_nanoseconds}, ).fetchall() @@ -585,16 +577,13 @@ def list_locked(self, threshold_seconds: int) -> Iterator[Message]: yield _message_from_row(result) def list_failed(self) -> Iterator[Message]: - """ - Return all the tasks in `FAILED` state. - """ + """Return all the tasks in `FAILED` state.""" with self._read_connection() as connection: rows = connection.execute( f""" SELECT * FROM {self.table} - WHERE - status = {MessageStatus.FAILED.value} + WHERE status = {MessageStatus.FAILED.value} """.strip() ).fetchall() @@ -603,7 +592,7 @@ def list_failed(self) -> Iterator[Message]: def retry(self, message_id: str) -> bool: """ - Mark a locked message as free again. + Mark a locked or failed message as ready again. Return `True` when the message exists, otherwise `False`. """ @@ -621,52 +610,58 @@ def retry(self, message_id: str) -> bool: return cursor.rowcount > 0 - def qsize(self) -> int: - """ - Get current size of the queue. - """ + def _count_statuses(self, *statuses: MessageStatus) -> int: + """Return the number of stored messages in the requested states.""" + if not statuses: + return self.stored_count() + placeholders = ", ".join("?" for _ in statuses) with self._read_connection() as connection: - cursor = connection.execute( - f""" - SELECT COUNT(*) FROM {self.table} - WHERE status NOT IN ({MessageStatus.DONE.value}, {MessageStatus.FAILED.value}) - """.strip() - ) - size = next(cursor)[0] + value = connection.execute( + f"SELECT COUNT(*) FROM {self.table} WHERE status IN ({placeholders})", + tuple(status.value for status in statuses), + ).fetchone() + return int(value[0]) - return size + def ready_count(self) -> int: + """Return the number of messages waiting to be claimed.""" + return self._count_statuses(MessageStatus.READY) - def empty(self) -> bool: - """ - Return True if the queue is empty. - """ + def locked_count(self) -> int: + """Return the number of messages currently claimed by workers.""" + return self._count_statuses(MessageStatus.LOCKED) - with self._read_connection() as connection: - value = connection.execute( - f"SELECT COUNT(*) as cnt FROM {self.table} WHERE status = {MessageStatus.READY.value}" - ).fetchone() - return not bool(value["cnt"]) + def done_count(self) -> int: + """Return the number of completed messages still stored.""" + return self._count_statuses(MessageStatus.DONE) - def full(self) -> bool: - """ - Return True if the queue is full. - """ + def failed_count(self) -> int: + """Return the number of failed messages still stored.""" + return self._count_statuses(MessageStatus.FAILED) - # Here I need to check compared to the maxsize value - # If maxsize is not set, the queue can grow forever - if self.maxsize is None: - return False + def active_count(self) -> int: + """Return the number of ready and locked messages.""" + return self._count_statuses(MessageStatus.READY, MessageStatus.LOCKED) + def stored_count(self) -> int: + """Return the total number of rows stored for every message state.""" with self._read_connection() as connection: - value = connection.execute( - f"SELECT COUNT(*) as cnt FROM {self.table} WHERE status = {MessageStatus.READY.value}" - ).fetchone() + value = connection.execute(f"SELECT COUNT(*) FROM {self.table}").fetchone() + return int(value[0]) - if value["cnt"] >= self.maxsize: - return True - else: + def qsize(self) -> int: + """Return the ready backlog size.""" + return self.ready_count() + + def empty(self) -> bool: + """Return True when the ready backlog is empty.""" + return self.qsize() == 0 + + def full(self) -> bool: + """Return True when the ready backlog has reached `maxsize`.""" + if self.maxsize is None: return False + return self.qsize() >= self.maxsize def prune(self, include_failed: bool = True) -> None: """ diff --git a/tests/test_size_semantics.py b/tests/test_size_semantics.py new file mode 100644 index 0000000..831100c --- /dev/null +++ b/tests/test_size_semantics.py @@ -0,0 +1,46 @@ +from pathlib import Path + +from litequeue import LiteQueue + + +def test_locked_work_is_not_ready_backlog(tmp_path: Path) -> None: + queue = LiteQueue(filename=tmp_path / "queue.sqlite3", maxsize=1) + queue.put("work") + + locked = queue.pop() + + assert locked is not None + assert queue.qsize() == 0 + assert queue.empty() + assert not queue.full() + assert queue.ready_count() == 0 + assert queue.locked_count() == 1 + assert queue.active_count() == 1 + assert queue.stored_count() == 1 + + +def test_counts_cover_every_message_status(tmp_path: Path) -> None: + queue = LiteQueue(filename=tmp_path / "queue.sqlite3") + queue.put("locked") + queue.put("done") + queue.put("failed") + queue.put("ready") + + locked = queue.pop() + done = queue.pop() + failed = queue.pop() + + assert locked is not None + assert done is not None + assert failed is not None + queue.done(done.message_id) + queue.mark_failed(failed.message_id) + + assert queue.ready_count() == 1 + assert queue.locked_count() == 1 + assert queue.done_count() == 1 + assert queue.failed_count() == 1 + assert queue.qsize() == 1 + assert queue.active_count() == 2 + assert queue.stored_count() == 4 + assert not queue.empty()