mengw15 opened a new issue, #6905:
URL: https://github.com/apache/texera/issues/6905
### What happened?
`LinkedBlockingMultiQueue.add_sub_queue` keeps `priority_groups` unsorted
when a higher-priority group is registered after a lower-priority one.
`DefaultSubQueueSelection.get_next`/`peek` walk that list in order and never
compare `pg.priority`, so list order *is* the priority order — an unsorted list
silently disables prioritisation.
```python
#
amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py
elif pg.priority > priority:
new_pg = LinkedBlockingMultiQueue.PriorityGroup(priority)
new_pg.add_queue(sub_queue)
self.priority_groups.append(new_pg) # appends to the end; should
insert at i
added = True
break
i += 1 # i is computed for exactly this insert and is otherwise
unused
```
This branch fires only when the new group belongs *before* the scanned one,
which is precisely when `append` is wrong. The `not added` fallback below it
appends correctly, because there the new group really is the lowest priority.
Impact on the worker: `InternalQueue` registers `SYSTEM=0`, control channels
`=1` and data channels `=2`, and channels are registered lazily on first `put`.
If any data channel is registered before its control channel, `priority_groups`
becomes `[0, 2, 1]` and control-channel elements are served *after*
data-channel ones. Since control carries pause/resume and statistics, a
paused-or-queried worker's response time then depends on how much data is
backed up. It fails silently — no exception, no log.
This is already known but untracked: #6444 added
`test_control_elements_dequeue_before_data_even_if_data_channel_registered_first`
in `test_internal_queue.py` marked `@pytest.mark.xfail` with this exact root
cause in its `reason`, and there is no issue for it.
### How to reproduce?
Both cases fail on `main` and pass once `append(new_pg)` becomes `insert(i,
new_pg)`:
```python
def test_groups_stay_sorted_when_a_higher_priority_arrives_later():
q = LinkedBlockingMultiQueue()
q.add_sub_queue("p0", 0)
q.add_sub_queue("p2", 2)
q.add_sub_queue("p1", 1) # higher priority, registered later
assert [pg.priority for pg in q.priority_groups] == [0, 1, 2]
def test_control_is_served_before_data_regardless_of_registration_order():
q = LinkedBlockingMultiQueue()
q.add_sub_queue("SYSTEM", 0)
q.add_sub_queue("data-chan", 2) # data channel registers first
q.add_sub_queue("control-chan", 1) # its control channel registers
second
q.put("data-chan", "DATA")
q.put("control-chan", "CONTROL")
assert [q.get(), q.get()] == ["CONTROL", "DATA"]
```
Equivalently, dropping the `xfail` from the #6444 test above makes it fail
on `main`.
Suggested fix: `self.priority_groups.insert(i, new_pg)`, and remove that
`xfail` so the existing test guards it.
### Version/Branch
main
### What browsers are you seeing the problem on?
N/A — Python worker (pyamber).
### Relevant log output
```shell
# on main
priority_groups = [0, 2, 1]
[q.get(), q.get()] -> ['DATA', 'CONTROL']
E AssertionError: assert ['DATA', 'CONTROL'] == ['CONTROL', 'DATA']
E At index 0 diff: 'DATA' != 'CONTROL'
# after the one-line fix (all 24 permutations of 4 priorities sort correctly)
priority_groups = [0, 1, 2]
[q.get(), q.get()] -> ['CONTROL', 'DATA']
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
