Tuning Latensi Worker Queue: Mengatasi Bufferbloat dan Mengatur Pacing Task

Anatomi Latensi: Memisahkan Queuing Delay dan Processing Time

Lonjakan latensi pada sistem pemrosesan asinkron sering disalahartikan sebagai penurunan performa eksekusi task. Pada kenyataannya, total waktu siklus sebuah task (Total Turnaround Time) terdiri dari dua komponen yang berbeda secara fundamental:

Total Turnaround Time = Queuing Delay (Wait Time) + Processing Time (Service Time)

Processing Time mengukur durasi CPU dan I/O yang dihabiskan worker untuk menyelesaikan satu unit kerja. Sedangkan Queuing Delay adalah durasi sejak task dikirim oleh produser hingga worker pertama kali mengeksekusi instruksi pertama task tersebut. Jika tail latency p99 membengkak namun processing time stabil pada p50 maupun p99, masalah berada pada antrean, bukan pada logika bisnis kode Anda.

Untuk mengisolasi metrik ini, sematkan metadata timestamp resolusi mikrodetik saat task di-dispatch oleh producer. Worker bertanggung jawab menghitung selisih waktu segera setelah menerima payload.

package main

import (
	"context"
	"encoding/json"
	"time"

	"github.com/prometheus/client_golang/prometheus"
)

var (
	queueWaitDuration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "worker_queue_wait_duration_seconds",
			Help:    "Waktu tunggu task dalam antrean sebelum dieksekusi.",
			Buckets: prometheus.DefBuckets,
		},
		[]string{"queue_name"},
	)
	queueProcessDuration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "worker_queue_process_duration_seconds",
			Help:    "Durasi eksekusi aktual task oleh worker.",
			Buckets: prometheus.DefBuckets,
		},
		[]string{"queue_name"},
	)
)

type TaskPayload struct {
	ID          string `json:"id"`
	EnqueuedAt  int64  `json:"enqueued_at_unix_nano"`
	PayloadData string `json:"data"`
}

func ProcessTask(ctx context.Context, raw []byte) error {
	startReceive := time.Now()

	var task TaskPayload
	if err := json.Unmarshal(raw, &task); err != nil {
		return err
	}

	// 1. Catat Queuing Delay
	enqueuedAt := time.Unix(0, task.EnqueuedAt)
	waitTime := startReceive.Sub(enqueuedAt).Seconds()
	queueWaitDuration.WithLabelValues("email_delivery").Observe(waitTime)

	// 2. Eksekusi task & catat Processing Time
	startExec := time.Now()
	err := executeBusinessLogic(ctx, task)
	execDuration := time.Since(startExec).Seconds()
	queueProcessDuration.WithLabelValues("email_delivery").Observe(execDuration)

	return err
}

func executeBusinessLogic(ctx context.Context, t TaskPayload) error {
	// Simulasi pekerjaan I/O bound
	time.Sleep(15 * time.Millisecond)
	return nil
}

Penyebab Bufferbloat pada Worker Queue

Bufferbloat terjadi saat buffer jaringan atau antrean memori internal menampung paket melebihi kapasitas throughput processing yang realistis. Pada worker pipeline, bufferbloat dipicu oleh mekanisme Prefetch broker yang terlalu agresif (misalnya RabbitMQ Basic.QoS atau prefetch internal Redis stream consumer).

Ketika client worker mengonfigurasi nilai prefetch tinggi (misalnya 100 task per worker concurrency), broker akan mendorong 100 task tersebut secara berurutan ke soket TCP worker. Task ini kemudian ditahan di dalam user-space memory buffer worker.

Dampak Negatif Head-of-Line Blocking

• Head-of-Line (HoL) Blocking: Jika task pertama membutuhkan waktu eksekusi 5 detik akibat lock database, 99 task ringan di belakangnya terperangkap di memori lokal worker tersebut, meskipun worker lain di node terpisah dalam kondisi idle.
• Jitter Penjadwalan Linux: Buffer memori yang membengkak meningkatkan footprint RSS proses worker, memicu cache eviction CPU L2/L3, dan memaksa Linux kernel scheduler (CFS) melakukan context-switching lebih mahal.
• Pembengkakan p99 Tail Latency: Task yang sebenarnya hanya membutuhkan 10ms eksekusi mengalami degradasi waktu tunggu hingga puluhan detik akibat duduk di buffer lokal worker yang keliru.

Tuning Prefetch dan Concurrency Limit

Aturan dasar tuning prefetch: sesuaikan prefetch dengan jenis task. Untuk task dengan durasi eksekusi bervariasi (non-deterministik) atau I/O bound panjang, prefetch count harus disetel ke 1.

Untuk task pendek dan deterministik (seperti agregasi metrik mikro), prefetch dapat dihitung melalui rumus Bandwidth-Delay Product antrean:

