Merge pull request #28 from kpumuk/demo-perf

Optimize the demo stand

Author: kpumuk

README.md

@@ -111,24 +111,30 @@ This will ensure we do not push any dependencies with known vulnerabilities, and
 There is a test environment prepared in the `demo/` directory. Simply start it with:
 
 ```bash
-docker-compose up --build
+mise run demo
+mise run demo-sidekiq73
+mise run demo-sidekiq80
+mise run demo-sidekiq81
 ```
 
 This will:
 
 - Start a Redis server
-- Start Sidekiq servers for different Sidekiq versions with some demo jobs (with Web interface)
+- Start one Sidekiq demo stack for the selected version, with demo jobs and the Web interface
 
-| Sidekiq   | Redis URL                  | Web Dashboard         |
-| --------- | -------------------------- | --------------------- |
-| **7.3.x** | `redis://localhost:6379/0` | http://localhost:9292 |
-| **8.0.0** | `redis://localhost:6379/1` | http://localhost:9293 |
-| **8.1.0** | `redis://localhost:6379/2` | http://localhost:9294 |
+| Command                   | Sidekiq   | Redis URL                  | Web Dashboard         |
+| ------------------------- | --------- | -------------------------- | --------------------- |
+| `mise run demo`           | **8.1.0** | `redis://localhost:6379/0` | http://localhost:9292 |
+| `mise run demo-sidekiq73` | **7.3.x** | `redis://localhost:6379/2` | http://localhost:9294 |
+| `mise run demo-sidekiq80` | **8.0.0** | `redis://localhost:6379/1` | http://localhost:9293 |
+| `mise run demo-sidekiq81` | **8.1.0** | `redis://localhost:6379/0` | http://localhost:9292 |
+
+The version-specific tasks can run in parallel because each demo stand uses its own Redis database and dashboard port.
 
 You can connect to a specific Sidekiq version using:
 
 ```bash
-go run ./cmd/lazykiq --redis redis://localhost:6379/2
+go run ./cmd/lazykiq --redis redis://localhost:6379/1
 ```
 
 ### Website

demo/Dockerfile

@@ -1,4 +1,4 @@
-FROM ruby:3.4-alpine
+FROM ruby:4.0.2-alpine@sha256:e973e9aa7573233432565798734a891d648b8f9f6ae20d57db7338b63c35db02
 
 ARG BUNDLE_GEMFILE=Gemfile
 

demo/README.md

@@ -2,13 +2,23 @@
 
 A demo environment for testing Sidekiq TUI tools. Generates continuous job traffic across multiple queues with realistic failure scenarios.
 
+The current profile is intentionally lightweight: older Sidekiq versions stay compact, while the 8.1 demo uses five thin worker processes to stretch the process-oriented UI without bringing back the original CPU load.
+
 ## Quick Start
 
 ```bash
-docker-compose up --build
+mise run demo
+mise run demo-sidekiq73
+mise run demo-sidekiq80
+mise run demo-sidekiq81
 ```
 
-Web UI: http://localhost:9292
+`mise run demo` starts the latest Sidekiq demo (8.1) on `http://localhost:9292` with `redis://localhost:6379/0`.
+`mise run demo-sidekiq73` starts Sidekiq 7.3 on `http://localhost:9294` with `redis://localhost:6379/2`.
+`mise run demo-sidekiq80` starts Sidekiq 8.0 on `http://localhost:9293` with `redis://localhost:6379/1`.
+`mise run demo-sidekiq81` starts Sidekiq 8.1 on `http://localhost:9292` with `redis://localhost:6379/0`.
+
+The three tasks can run in parallel because each demo stand uses its own Redis database and dashboard port.
 
 ## Structure
 
@@ -29,15 +39,15 @@ Web UI: http://localhost:9292
 
 **5 Queues** (by priority): critical, default, mailers, batch, low
 
