Fixes #39228 - Introduce ad hoc host applicability scheduler#11709
Fixes #39228 - Introduce ad hoc host applicability scheduler#11709jturel wants to merge 7 commits into
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughRemoved the single-host Dynflow entry action and event-driven scheduling; introduced a queue-driven batch pipeline: queue inserts use dedupe/notify, a Scheduler service drains and dispatches Dynflow/BulkGenerate work, added a unique-index migration for the queue, and updated/added tests. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant Queue as ApplicableHostQueue
participant DB
participant Service as Katello::Applicability::Scheduler
participant Dynflow as Actions::Katello::Applicability::Scheduler
participant Bulk as Actions::Katello::Applicability::Hosts::BulkGenerate
Client->>Queue: push_hosts(host_ids)
Queue->>DB: insert_all(rows) / ensure unique_by host_id
Queue->>Service: notify "applicability_push_hosts"
alt Service handles immediately
Service->>Queue: pop_hosts(batch_size)
Queue->>DB: lock selected rows by id, collect host_ids, delete rows
Queue-->>Service: host_ids
Service->>Bulk: ForemanTasks.async_task(host_ids)
Bulk->>DB: load ContentFacets and calculate_and_import_applicability
else Service triggers Dynflow scheduler
Service->>Dynflow: ForemanTasks.async_task() (start Scheduler action)
Dynflow->>Service: invoke drain_loop
loop while queue not empty
Service->>Queue: pop_hosts(batch_size)
Queue->>DB: lock/delete rows, return host_ids
Service->>Bulk: ForemanTasks.async_task(host_ids)
Service->>Service: sleep(DRAIN_SLEEP_SECONDS) when appropriate
end
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Suggested reviewers
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Warning Review ran into problems🔥 ProblemsTimed out fetching pipeline failures after 30000ms Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
jturel
force-pushed
the
applicability_scheduler_ad_hoc
branch
2 times, most recently
from
f23d016 to
7804f9d
Compare
jturel
changed the title
| Rails.logger.warn(_("Content Facet for host with id %s is non-existent. Skipping applicability calculation.") % host_id) | ||
| end | ||
|
|
||
| ::Katello::Host::ContentFacet.includes(:host).where(host_id: input[:host_ids]).find_each do |content_facet| |
There was a problem hiding this comment.
This is a more efficient querying pattern that also avoids the need to check for some errors
| @@ -32,10 +28,6 @@ def queue | |||
| ::Katello::HOST_TASKS_QUEUE | |||
| end | |||
|
|
|||
| def resource_locks | |||
There was a problem hiding this comment.
This wasn't doing anything since this action doesn't lock any resources.
| include ::Dynflow::Action::Singleton | ||
|
|
||
| def queue | ||
| ::Katello::HOST_TASKS_QUEUE |
There was a problem hiding this comment.
Should this run on the default queue? As it stands this will occupy one of the workers on the host tasks queue. But if on the default queue, processing could be delayed.
| def self.trigger_drain | ||
| return if DRAIN_MUTEX.locked? | ||
|
|
||
| DRAIN_MUTEX.synchronize do |
There was a problem hiding this comment.
In my benchmark which calls this method 10000 times, over 6000 of them short-circuited on the lock. That removes contention around draining the queue and frees up request threads faster.
| HostQueueElement.where(host_id: queue.map(&:host_id)).delete_all | ||
| queue | ||
| query = HostQueueElement.order(:id).select(:id).limit(amount) | ||
| result = ActiveRecord::Base.connection.execute(format(DELETE_QUERY, query.to_sql)) |
There was a problem hiding this comment.
I wanted to see if I could reduce this method to a single query. And when I did I decided to include it here :)
It's achieved by making host_id unique in the table which wasn't the case before.
| begin | ||
| ForemanTasks.async_task(Actions::Katello::Applicability::Hosts::BulkGenerate, host_ids: host_ids) | ||
| rescue => e | ||
| Rails.logger.error("Error while draining applicability queue #{e}") |
There was a problem hiding this comment.
When this error is reached it means the generation task failed to spawn but the hosts have already been removed from the applicability queue and are lost. I wonder if they should be placed back on the queue...
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
app/services/katello/applicability/scheduler.rb (1)
35-50: Usetry_lockinstead oflocked?+synchronize.
locked?is only a snapshot. Another thread can acquire the mutex beforesynchronize, so request threads can still block here under load even though this path is meant to exit fast. Usetry_lockwith anensureblock to unlock immediately if acquired.♻️ Minimal change
- return if DRAIN_MUTEX.locked? - - DRAIN_MUTEX.synchronize do + return unless DRAIN_MUTEX.try_lock + begin depth = Katello::ApplicableHostQueue.queue_depth return if depth == 0 return if scheduler_task if depth < Katello::ApplicableHostQueue.batch_size && bulk_generate_tasks.empty? host_ids = Katello::ApplicableHostQueue.pop_hosts ForemanTasks.async_task(Actions::Katello::Applicability::Hosts::BulkGenerate, host_ids: host_ids) else # High applicability activity detected trigger_scheduler_task end + ensure + DRAIN_MUTEX.unlock end🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/services/katello/applicability/scheduler.rb` around lines 35 - 50, The current trigger_drain uses DRAIN_MUTEX.locked? followed by DRAIN_MUTEX.synchronize which can race; replace that pattern by attempting to acquire the mutex with acquired = DRAIN_MUTEX.try_lock and immediately return unless acquired, then run the existing body (read depth, check scheduler_task, branch to ForemanTasks or trigger_scheduler_task) inside a begin...ensure block and call DRAIN_MUTEX.unlock in the ensure to always release the lock; keep the existing checks (Katello::ApplicableHostQueue.queue_depth, batch_size, bulk_generate_tasks.empty?, scheduler_task, trigger_scheduler_task, and ForemanTasks.async_task with host_ids) intact.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@app/services/katello/applicability/scheduler.rb`:
- Around line 23-29: The drain_loop currently pops host_ids via
::Katello::ApplicableHostQueue.pop_hosts then calls ForemanTasks.async_task,
which removes queue rows before the Dynflow task is known to be durable; change
the flow so that you do not permanently remove queue rows until
ForemanTasks.async_task has successfully returned (i.e., create the task first),
and if async_task raises or fails, reinsert/requeue the popped host_ids back
into the queue (or use a remove_after_handoff/remove_confirm method) so hosts
are not lost; apply the same fix to the other code path that also pops before
handoff (the inline notification callback) so both bulk
(drain_loop/ForemanTasks.async_task) and inline paths only delete rows after a
successful task creation or else requeue on failure.
In `@app/services/katello/applicable_host_queue.rb`:
- Around line 14-15: The insert_all call on HostQueueElement should explicitly
pass the unique index using unique_by to match the migration and codebase
conventions; update the HostQueueElement.insert_all(ids.map { |host_id| {
host_id: host_id } }) call to include unique_by: with the unique index name (the
index on katello_host_queue_elements.host_id) so the intent is explicit while
preserving DB-level conflict handling and still firing
ActiveSupport::Notifications.instrument("applicability_push_hosts") when
result.rows.count > 0.
---
Nitpick comments:
In `@app/services/katello/applicability/scheduler.rb`:
- Around line 35-50: The current trigger_drain uses DRAIN_MUTEX.locked? followed
by DRAIN_MUTEX.synchronize which can race; replace that pattern by attempting to
acquire the mutex with acquired = DRAIN_MUTEX.try_lock and immediately return
unless acquired, then run the existing body (read depth, check scheduler_task,
branch to ForemanTasks or trigger_scheduler_task) inside a begin...ensure block
and call DRAIN_MUTEX.unlock in the ensure to always release the lock; keep the
existing checks (Katello::ApplicableHostQueue.queue_depth, batch_size,
bulk_generate_tasks.empty?, scheduler_task, trigger_scheduler_task, and
ForemanTasks.async_task with host_ids) intact.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: e4cf81a2-3ea7-44c2-80f0-9bc16a379b23
📥 Commits
Reviewing files that changed from the base of the PR and between d6c704d and 7804f9d893d8ee16f7d19132ca13bab226c87a6f.
📒 Files selected for processing (12)
app/lib/actions/katello/applicability/host/generate.rbapp/lib/actions/katello/applicability/hosts/bulk_generate.rbapp/lib/actions/katello/applicability/scheduler.rbapp/models/katello/events/generate_host_applicability.rbapp/models/katello/host/content_facet.rbapp/services/katello/applicability/scheduler.rbapp/services/katello/applicable_host_queue.rbdb/migrate/20260413033844_add_constraints_katello_host_queue_element.katello.rblib/katello/engine.rbtest/actions/katello/applicability/repository_regenerate_applicability_test.rbtest/controllers/api/v2/host_errata_controller_test.rbtest/services/katello/applicable_host_queue_test.rb
💤 Files with no reviewable changes (5)
- lib/katello/engine.rb
- test/actions/katello/applicability/repository_regenerate_applicability_test.rb
- app/models/katello/host/content_facet.rb
- app/lib/actions/katello/applicability/host/generate.rb
- app/models/katello/events/generate_host_applicability.rb
| def self.drain_loop | ||
| until (host_ids = ::Katello::ApplicableHostQueue.pop_hosts).empty? | ||
| begin | ||
| ForemanTasks.async_task(Actions::Katello::Applicability::Hosts::BulkGenerate, host_ids: host_ids) | ||
| rescue => e | ||
| Rails.logger.error("Error while draining applicability queue #{e}") | ||
| end |
There was a problem hiding this comment.
Don't delete queue rows before the Dynflow handoff is durable.
Both paths remove host_ids from katello_host_queue_elements before ForemanTasks.async_task is known to have succeeded. If task creation raises, those hosts are already lost: drain_loop just logs, and the inline path can also propagate the exception out of the notification callback. Keep the rows queued until the task exists, or requeue the popped IDs on failure.
Also applies to: 43-45
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@app/services/katello/applicability/scheduler.rb` around lines 23 - 29, The
drain_loop currently pops host_ids via ::Katello::ApplicableHostQueue.pop_hosts
then calls ForemanTasks.async_task, which removes queue rows before the Dynflow
task is known to be durable; change the flow so that you do not permanently
remove queue rows until ForemanTasks.async_task has successfully returned (i.e.,
create the task first), and if async_task raises or fails, reinsert/requeue the
popped host_ids back into the queue (or use a
remove_after_handoff/remove_confirm method) so hosts are not lost; apply the
same fix to the other code path that also pops before handoff (the inline
notification callback) so both bulk (drain_loop/ForemanTasks.async_task) and
inline paths only delete rows after a successful task creation or else requeue
on failure.
adamruzicka
left a comment
adamruzicka
left a comment
There was a problem hiding this comment.
We seem to be hitting the same problem over and over again - we need a global singleton, that would process a queue in a possibly blocking fashion and we don't have a good model for this. All we can sensibly express are things which get started (even repeatedly) but which eventually finish.
We tried various things in the past, but nothing was ideal. Thinking out of the box, I'd have two alternatives:
- a completely standalone process
- having it as a concurrent-ruby agent spawned from within dynflow itself
1 is a bit heavy handed, 2 would need some work in dynflow to be able to tap into the things which are already there from the outside (ie non-dynflow user code)
| return if DRAIN_MUTEX.locked? | ||
|
|
||
| DRAIN_MUTEX.synchronize do |
There was a problem hiding this comment.
The mutex is local to a process. It will prevent multiple threads within a single process to enter the critical section at once, but it will not help all that much in multi-process environments.
How about keeping the mutex to prevent local contention, but switch the query from just delete to select for update followed by a delete to allow multiple processes to process the queue at once? Assuming we don't care about ordering.
There was a problem hiding this comment.
I'd been testing with four processes going full throttle and I didn't see any errors. Is it the case that identical chunks of items could be plucked off the table at once?
Your suggestion is the missing piece here. I took a look at the locking options a few days ago and wasn't sure which one, if any, was the right fit. Will try it out.
There was a problem hiding this comment.
Made this change. It brought the throughput to roughly the same as the current implementation which I'm fine with.
There was a problem hiding this comment.
It brought the throughput to roughly the same as the current implementation
That's not the outcome I was hoping for. Would it change if instead of just .lock you did .lock('FOR UPDATE SKIP LOCKED')?
There was a problem hiding this comment.
Throughput is about the same as just FOR UPDATE, at least with my test
There was a problem hiding this comment.
Perhaps I was wrong on that - I lowered the conditional sleep in the drain loop and saw improvements. Check the table in PR description for the full picture.
| ActiveSupport::Notifications.subscribe("applicability_push_hosts") do | ||
| trigger_drain | ||
| end |
There was a problem hiding this comment.
My mental model for this is that this registers a callback (the body) to be executed when the event in question is emitted. This callback gets executed in the same process and thread that emitted the event. If that's true, does the callback execution block the caller thread?
There was a problem hiding this comment.
That's right - it does block. The mutex is benefiting in that way as well. It's still ~ 30% faster than the current implementation.
|
@adamruzicka interesting reading about the Agent - first time I've come across it. What's the level of effort if a lone open source contributor had some high level notes from you about how they see the implementation going? At the moment I'm eager to move this forward - it's the last big step to remove Katello's event queue & event daemon. Making this fit better with Dynflow would be nice, though. That (or something like it) could come later and this could be adapted at that point. |
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
app/services/katello/applicability/scheduler.rb (1)
24-29:⚠️ Potential issue | 🟠 MajorDon't remove queue entries before task creation succeeds.
Both paths call
ApplicableHostQueue.pop_hostsfirst, which already deletes the rows. IfForemanTasks.async_taskraises, the rescue only logs and those hosts are lost from the applicability queue. Requeue on failure, or split this into reserve/ack steps so the handoff is durable.Also applies to: 43-45
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/services/katello/applicability/scheduler.rb` around lines 24 - 29, The current loop calls ::Katello::ApplicableHostQueue.pop_hosts which removes entries immediately, then calls ForemanTasks.async_task(Actions::Katello::Applicability::Hosts::BulkGenerate...), losing hosts if async_task raises; change the flow so popping is only an acknowledgement after successful task creation: either (A) use a reserve/claim API on ::Katello::ApplicableHostQueue (e.g. reserve_hosts or a new claim method) and call pop/ack only after ForemanTasks.async_task returns successfully, or (B) if no reserve API exists, re-enqueue the host_ids in the rescue block (e.g. push/requeue method on ApplicableHostQueue) when ForemanTasks.async_task raises; apply the same fix for the other occurrence around lines 43-45.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@app/services/katello/applicable_host_queue.rb`:
- Around line 14-15: Revert the narrowed conflict handling by removing the
unique_by: :host_id option from the HostQueueElement.insert_all call so
PostgreSQL will silently skip duplicates across all unique indexes; capture the
insert_all return value (e.g., result = HostQueueElement.insert_all(...)) and
only call ActiveSupport::Notifications.instrument("applicability_push_hosts")
when result.rows.length > 0 to avoid waking the scheduler when nothing was
inserted.
- Around line 21-23: The pop_hosts method currently returns string IDs from
ActiveRecord::Base.connection.execute; update the return to convert those values
to integers (e.g., map the flattened result to integers) so callers like
scheduler.rb and BulkGenerate receive integer host_ids; locate the call that
runs format(DELETE_QUERY, query.to_sql) and replace the final
result.values.flatten with a conversion to integers (refer to HostQueueElement,
DELETE_QUERY and pop_hosts).
---
Duplicate comments:
In `@app/services/katello/applicability/scheduler.rb`:
- Around line 24-29: The current loop calls
::Katello::ApplicableHostQueue.pop_hosts which removes entries immediately, then
calls
ForemanTasks.async_task(Actions::Katello::Applicability::Hosts::BulkGenerate...),
losing hosts if async_task raises; change the flow so popping is only an
acknowledgement after successful task creation: either (A) use a reserve/claim
API on ::Katello::ApplicableHostQueue (e.g. reserve_hosts or a new claim method)
and call pop/ack only after ForemanTasks.async_task returns successfully, or (B)
if no reserve API exists, re-enqueue the host_ids in the rescue block (e.g.
push/requeue method on ApplicableHostQueue) when ForemanTasks.async_task raises;
apply the same fix for the other occurrence around lines 43-45.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: a48b6958-82ce-49b4-8215-e469ce86051f
📥 Commits
Reviewing files that changed from the base of the PR and between 7804f9d893d8ee16f7d19132ca13bab226c87a6f and 5db87c3144a715226fe38344241b546db3748f47.
📒 Files selected for processing (3)
app/services/katello/applicability/scheduler.rbapp/services/katello/applicable_host_queue.rbtest/services/katello/applicability/scheduler_test.rb
| HostQueueElement.insert_all(ids.map { |host_id| { host_id: host_id } }, unique_by: :host_id) | ||
| ActiveSupport::Notifications.instrument("applicability_push_hosts") |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
In Rails ActiveRecord on PostgreSQL, what is the behavioral difference between insert_all(...)andinsert_all(..., unique_by: :host_id) on unique-index conflicts, and does the returned result expose whether any rows were actually inserted?
💡 Result:
In Rails ActiveRecord on PostgreSQL, both insert_all(...) and insert_all(..., unique_by: :host_id) generate INSERT ... ON CONFLICT DO NOTHING statements, skipping rows that conflict with unique indexes. Without unique_by: skips conflicts on any unique index on the table. Conflicts on other unique indexes (not specified by unique_by) raise ActiveRecord::RecordNotUnique. With unique_by: :host_id: skips only conflicts on the unique index matching host_id (by column(s) or index name). Conflicts on other unique indexes raise ActiveRecord::RecordNotUnique. Both return an ActiveRecord::Result object. The result exposes whether rows were actually inserted via result.length or result.rows.length (number of successfully inserted rows, from RETURNING clause, which only returns inserted rows) and result.affected_rows (total rows affected by PostgreSQL INSERT ... ON CONFLICT DO NOTHING, matching inserted rows). If returning: false, rows are empty but affected_rows still indicates inserted count.
Citations:
- 1: https://apidock.com/rails/v7.2.3/ActiveRecord/Relation/insert_all
- 2: https://apidock.com/rails/v8.0.0/ActiveRecord/Relation/insert_all
- 3: https://www.bigbinary.com/blog/bulk-insert-support-in-rails-6
- 4: https://api.rubyonrails.org/v7.0.4/classes/ActiveRecord/Persistence/ClassMethods.html
- 5: https://github.com/rails/rails/blob/main/activerecord/lib/active_record/insert_all.rb
- 6: http://www.postgresql.org/docs/15/sql-insert.html
- 7: https://api.rubyonrails.org/classes/ActiveRecord/Result.html
- 8: https://github.com/rails/rails/blob/fa8f0812160665bff083a089d2bb2fc1817ea03e/activerecord/lib/active_record/result.rb
- 9: Add
affected_rowstoActiveRecord::Resultrails/rails#55060
Revert unique_by: :host_id to keep conflict handling across all unique indexes.
The plain insert_all intentionally causes PostgreSQL to skip duplicate rows across all unique indexes on the table. Adding unique_by: :host_id narrows conflict handling to only that index and can surface ActiveRecord::RecordNotUnique errors from other unique constraints, breaking the intended silent-skip behavior.
Additionally, the unconditional ActiveSupport::Notifications.instrument("applicability_push_hosts") wakes the scheduler even when every row is skipped due to conflicts. Only notify when rows were actually inserted using result.rows.length > 0.
Fix
- HostQueueElement.insert_all(ids.map { |host_id| { host_id: host_id } }, unique_by: :host_id)
- ActiveSupport::Notifications.instrument("applicability_push_hosts")
+ result = HostQueueElement.insert_all(ids.map { |host_id| { host_id: host_id } })
+ ActiveSupport::Notifications.instrument("applicability_push_hosts") if result.rows.length > 0🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@app/services/katello/applicable_host_queue.rb` around lines 14 - 15, Revert
the narrowed conflict handling by removing the unique_by: :host_id option from
the HostQueueElement.insert_all call so PostgreSQL will silently skip duplicates
across all unique indexes; capture the insert_all return value (e.g., result =
HostQueueElement.insert_all(...)) and only call
ActiveSupport::Notifications.instrument("applicability_push_hosts") when
result.rows.length > 0 to avoid waking the scheduler when nothing was inserted.
jturel
force-pushed
the
applicability_scheduler_ad_hoc
branch
from
7bae342 to
f60769d
Compare
I'd say moderate? Most, if not all, the building blocks should already be there in some shape or form so it should be just a matter of assembling them in a new way and exposing that to the world. |
If errors happen, draining will be re-attempted
|
The more I look at this PR and #11708 the less satisfied I am with them as solutions to the problem. My curiosity has been satisfied by writing them up but I'm not sure I want to pursue either despite my zeal to remove the event queue and daemon from the codebase. It all comes back to Adam's comment above regarding the fact there's no native support for something like this in the system. Everything is a workaround. I may take a look at the concurrent ruby agent + dynflow option in the near future. Dropping this back down to a draft for the time being. |
What do you envision the agent gaining us that the current implementation is missing? Is it mostly a reduction in complexity of the changes? Or is there more functionality to be gained? I agree that this applicability logic scenario lends itself better to a long-running agent logically. Trying to weigh the benefits introduced here and being one step closer to getting rid of the event daemon versus having a more perfect solution. Although perhaps we aren't in a huge rush to get rid of the event daemon. |
|
@ianballou thanks for chiming in. I've been looking at the Actor (agent was a typo) option: Dynflow/dynflow#468 I see these advantages:
This PR served its purpose by further illustrating the gaps highlighted by Adam. Branching is coming up soon. A new solution could come to life in the following release which seems like the soonest the event daemon could be removed anyway, if that was a (hypothetical) priority. |
|
This all makes a bit more sense looking into Actor, nice. The changes in jturel#1 make sense, looking forward to diving in more. |
|
Closing this as I hope to use a Dynflow mechanism to solve this in the (near?) future |
3 participants
What are the changes introduced in this pull request?
Introduces a new mechanism for processing the host applicability queue that operates without the
Katello Event QueueHow it works:
Scheduler Serviceis notifiedScheduler Actiontask is started to drain the queue fullyBulkGeneratetask is started which sufficiently drains the queuePros
Cons
EventQueue.push_eventConsiderations taken when implementing this change?
What are the testing steps for this pull request?
I encourage you to run the benchmark command shared below to get a feel for the execution.
There are several real-world scenarios that can be used to test:
Reminder: applicability calculation itself didn't change, just the mechanism around calling it.
Performance
Test: Push 25,000 host ids onto the applicabilty queue, multi-threaded. This is not a test of time to calculate applicability but rather how long it takes to handle each applicability request notification. db pool @ 30 connections.
The new solution performs better as the batch size increases. It also creates BulkGenerate tasks more efficiently in terms of the ratio between batch size & request count. Keep in mind the throughput numbers seem to vary +/- 20% in my environment which I attribute to being a heavy load on my VM.
EventQueue+is the existing implementation with the sleep in the drain loop + the BulkGenerate querying changes here for a more apples-to-apples comparison.TODOs
Summary by CodeRabbit
New Features
Refactor
Removed
Bug Fixes
Chores
Tests