Optimal Prefetch = CEIL((Network RTT + Mean Processing Time) / Mean Processing Time) * Concurrency

Contoh konfigurasi QoS pada AMQP (Go):

// Pastikan worker concurrency diatur seimbang dengan core CPU / I/O pool
const workerConcurrency = 4

// prefetchCount: Batasi jumlah task yang berada di unacknowledged state
// prefetchSize: 0 (unlimited byte size)
// global: false (pengaturan diaplikasikan per-channel, bukan per-koneksi)
err := ch.Qos(
	workerConcurrency * 1, // Prefetch = 4 (1 per worker goroutine)
	0,
	false,
)
if err != nil {
	panic("Gagal menyetel QoS Prefetch")
}

Implementasi Adaptive Batch Pacing dan Backpressure

Sistem sering kali kewalahan bukan karena kekurangan kapasitas agregat, melainkan karena producer mengirim beban dalam bentuk spike (burst). Menjalankan consumer tanpa batasan laju akan memindahkan kemacetan ke downstream database.

Gunakan adaptive pacing berbasis token bucket yang memantau ambang batas p99 queuing delay. Jika queuing delay naik di atas Service Level Objective (SLO), worker menurunkan penerimaan batch secara sementara untuk memberikan waktu downstream pulih.

package main

import (
	"context"
	"sync/atomic"
	"time"
)

type AdaptiveConsumer struct {
	currentRateLimit int64 // Task per detik
	latencySloMicros int64 // Contoh: 100_000 (100ms)
}

func (ac *AdaptiveConsumer) AdjustPacing(lastWaitDuration time.Duration) {
	current := atomic.LoadInt64(&ac.currentRateLimit)
	if lastWaitDuration.Microseconds() > ac.latencySloMicros {
		// Queue wait time melebihi batas SLO: Lakukan backpressure step-down (penurunan laju 10%)
		newLimit := int64(float64(current) * 0.9)
		if newLimit >= 1 {
			atomic.StoreInt64(&ac.currentRateLimit, newLimit)
		}
	} else {
		// Stabil di bawah ambang batas: Tambahkan kapasitas secara linier (Additive Increase)
		atomic.StoreInt64(&ac.currentRateLimit, current+1)
	}
}

func (ac *AdaptiveConsumer) ConsumeWithBackpressure(ctx context.Context, taskChan <-chan []byte) {
	for {
		select {
		case <-ctx.Done():
			return
		case rawTask := <-taskChan:
			start := time.Now()
			
			_ = ProcessTask(ctx, rawTask)

			ac.AdjustPacing(time.Since(start))
		}
	}
}

Observabilitas Latensi di Linux: Scheduling vs Worker Delay

Saat latensi pipeline melonjak, Anda harus memastikan apakah penundaan terjadi di dalam aplikasi atau akibat CPU throttling dari Linux scheduler. Jika thread worker Anda diletakkan pada target cgroup yang dibatasi kuota CPU (CPU limits di Kubernetes/systemd), kernel akan menunda giliran eksekusi thread worker.

Gunakan perkakas Linux eBPF dari paket bcc-tools, yaitu runqlat, untuk melihat distribusi waktu yang dihabiskan task pada runqueue scheduler Linux sebelum dieksekusi CPU core:

# Pantau latensi runqueue CPU secara interval 5 detik
sudo /usr/share/bcc/tools/runqlat 5 1

Output normal:
     usecs               : count     distribution
         0 -> 1          : 18239    |****************************************|
         2 -> 3          : 4120     |*********                               |
         4 -> 7          : 512      |*                                       |

Output CPU Throttling / Saturation:
     usecs               : count     distribution
      1024 -> 2047       : 421      |****                                    |
      2048 -> 4095       : 1892     |****************                        |
      4096 -> 8191       : 4310     |****************************************|

Jika histogram bergeser ke atas 1 milidetik (>1000 µs), p99 tinggi pada worker Anda dipicu oleh CPU contention di level OS, bukan oleh buffer queue broker. Solusinya adalah mematikan CPU hard limit atau meningkatkan cpu.shares.

Verifikasi Pengujian Beban (Load Testing)

Untuk memvalidasi efektivitas eliminasi bufferbloat, uji worker dengan skenario pembebanan burst asimetris: 10.000 task dikirimkan secara serentak, dengan 5% di antaranya berupa slow task (pemrosesan 200ms) dan 95% sisanya berupa fast task (pemrosesan 5ms).

Catatan Trade-off: Mengurangi prefetch ke angka 1 sedikit menurunkan throughput absolut (~5-8%) karena adanya tambahan RTT TCP ack antar-broker dan worker. Namun, pengorbanan throughput kecil ini sebanding dengan jaminan hilangnya spike tail latency p99 pada pipeline produksi.