-**Limits**: 10k jobs/queue, 20k retry queue max
+**Limits**: 1.5k jobs/queue, 1k retry queue max, 500 scheduled max
 
 **Extra cases**:
 - ActiveJob-wrapped jobs with GlobalID-serialized arguments
 - ActionMailer-wrapped jobs with arguments (including GlobalID)
 - Tagged jobs sharing the same 5 tags (intersection)
 - **DataSyncJob** uses `Sidekiq::IterableJob` to process entity IDs one at a time:
   - Iterates over 1-10 entity IDs per job
-  - Random sleep duration per iteration (0.2-5s)
+  - Random sleep duration per iteration (0.2-1.0s)
   - 8% failure rate per iteration for realistic retry behavior
   - Demonstrates job interruption and resumption on worker restart
 

demo/config/sidekiq.yml

@@ -1,5 +1,5 @@
 ---
-:concurrency: 8
+:concurrency: 4
 :queues:
   - [critical, 10]
   - [default, 5]

demo/docker-compose.yml

@@ -1,7 +1,29 @@
+x-sidekiq-service: &sidekiq-service
+  build:
+    context: .
+  depends_on:
+    redis:
+      condition: service_healthy
+  volumes:
+    - .:/app
+
 services:
   redis:
     image: redis:7-alpine
-    command: ["redis-server", "--save", ""]
+    command:
+      [
+        "redis-server",
+        "--save",
+        "",
+        "--hz",
+        "1",
+        "--dynamic-hz",
+        "no",
+        "--latency-tracking",
+        "no",
+        "--slowlog-log-slower-than",
+        "-1"
+      ]
     ports:
       - "6379:6379"
     healthcheck:
@@ -10,47 +32,38 @@ services:
       timeout: 3s
       retries: 5
 
-  sidekiq-7.3:
+  sidekiq-8.1:
+    <<: *sidekiq-service
     build:
       context: .
       args:
-        BUNDLE_GEMFILE: Gemfile.sidekiq-7.3
-    depends_on:
-      redis:
-        condition: service_healthy
+        BUNDLE_GEMFILE: Gemfile.sidekiq-8.1
     environment:
+      LAZYKIQ_DEMO_LAYOUT: stretched
       REDIS_URL: redis://redis:6379/0
     ports:
       - "9292:9292"
-    volumes:
-      - .:/app
 
   sidekiq-8.0:
+    <<: *sidekiq-service
     build:
       context: .
       args:
         BUNDLE_GEMFILE: Gemfile.sidekiq-8.0
-    depends_on:
-      redis:
-        condition: service_healthy
     environment:
+      LAZYKIQ_DEMO_LAYOUT: compact
       REDIS_URL: redis://redis:6379/1
     ports:
       - "9293:9292"
-    volumes:
-      - .:/app
 
-  sidekiq-8.1:
+  sidekiq-7.3:
+    <<: *sidekiq-service
     build:
       context: .
       args:
-        BUNDLE_GEMFILE: Gemfile.sidekiq-8.1
-    depends_on:
-      redis:
-        condition: service_healthy
+        BUNDLE_GEMFILE: Gemfile.sidekiq-7.3
     environment:
+      LAZYKIQ_DEMO_LAYOUT: compact
       REDIS_URL: redis://redis:6379/2
     ports:
       - "9294:9292"
-    volumes:
-      - .:/app

demo/lib/jobs/data_sync_job.rb

@@ -26,7 +26,7 @@ def each_iteration(entity_id, source, destination, entity_ids)
       raise error, "Sync failed for entity #{entity_id} from #{source} to #{destination}"
     end
 
-    # Log progress (visible in Sidekiq logs)
-    logger.info "Synced entity #{entity_id} from #{source} to #{destination} (#{duration.round(2)}s)"
+    # Keep logs quiet; the dashboard already shows the live job mix.
+    logger.debug { "Synced entity #{entity_id} from #{source} to #{destination} (#{duration.round(2)}s)" }
   end
 end

demo/scheduler.rb

