This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch release/v1.2
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/release/v1.2 by this push:
new 7106df4963 fix(pyamber, v1.2): repair LinkedBlockingMultiQueue removal
paths; add unit tests (#7101)
7106df4963 is described below
commit 7106df496367311d8e4b8325da4cd62a0561f132
Author: Yicong Huang <[email protected]>
AuthorDate: Thu Jul 30 03:42:51 2026 -0400
fix(pyamber, v1.2): repair LinkedBlockingMultiQueue removal paths; add unit
tests (#7101)
### What changes were proposed in this PR?
Backport of #6903 to `release/v1.2`, cherry-picked from
38d94ead0ec267fb205857d16b61a1eed36172df.
Follows the Direct Backport Push convention; opened as a PR (rather than
a direct push) as part of a backport-coverage audit for fixes merged to
`main` since early June that were never labeled for backport.
### Any related issues, documentation, discussions?
Backport of #6903. Originally linked #6899.
### How was this PR tested?
Release-branch CI runs on this PR. The cherry-pick applied cleanly onto
`release/v1.2`; no manual conflict resolution was needed.
### Was this PR authored or co-authored using generative AI tooling?
Yes — backport prepared with Claude Code (mechanical cherry-pick; the
change itself is #6903 by its original author).
Co-authored-by: Meng Wang <[email protected]>
---
.../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
