fix(pyamber): keep LinkedBlockingMultiQueue priority groups sorted#6906
fix(pyamber): keep LinkedBlockingMultiQueue priority groups sorted#6906mengw15 wants to merge 2 commits into
Conversation
Automated Reviewer SuggestionsBased on the
|
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #6906 +/- ##
=========================================
Coverage 78.58% 78.58%
+ Complexity 3731 3723 -8
=========================================
Files 1161 1161
Lines 46066 46066
Branches 5106 5106
=========================================
Hits 36199 36199
+ Misses 8258 8256 -2
- Partials 1609 1611 +2
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 423 | 0.258 | 23,890/26,609/26,609 us | 🔴 +24.6% / 🔴 +86.4% |
| 🔴 | bs=100 sw=10 sl=64 | 914 | 0.558 | 109,946/147,148/147,148 us | 🔴 +21.0% / 🔴 +34.9% |
| 🔴 | bs=1000 sw=10 sl=64 | 1,076 | 0.656 | 924,299/1,060,247/1,060,247 us | 🔴 +8.1% / 🟢 -7.8% |
Baseline details
Latest main 03a0e3a from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 423 tuples/sec | 489 tuples/sec | 754.55 tuples/sec | -13.5% | -43.9% |
| bs=10 sw=10 sl=64 | MB/s | 0.258 MB/s | 0.298 MB/s | 0.461 MB/s | -13.4% | -44.0% |
| bs=10 sw=10 sl=64 | p50 | 23,890 us | 19,170 us | 12,816 us | +24.6% | +86.4% |
| bs=10 sw=10 sl=64 | p95 | 26,609 us | 26,524 us | 16,594 us | +0.3% | +60.3% |
| bs=10 sw=10 sl=64 | p99 | 26,609 us | 26,524 us | 19,806 us | +0.3% | +34.3% |
| bs=100 sw=10 sl=64 | throughput | 914 tuples/sec | 965 tuples/sec | 969.38 tuples/sec | -5.3% | -5.7% |
| bs=100 sw=10 sl=64 | MB/s | 0.558 MB/s | 0.589 MB/s | 0.592 MB/s | -5.3% | -5.7% |
| bs=100 sw=10 sl=64 | p50 | 109,946 us | 101,952 us | 103,584 us | +7.8% | +6.1% |
| bs=100 sw=10 sl=64 | p95 | 147,148 us | 121,602 us | 109,097 us | +21.0% | +34.9% |
| bs=100 sw=10 sl=64 | p99 | 147,148 us | 121,602 us | 117,304 us | +21.0% | +25.4% |
| bs=1000 sw=10 sl=64 | throughput | 1,076 tuples/sec | 1,088 tuples/sec | 1,004 tuples/sec | -1.1% | +7.2% |
| bs=1000 sw=10 sl=64 | MB/s | 0.656 MB/s | 0.664 MB/s | 0.613 MB/s | -1.2% | +7.1% |
| bs=1000 sw=10 sl=64 | p50 | 924,299 us | 917,704 us | 1,002,357 us | +0.7% | -7.8% |
| bs=1000 sw=10 sl=64 | p95 | 1,060,247 us | 980,955 us | 1,046,463 us | +8.1% | +1.3% |
| bs=1000 sw=10 sl=64 | p99 | 1,060,247 us | 980,955 us | 1,073,661 us | +8.1% | -1.2% |
Raw CSV
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,473.00,200,128000,423,0.258,23889.88,26608.64,26608.64
1,100,10,64,20,2189.00,2000,1280000,914,0.558,109945.55,147147.87,147147.87
2,1000,10,64,20,18594.55,20000,12800000,1076,0.656,924298.89,1060246.64,1060246.64There was a problem hiding this comment.
Pull request overview
This PR fixes Python worker queue prioritization by ensuring LinkedBlockingMultiQueue.add_sub_queue keeps priority_groups sorted when inserting a new higher-priority group, and it re-enables an existing InternalQueue unit test that previously had to be marked xfail due to the ordering bug.
Changes:
- Insert newly created
PriorityGroupat the correct index (insert(i, ...)) instead of appending, preserving sorted priority order. - Remove the
xfailmarker fromtest_control_elements_dequeue_before_data_even_if_data_channel_registered_firstnow that control-vs-data priority is correctly enforced.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py | Fixes priority_groups ordering by inserting new priority groups at the correct position. |
| amber/src/test/python/core/models/test_internal_queue.py | Drops xfail so the control-before-data dequeue behavior is enforced by tests. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
add_sub_queue appended a newly created PriorityGroup instead of inserting it at the scanned index, so registering a lower-priority group before a higher-priority one left priority_groups unsorted. DefaultSubQueueSelection walks that list in order without comparing priorities, so control-channel elements could be served after data-channel ones. Un-xfails the InternalQueue test that already covered this.
998eeb1 to
07c52c9
Compare
What changes were proposed in this PR?
add_sub_queueappended a newly createdPriorityGroupto the end ofpriority_groupsin the branch that fires when the new group belongs before the scanned one (pg.priority > priority), leaving the list unsorted. Theicounter maintained by that loop exists for exactly this insert and was otherwise unused.DefaultSubQueueSelection.get_next/peekiteratepriority_groupsin list order and never comparepg.priority, so list order is the priority order and an unsorted list silently disables prioritisation.For the worker this means:
InternalQueueregistersSYSTEM=0, control channels=1, data channels=2, and registers each channel lazily on its firstput. If any data channel is registered before its control channel,priority_groupsends up[0, 2, 1]and control-channel elements are served after data-channel ones — so a paused or queried worker's response time starts depending on how much data is backed up, with no exception and nothing in the logs. Thenot addedfallback further down keeps itsappend, which is correct there since that branch only runs when the new group really is the lowest priority.The fix is
insert(i, new_pg). No new test was needed: #6444 already addedtest_control_elements_dequeue_before_data_even_if_data_channel_registered_firstfor this exact scenario but had to mark it@pytest.mark.xfail, naming this root cause in itsreason; this PR drops that marker so the existing test guards the fix.Any related issues, documentation, discussions?
Closes #6905. The
xfailremoved here was introduced by #6444. Spotted by Copilot while reviewing #6903, which touches neighbouring lines in the same method but is otherwise unrelated (it repairs the removal paths).How was this PR tested?
pytest -m "not integration"passes with 849 tests and noxpassedleft;ruff checkandruff format --checkare clean. The un-xfailed test fails onmainand passes with the fix, and reverting the one-line change turns it red again, so it genuinely guards this line. Also checked all 24 registration orders of four priorities now sort correctly, and that same-priority sub-queues still share one group.Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5)