@@ -6,12 +6,14 @@
 
 class JobScheduler
   QUEUES = %w[critical default mailers batch low unsafe unprocessed_1 unprocessed_2 unprocessed_3 unprocessed_4 unprocessed_5].freeze
-  MAX_JOBS_PER_QUEUE = 10_000
-  MAX_UNSAFE_JOBS = 42
-  MAX_UNPROCESSED_JOBS = 5
-  MAX_RETRY_QUEUE = 20_000
-  MAX_SCHEDULED_JOBS = 5_000
-  SCHEDULE_BATCH_SIZE = 100
+  MAX_JOBS_PER_QUEUE = 1_500
+  MAX_UNSAFE_JOBS = 12
+  MAX_UNPROCESSED_JOBS = 2
+  MAX_RETRY_QUEUE = 1_000
+  MAX_SCHEDULED_JOBS = 500
+  SCHEDULE_BATCH_SIZE = 12
+  SCHEDULER_INTERVAL = 2.0
+  SCHEDULED_REFRESH_EVERY = 4
   ACTIVEJOB_WRAPPER = "Sidekiq::ActiveJob::Wrapper"
   ACTION_MAILER_DELIVERY = "ActionMailer::MailDeliveryJob"
   COMMON_TAGS = %w[tag-alpha tag-bravo tag-charlie tag-delta tag-echo].freeze
@@ -152,6 +154,9 @@ def self.tagged_payload(job_class, queue, args, tags)
   def initialize
     @weighted_jobs = build_weighted_job_list
     @running = false
+    @loop_count = 0
+    @scheduled_sizes = Hash.new(0)
+    @scheduled_total = 0
   end
 
   def start
@@ -163,9 +168,11 @@ def start
     puts "Queues: #{QUEUES.join(", ")}"
 
     while @running
+      refresh_scheduled_sizes if refresh_scheduled_sizes?
       maintain_queues
       maintain_scheduled
-      sleep 0.5
+      @loop_count += 1
+      sleep SCHEDULER_INTERVAL
     end
   end
 
@@ -189,7 +196,7 @@ def maintain_queues
     end
 
     queue_sizes = fetch_queue_sizes
-    scheduled_sizes = fetch_scheduled_sizes
+    scheduled_sizes = @scheduled_sizes
 
     QUEUES.each do |queue_name|
       current_size = queue_sizes[queue_name] || 0
@@ -223,12 +230,11 @@ def schedule_jobs_for_queue(queue_name, count)
   end
 
   def maintain_scheduled
-    scheduled_size = Sidekiq::ScheduledSet.new.size
     queue_sizes = fetch_queue_sizes
-    scheduled_sizes = fetch_scheduled_sizes
+    scheduled_sizes = @scheduled_sizes
 
-    if scheduled_size < MAX_SCHEDULED_JOBS * 0.8
-      jobs_to_add = [MAX_SCHEDULED_JOBS - scheduled_size, SCHEDULE_BATCH_SIZE].min
+    if @scheduled_total < MAX_SCHEDULED_JOBS * 0.8
+      jobs_to_add = [MAX_SCHEDULED_JOBS - @scheduled_total, SCHEDULE_BATCH_SIZE].min
       add_scheduled_jobs(jobs_to_add, queue_sizes, scheduled_sizes) if jobs_to_add > 0
     end
   end
@@ -249,6 +255,7 @@ def add_scheduled_jobs(count, queue_sizes, scheduled_sizes)
       delay = rand(1..86400) # Schedule between 1 second and 24 hours
       enqueue_job(job_def, delay: delay)
       scheduled_sizes[queue_name] = scheduled_size + 1
+      @scheduled_total += 1
       added += 1
     end
 
@@ -257,11 +264,24 @@ def add_scheduled_jobs(count, queue_sizes, scheduled_sizes)
 
   def fetch_scheduled_sizes
     sizes = Hash.new(0)
+    total = 0
     Sidekiq::ScheduledSet.new.each do |job|
       queue_name = job.item["queue"]
