This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6903-4af3a85bcd4ecd3c1f8b5588ed4a1e25361f7ef7 in repository https://gitbox.apache.org/repos/asf/texera.git
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
