This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 38d94ead0e fix(pyamber): repair LinkedBlockingMultiQueue removal
paths; add unit tests (#6903)
38d94ead0e is described below
commit 38d94ead0ec267fb205857d16b61a1eed36172df
Author: Meng Wang <[email protected]>
AuthorDate: Sat Jul 25 23:03:19 2026 -0700
fix(pyamber): repair LinkedBlockingMultiQueue removal paths; add unit tests
(#6903)
### What changes were proposed in this PR?
`LinkedBlockingMultiQueue` sits at ~70% line coverage, and writing tests
for the untested parts surfaced two real defects — both on the removal
paths, which is why those lines were never exercised. This fixes both
and adds the tests that pin the corrected behavior.
**`SubQueue.remove` corrupted the chain and desynchronised the count.**
`head` is a sentinel, so the candidate node is `trail.next`, but
`remove` matched `trail.item` (the predecessor's own item) and then
unlinked `trail.next`; `unlink` in turn cleared `trail.item` instead of
the removed node's, and never decremented `self.count`. On `["a", "b",
"c"]`, `remove("a")` returned `True` but left a `None` in place of `"a"`
and dropped `"b"` entirely, left `SubQueue.size()` at 3 while the queue
total went to 2, and made the next `get()` return `None` instead of an
item. Fixed by matching `trail.next.item`, clearing `next_.item`, and
decrementing `self.count` in `unlink`.
**`remove_sub_queue` raised `AttributeError` for any enabled
sub-queue.** `add_sub_queue` constructed priority groups through class
access (`LinkedBlockingMultiQueue.PriorityGroup(...)`), and the `@inner`
descriptor only rebinds `owner` to the outer instance on instance access
— so `PriorityGroup.owner` stayed the class and
`self.owner.total_count.get_and_dec(...)` in `remove_queue` failed with
`type object 'LinkedBlockingMultiQueue' has no attribute 'total_count'`.
Fixed by constructing through `self.PriorityGroup(...)`, matching how
`SubQueue` is already built. `Node` and `DefaultSubQueueSelection` are
also built through class access but never read `self.owner`, so they are
left unchanged.
Neither method is reachable from production code today —
`internal_queue.py`, the only caller of this class, uses
`put`/`get`/`size`/`enable`/`disable`/`is_empty`/`is_enabled`/`in_mem_size`/`add_sub_queue`
— so no existing caller changes behavior.
The spec grows by 30 tests covering `peek` at all three levels,
`SubQueue.clear`, the corrected removal paths,
`PriorityGroup.remove_queue`, `remove_sub_queue`, and the small guards;
the `queue` fixture moved to module scope so the new classes share it.
Two current behaviors are pinned rather than changed: `clear()` does not
reset `in_mem_size`, and `SubQueue.__str__` only supports `str` items.
### Any related issues, documentation, discussions?
Closes #6899.
### How was this PR tested?
The target file passes with 42 tests (12 existing, 30 new), and the full
`pytest -m "not integration"` suite passes with no regressions; `ruff
check` and `ruff format --check` are clean. Both defects were reproduced
before the fix and confirmed gone after. For the failure path, each fix
was reverted in turn: reverting the `remove` fix reddens 4 removal
tests, and reverting the owner-binding fix brings back the original
`AttributeError` — so the new tests genuinely guard both fixes.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5)
---
.../linked_blocking_multi_queue.py | 11 +-
.../test_linked_blocking_multi_queue.py | 275 ++++++++++++++++++++-
2 files changed, 275 insertions(+), 11 deletions(-)
diff --git
a/amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py
b/amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py
index 735f0f6dc0..fcf175d6bc 100644
---
a/amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py
+++
b/amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py
@@ -154,9 +154,11 @@ class LinkedBlockingMultiQueue(IKeyedQueue):
return False
self.fully_lock()
try:
+ # head is a sentinel, so the candidate node is always
+ # trail.next and trail is its predecessor.
trail = self.head
while trail.next is not None:
- if trail.item == obj:
+ if trail.next.item == obj:
self.unlink(trail, trail.next)
return True
trail = trail.next
@@ -169,10 +171,11 @@ class LinkedBlockingMultiQueue(IKeyedQueue):
trail: LinkedBlockingMultiQueue.Node,
next_: LinkedBlockingMultiQueue.Node,
) -> None:
- trail.item = None
+ next_.item = None
trail.next = next_.next
if self.last == next_:
self.last = trail
+ self.count.dec()
if self.enabled:
self.owner.total_count.get_and_dec()
@@ -418,7 +421,7 @@ class LinkedBlockingMultiQueue(IKeyedQueue):
added = True
break
elif pg.priority > priority:
- new_pg =
LinkedBlockingMultiQueue.PriorityGroup(priority)
+ new_pg = self.PriorityGroup(priority)
new_pg.add_queue(sub_queue)
self.priority_groups.append(new_pg)
added = True
@@ -426,7 +429,7 @@ class LinkedBlockingMultiQueue(IKeyedQueue):
i += 1
if not added:
- new_pg = LinkedBlockingMultiQueue.PriorityGroup(priority)
+ new_pg = self.PriorityGroup(priority)
new_pg.add_queue(sub_queue)
self.priority_groups.append(new_pg)
diff --git
a/amber/src/test/python/core/util/customized_queue/test_linked_blocking_multi_queue.py
b/amber/src/test/python/core/util/customized_queue/test_linked_blocking_multi_queue.py
index 1df9423b69..8cdaada6bd 100644
---
a/amber/src/test/python/core/util/customized_queue/test_linked_blocking_multi_queue.py
+++
b/amber/src/test/python/core/util/customized_queue/test_linked_blocking_multi_queue.py
@@ -25,14 +25,26 @@ from core.util.customized_queue.linked_blocking_multi_queue
import (
)
-class TestLinkedBlockingMultiQueue:
- @pytest.fixture
- def queue(self):
- lbmq = LinkedBlockingMultiQueue()
- lbmq.add_sub_queue("control", 0)
- lbmq.add_sub_queue("data", 1)
- return lbmq
[email protected]
+def queue():
+ """A queue with `control` at priority 0 and `data` at priority 1."""
+ lbmq = LinkedBlockingMultiQueue()
+ lbmq.add_sub_queue("control", 0)
+ lbmq.add_sub_queue("data", 1)
+ return lbmq
+
+
+def chain_items(sub_queue):
+ """The items still reachable from a SubQueue's sentinel head, in order."""
+ items = []
+ node = sub_queue.head.next
+ while node is not None:
+ items.append(node.item)
+ node = node.next
+ return items
+
+class TestLinkedBlockingMultiQueue:
def test_sub_can_emit(self, queue):
assert queue.is_empty()
queue.put("data", 1)
@@ -235,3 +247,252 @@ class TestLinkedBlockingMultiQueue:
assert total == sum(filter(lambda x: x % 3 != 0, range(11)))
reraise()
+
+
+class TestPeek:
+ """peek() at all three levels. It never waits —
LinkedBlockingMultiQueue.peek
+ short-circuits on total_count == 0 — so these are safe single-threaded."""
+
+ def test_returns_none_on_empty_queue(self, queue):
+ assert queue.peek() is None
+
+ def test_does_not_consume_the_item(self, queue):
+ queue.put("data", 1)
+ assert queue.peek() == 1
+ assert queue.peek() == 1
+ assert queue.size() == 1
+ assert queue.get() == 1
+
+ def test_prefers_the_higher_priority_sub_queue(self, queue):
+ queue.put("data", 1)
+ queue.put("control", "s")
+ assert queue.peek() == "s"
+
+ def test_returns_none_when_the_only_non_empty_sub_queue_is_disabled(self,
queue):
+ queue.put("data", 1)
+ queue.disable("data")
+ assert queue.peek() is None
+
+ def test_priority_group_peek_perturbs_round_robin_state(self):
+ lbmq = LinkedBlockingMultiQueue()
+ lbmq.add_sub_queue("first", 1)
+ lbmq.add_sub_queue("second", 1)
+ group = lbmq.get_sub_queue("first").priority_group
+ lbmq.put("second", "s")
+
+ assert group.next_idx == 0
+ assert group.peek() == "s"
+ # Skipping the empty `first` queue advanced next_idx, so peek is not
+ # side-effect free on the group's round-robin cursor.
+ assert group.next_idx == 1
+
+ def test_priority_group_peek_returns_none_when_all_queues_empty(self,
queue):
+ group = queue.get_sub_queue("control").priority_group
+ assert group.peek() is None
+
+ def test_selection_peek_returns_first_non_none_across_groups(self, queue):
+ queue.put("data", 1)
+ assert queue.sub_queue_selection.peek() == 1
+
+
+class TestSubQueueClear:
+ def test_clear_resets_count_and_collapses_the_chain(self, queue):
+ queue.put("data", 1)
+ queue.put("data", 2)
+ sub = queue.get_sub_queue("data")
+
+ sub.clear()
+
+ assert sub.size() == 0
+ assert sub.is_empty()
+ assert sub.head is sub.last
+ assert chain_items(sub) == []
+ assert queue.size() == 0
+
+ def test_clear_leaves_total_count_alone_when_disabled(self, queue):
+ queue.put("data", 1)
+ queue.put("control", "s")
+ queue.disable("data")
+ assert queue.size() == 1
+
+ queue.get_sub_queue("data").clear()
+
+ assert queue.size() == 1
+
+ def test_clear_does_not_reset_in_mem_size(self, queue):
+ queue.put("data", 1)
+ before = queue.in_mem_size("data")
+ assert before > 0
+
+ queue.get_sub_queue("data").clear()
+
+ # Current behavior: clear() drops the items but not the in-memory
+ # size accounting.
+ assert queue.in_mem_size("data") == before
+
+
+class TestSubQueueRemove:
+ def test_remove_none_returns_false(self, queue):
+ assert queue.get_sub_queue("data").remove(None) is False
+
+ def test_remove_missing_item_returns_false(self, queue):
+ queue.put("data", "a")
+ sub = queue.get_sub_queue("data")
+
+ assert sub.remove("missing") is False
+ assert chain_items(sub) == ["a"]
+ assert sub.size() == queue.size() == 1
+
+ def test_remove_first_item(self, queue):
+ for item in ["a", "b", "c"]:
+ queue.put("data", item)
+ sub = queue.get_sub_queue("data")
+
+ assert sub.remove("a") is True
+ assert chain_items(sub) == ["b", "c"]
+ assert sub.size() == queue.size() == 2
+ assert queue.get() == "b"
+
+ def test_remove_middle_item(self, queue):
+ for item in ["a", "b", "c"]:
+ queue.put("data", item)
+ sub = queue.get_sub_queue("data")
+
+ assert sub.remove("b") is True
+ assert chain_items(sub) == ["a", "c"]
+ assert sub.size() == queue.size() == 2
+
+ def test_remove_last_item_fixes_up_the_tail(self, queue):
+ for item in ["a", "b", "c"]:
+ queue.put("data", item)
+ sub = queue.get_sub_queue("data")
+
+ assert sub.remove("c") is True
+ assert chain_items(sub) == ["a", "b"]
+ assert sub.last.item == "b"
+ # The tail pointer must still be usable for the next enqueue.
+ queue.put("data", "d")
+ assert chain_items(sub) == ["a", "b", "d"]
+
+ def test_remove_only_item_collapses_the_chain(self, queue):
+ queue.put("data", "only")
+ sub = queue.get_sub_queue("data")
+
+ assert sub.remove("only") is True
+ assert sub.head is sub.last
+ assert sub.is_empty()
+ assert queue.size() == 0
+
+ def test_remove_on_disabled_sub_queue_leaves_total_count(self, queue):
+ queue.put("data", "a")
+ queue.put("data", "b")
+ queue.disable("data")
+ assert queue.size() == 0
+
+ assert queue.get_sub_queue("data").remove("a") is True
+
+ # total_count already excludes a disabled sub-queue, so only the
+ # sub-queue's own count moves.
+ assert queue.size() == 0
+ assert queue.size("data") == 1
+ queue.enable("data")
+ assert queue.size() == 1
+
+
+class TestPriorityGroupRemoveQueue:
+ def test_remove_queue_shrinks_the_group(self, queue):
+ sub = queue.get_sub_queue("data")
+ group = sub.priority_group
+
+ group.remove_queue(sub)
+
+ assert group.queues == []
+
+ def test_remove_queue_resets_next_idx_when_it_reached_the_end(self):
+ lbmq = LinkedBlockingMultiQueue()
+ lbmq.add_sub_queue("first", 1)
+ lbmq.add_sub_queue("second", 1)
+ group = lbmq.get_sub_queue("first").priority_group
+ lbmq.put("first", "a")
+ assert lbmq.get() == "a"
+ assert group.next_idx == 1
+
+ group.remove_queue(lbmq.get_sub_queue("second"))
+
+ assert len(group.queues) == 1
+ assert group.next_idx == 0
+
+ def test_remove_queue_ignores_a_non_matching_key(self, queue):
+ control_group = queue.get_sub_queue("control").priority_group
+
+ # `data` lives in a different group, so this is a no-op.
+ control_group.remove_queue(queue.get_sub_queue("data"))
+
+ assert [q.key for q in control_group.queues] == ["control"]
+
+
+class TestRemoveSubQueue:
+ def test_missing_key_returns_none_and_mutates_nothing(self, queue):
+ assert queue.remove_sub_queue("nope") is None
+ assert sorted(queue.sub_queues) == ["control", "data"]
+ assert len(queue.priority_groups) == 2
+
+ def test_removes_an_enabled_sub_queue_and_its_items(self, queue):
+ queue.put("control", "s")
+ queue.put("data", 1)
+
+ removed = queue.remove_sub_queue("control")
+
+ assert removed is not None
+ assert removed.key == "control"
+ assert sorted(queue.sub_queues) == ["data"]
+ # The removed sub-queue's item no longer counts toward the total.
+ assert queue.size() == 1
+
+ def test_drops_the_priority_group_once_it_is_empty(self, queue):
+ queue.remove_sub_queue("control")
+
+ assert len(queue.priority_groups) == 1
+ assert queue.priority_groups[0].priority == 1
+
+
+class TestSmallGuards:
+ def test_is_empty_tracks_size(self, queue):
+ sub = queue.get_sub_queue("data")
+ assert sub.is_empty()
+ queue.put("data", 1)
+ assert not sub.is_empty()
+
+ def test_str_renders_an_arrow_chain(self, queue):
+ sub = queue.get_sub_queue("control")
+ assert str(sub) == ""
+ for item in ["a", "b", "c"]:
+ queue.put("control", item)
+ assert str(sub) == "a -> b -> c -> "
+
+ def test_str_only_supports_str_items(self, queue):
+ queue.put("data", 1)
+ with pytest.raises(TypeError):
+ str(queue.get_sub_queue("data"))
+
+ def test_put_none_raises_value_error(self, queue):
+ with pytest.raises(ValueError, match="Does not support NoneType."):
+ queue.put("data", None)
+
+ def test_get_next_returns_none_when_everything_is_empty(self, queue):
+ # Called directly: q.get() on an empty queue would block forever.
+ assert queue.sub_queue_selection.get_next() is None
+
+ def test_set_priority_groups_to_empty(self, queue):
+ queue.put("data", 1)
+ queue.sub_queue_selection.set_priority_groups([])
+
+ assert queue.sub_queue_selection.priority_groups == []
+ assert queue.sub_queue_selection.get_next() is None
+ assert queue.sub_queue_selection.peek() is None
+
+ def test_len_matches_size(self, queue):
+ assert len(queue) == 0
+ queue.put("data", 1)
+ queue.put("control", "s")
+ assert len(queue) == queue.size() == 2