-      sizes[queue_name] += 1 if queue_name
+      next unless queue_name
+
+      sizes[queue_name] += 1
+      total += 1
     end
-    sizes
+    @scheduled_sizes = sizes
+    @scheduled_total = total
+  end
+
+  def refresh_scheduled_sizes?
+    @scheduled_sizes.empty? || (@loop_count % SCHEDULED_REFRESH_EVERY).zero?
+  end
+
+  def refresh_scheduled_sizes
+    fetch_scheduled_sizes
   end
 
   def enqueue_job(job_def, delay: nil)

demo/start.rb

@@ -4,6 +4,24 @@
 require_relative "boot"
 require_relative "scheduler"
 
+def worker_profiles_for(layout)
+  case layout
+  when "stretched"
+    [
+      {queues: ["default,5"], concurrency: 2},
+      {queues: ["low,1"], concurrency: 1},
+      {queues: ["critical,10"], concurrency: 2},
+      {queues: ["mailers,5"], concurrency: 1},
+      {queues: ["batch,3"], concurrency: 1, unsafe_capsule: true}
+    ]
+  else
+    [
+      {queues: ["default,5", "low,1"], concurrency: 4},
+      {queues: ["critical,10", "mailers,5", "batch,3"], concurrency: 4, unsafe_capsule: true}
+    ]
+  end
+end
+
 # Determine mode
 mode = ARGV[0] || "all"
 
@@ -34,23 +52,16 @@
 
   pids = []
 
-  # Fork worker processes with 8 workers each, different queues per process
-  worker_queues = [
-    # Process 1: default and low queues (weighted)
-    ["default,5", "low,1"],
-    # Process 2: default and low queues (weighted)
-    ["default,5", "low,1"],
-    # Process 3: critical and mailers queues (weighted)
-    ["critical,10", "mailers,5"],
-    # Process 4: batch queue only (weighted) with unsafe capsule
-    ["batch,3"]
-  ]
-
-  worker_queues.each_with_index do |queues, i|
+  # Stretch the latest demo with more Sidekiq processes while keeping older
+  # versions compact and cheap.
+  worker_profiles = worker_profiles_for(ENV.fetch("LAZYKIQ_DEMO_LAYOUT", "compact"))
+
+  worker_profiles.each do |profile|
     pids << fork do
       cmd = ["bundle", "exec", "sidekiq", "-r", "./boot.rb", "-C", "config/sidekiq.yml"] +
-        queues.flat_map { |q| ["-q", q] }
-      if i == 3
+        profile[:queues].flat_map { |q| ["-q", q] } +
+        ["-c", profile[:concurrency].to_s]
+      if profile[:unsafe_capsule]
         exec({"LAZYKIQ_UNSAFE_CAPSULE" => "1"}, *cmd)
       else
         exec(*cmd)

internal/sidekiq/parse.go

@@ -38,6 +38,16 @@ func parseOptionalInt64(field any) (int64, bool) {
 	return 0, false
 }
 
+// parseOptionalInt parses various types to int with success indication.
+// Values outside the platform int range are rejected.
+func parseOptionalInt(field any) (int, bool) {
+	parsed, ok := parseOptionalInt64(field)
+	if !ok || parsed < math.MinInt || parsed > math.MaxInt {
+		return 0, false
+	}
+	return int(parsed), true
+}
+
 // parseOptionalFloat64 parses various types to float64 with success indication.
 func parseOptionalFloat64(field any) (float64, bool) {
 	switch value := field.(type) {

internal/sidekiq/process.go

@@ -300,8 +300,8 @@ func (p *Process) refreshFromFields(fields []any) {
 		}
 	}
 
-	if busyCount, ok := parseOptionalInt64(fieldAt(fields, 1)); ok {
-		p.Busy = int(busyCount)
+	if busyCount, ok := parseOptionalInt(fieldAt(fields, 1)); ok {
+		p.Busy = busyCount
 	}
 
 	if beat, ok := parseOptionalFloat64(fieldAt(fields, 2)); ok {