A Python worker can serve data ahead of control messages, delaying pause and resume

[mengw15]

What happened?

Every Python worker sorts its incoming messages into lanes and always serves the most urgent lane first:

lane	priority	carries
SYSTEM	0	the worker's own shutdown signal
control	1	pause, resume, statistics queries
data	2	tuple batches

Sometimes it serves them in the wrong order and control ends up behind data. When that happens, pausing a worker only takes effect once its data backlog has drained. Nothing raises and nothing is logged — the worker just answers late.

What goes wrong. The lanes live in a plain Python list, and the worker picks the next lane by walking that list from the front. The priority number is read only when a lane is created; from then on, list position is the priority.

A lane is created the first time one of its channels sends something, and new lanes are appended to the end of the list — even when they belong in front. So the order in which channels happen to wake up decides the order in which they get served:

a data channel sends before its control channel

   lanes = [ SYSTEM , data , control ]
                        ^ now served ahead of control

The lanes come out correctly ordered only if channels happen to register from most urgent to least. For four lanes, 23 of the 24 possible registration orders leave at least one lane out of place.

It then compounds. Deciding whether a lane for a given priority already exists is done by the same front-to-back scan: it stops at the first lane whose priority number is larger, and on a sorted list that is enough to prove no matching lane exists. On an unsorted list it is not. So a second control channel opens a second control lane instead of joining the first:

   lanes = [ SYSTEM , data , control-A , control-B ]
                             ^^^^^^^^^^^^^^^^^^^^^ two lanes at the same priority

Channels that share a lane are served in rotation; channels in separate lanes are served strictly front to back. So control-A now starves control-B:

control-A and control-B each holding 3 messages, six gets

   actual:    A A A B B B
   expected:  A B A B A B

The line. In add_sub_queue:

# amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py
i = 0
for pg in self.priority_groups:
    if pg.priority == priority:                # a lane with this priority exists
        pg.add_queue(sub_queue)
        ...
        break
    elif pg.priority > priority:               # the new lane belongs BEFORE pg
        ...
        self.priority_groups.append(new_pg)    # <-- but append puts it last
        ...
        break
    i += 1                                     # counted here, read nowhere

if not added:                                  # loop ran to the end: the new lane
    ...                                        # really is the least urgent one
    self.priority_groups.append(new_pg)        # <-- correct here

The two append calls look identical but sit in different situations. The second one runs only after the loop has gone all the way through, so the new lane genuinely belongs last. The first one runs on a break from the middle of the loop, with lanes still unscanned behind it — there the new lane belongs at position i, which is what the otherwise-unused i counter was being maintained for.

Reported once already, as a skipped test. #6444 added test_control_elements_dequeue_before_data_even_if_data_channel_registered_first to test_internal_queue.py and marked it xfail, naming this exact cause in the reason. No issue was filed, so it has been sitting there green-but-skipped ever since.

How to reproduce?

All three fail on main, and all three pass once that first append(new_pg) becomes insert(i, new_pg):

def test_lanes_stay_sorted_when_a_more_urgent_one_arrives_later():
    q = LinkedBlockingMultiQueue()
    q.add_sub_queue("p0", 0)
    q.add_sub_queue("p2", 2)
    q.add_sub_queue("p1", 1)   # more urgent than p2, but registered after it
    assert [pg.priority for pg in q.priority_groups] == [0, 1, 2]


def test_control_is_served_first_whatever_the_registration_order():
    q = LinkedBlockingMultiQueue()
    q.add_sub_queue("SYSTEM", 0)
    q.add_sub_queue("data-chan", 2)      # the data channel sends first
    q.add_sub_queue("control-chan", 1)   # its control channel sends second
    q.put("data-chan", "DATA")
    q.put("control-chan", "CONTROL")
    assert [q.get(), q.get()] == ["CONTROL", "DATA"]


def test_channels_at_the_same_priority_share_one_lane():
    q = LinkedBlockingMultiQueue()
    q.add_sub_queue("SYSTEM", 0)
    q.add_sub_queue("data-chan", 2)    # puts the list out of order
    q.add_sub_queue("control-a", 1)
    q.add_sub_queue("control-b", 1)    # should join control-a's lane, not open a new one
    assert [pg.priority for pg in q.priority_groups] == [0, 1, 2]

    for _ in range(2):
        q.put("control-a", "A")
        q.put("control-b", "B")
    # channels sharing a lane rotate; they are not drained one after the other
    assert [q.get() for _ in range(4)] == ["A", "B", "A", "B"]

Or simply drop the xfail from the #6444 test above and watch it fail.

Suggested fix. Change that first self.priority_groups.append(new_pg) to self.priority_groups.insert(i, new_pg), and drop the xfail so the existing test guards it from now on.

Version/Branch

main

What browsers are you seeing the problem on?

N/A — Python worker (pyamber).

Relevant log output

# on main
lanes = [0, 2, 1]
[q.get(), q.get()] -> ['DATA', 'CONTROL']

E   AssertionError: assert ['DATA', 'CONTROL'] == ['CONTROL', 'DATA']
E     At index 0 diff: 'DATA' != 'CONTROL'

# on main, with a second control channel — note the duplicated priority 1
lanes = [0, 2, 1, 1]
six gets from two control channels -> ['A', 'A', 'A', 'B', 'B', 'B']

# after the one-line fix — all 24 registration orders come out sorted
lanes = [0, 1, 2]
[q.get(), q.get()] -> ['CONTROL', 'DATA']
six gets from two control channels -> ['A', 'B', 'A', 'B', 'A', 'B']

[chenlica]

@mengw15 Please make the subject and description more human readable.