Pemrosesan berkas citra RAW (seperti DNG, CR3, atau ARW) dalam sistem antrean terdistribusi memiliki karakteristik resource-heavy. Berbeda dengan kompresi JPEG standar, pipeline RAW memerlukan demosaicing, koreksi profil lensa, color matrix transforms, dan floating-point raster arrays (32-bit per channel). Jika worker pool mengeksekusi task hanya berdasarkan ketersediaan core CPU tanpa memperhitungkan alokasi byte buffer, lonjakan antrean (concurrency spike) akan memicu Out-of-Memory (OOM) killer pada level OS.

Artikel ini membahas mitigasi OOM melalui pola memory capping, implementasi distributed lock untuk menjamin idempotensi, serta pemanfaatan caching pada stage pipeline antara (intermediate pipes).

1. Karakteristik Beban Kerja RAW Pipeline

Terinspirasi dari alur pemrosesan non-destruktif Ansel atau Darktable, pipeline rendering RAW umumnya dibagi menjadi modul berurutan:

  1. Raw Decoding & Demosaicing: Mengubah array Bayer (1 channel per pixel) menjadi RGB float. Sensor 45MP menghasilkan buffer raster sekitar 45,000,000 * 3 channel * 4 byte (float32) = ~540 MB hanya untuk satu state image di memori.
  2. Input Color Profile & Lens Correction: Transformasi geometri dan matriks warna yang menduplikasi buffer target untuk menghindari race condition jika berjalan multithreaded.
  3. Exposure, Tonemapping, & Denoise: Komputasi filtering berat yang sering membutuhkan buffer intermediate tambahan.
  4. Output Color Profile & Export: Kuantisasi dari float32 ke 8-bit/16-bit integer (misal sRGB JPEG/TIFF).

Jika satu worker thread memproses satu gambar 45MP, konsumsi memori puncak (peak memory) per task bisa mencapai 1.5 GB hingga 2.5 GB. Jika runtime menjalankan 8 worker paralel pada mesin dengan RAM 8 GB, server dipastikan mengalami crash akibat swap exhaustion atau OOM kill.

2. Pola Memory Capping dengan Weighted Semaphore

Pendekatan standar antrean menggunakan ukuran fixed pool (misalnya num_workers = runtime.NumCPU()). Ini fatal untuk RAW processing karena ukuran file tidak berbanding lurus dengan resolusi raster setelah didekode.

Solusinya adalah Memory-Budgeted Worker Pool menggunakan Weighted Semaphore. Setiap task membaca metadata resolusi gambar terlebih dahulu (dari EXIF header tanpa mendekompresi payload raw), menghitung estimasi alokasi RAM puncak, lalu meminta unit tiket memori sebelum proses dimulai.

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"golang.org/x/sync/semaphore"
)

type RAWTask struct {
	ImageID   string
	Width     int
	Height    int
	InputPath string
}

type RAWProcessorPool struct {
	memSem     *semaphore.Weighted
	totalBytes int64
}

func NewRAWProcessorPool(maxMemoryBytes int64) *RAWProcessorPool {
	return &RAWProcessorPool{
		memSem:     semaphore.NewWeighted(maxMemoryBytes),
		totalBytes: maxMemoryBytes,
	}
}

// Estimasi alokasi: Base raster (float32 * 3) + buffer intermediate (faktor 2.5x)
func (p *RAWProcessorPool) EstimateRequiredBytes(width, height int) int64 {
	baseRaster := int64(width) * int64(height) * 3 * 4 // float32 = 4 bytes
	overheadMultiplier := 2.5
	return int64(float64(baseRaster) * overheadMultiplier)
}

func (p *RAWProcessorPool) Process(ctx context.Context, task RAWTask) error {
	estimatedMemory := p.EstimateRequiredBytes(task.Width, task.Height)

	if estimatedMemory > p.totalBytes {
		return fmt.Errorf("task %s requires %d bytes, exceeds pool limit %d", task.ImageID, estimatedMemory, p.totalBytes)
	}

	// Blokir eksekusi sampai kapasitas memori mencukupi
	if err := p.memSem.Acquire(ctx, estimatedMemory); err != nil {
		return err
	}
	defer p.memSem.Release(estimatedMemory)

	return p.executePipeline(ctx, task)
}

func (p *RAWProcessorPool) executePipeline(ctx context.Context, task RAWTask) error {
	// Simulasi pemrosesan raw pipeline
	select {
	case <-time.After(500 * time.Millisecond):
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

Catatan: Alokasi memori runtime Go tetap tunduk pada garbage collection (GC). Gunakan debug.SetMemoryLimit pada Go 1.19+ untuk memaksa GC bekerja lebih agresif sebelum mendekati batas cgroup container.

3. Distributed Locking & Idempotensi Retry

Kegagalan task (karena timeout IO atau SIGKILL) menyebabkan broker antrean (RabbitMQ/Redis Streams) mengirim ulang (redelivery) pesan. Tanpa locking, dua worker berbeda dapat memproses gambar yang sama secara bersamaan, menduplikasi beban RAM.

Pola Lock Redis (Redlock/Single-Instance Token)

Gunakan Redis lock dengan TTL dan token acak unik per eksekusi:

SET raw:lock:image:{image_id} {worker_uuid} NX PX 60000
  • NX: Hanya set jika key belum ada. Jika gagal mendapatkan lock, kembalikan task ke antrean dengan exponential backoff.
  • Heartbeat / Lock Renewal: Background goroutine harus memperpanjang TTL jika pipeline modul (misal lens correction + noise reduction) memerlukan waktu lebih lama dari perkiraan.
  • Idempotensi Hasil: Simpan checksum state input. Jika hash target storage (misalnya MinIO/S3 path) sudah berisi artefak final dengan metadata yang cocok, lewati (no-op) dan lakukan ACK pada antrean.

4. Intermediate Pipe Caching

Dalam software seperti Darktable, pipeline tidak selalu diulang dari awal jika pengguna hanya mengubah parameter tone curve. Dalam pemrosesan batch background, caching intermediate stage berguna saat satu RAW diekspor ke berbagai varian (misal: thumbnail WebP, TIFF 16-bit untuk print, dan JPEG full-size).

  1. Stage 1 (Decode & Demosaic): Simpan output linear float buffer ke file sementara (atau memory-mapped file seperti /dev/shm) jika varian output berikutnya dieksekusi di node worker yang sama.
  2. Cache Key: Bentuk hash dari kombinasi Hash(ImageID + RawPayloadChecksum + BasePipelineConfig).
  3. Eviction: Simpan file cache per task ID dengan sistem FIFO atau auto-delete via cleanup hook saat seluruh varian selesai digenerate, untuk mencegah diska penuh (disk exhaustion).

5. Mitigasi dan Tips Debugging

  • Cgroup OOM Monitoring: Monitor memory.oom_control di Linux. Jika worker tiba-tiba menghilang dengan exit code 137, cgroup container mematikan worker karena konsumsi melebih batas hard limit, menandakan faktor overheadMultiplier pada kalkulasi memory capping perlu dinaikkan.
  • Hindari ReadFile Utuh: Jangan membaca payload RAW langsung ke memory slice (os.ReadFile). Gunakan memory-mapped file (mmap) untuk membaca header dan raw strips secara parsial bila parser mendukung.
  • Offload Cgo Memory: Jika menggunakan wrapper LibRaw via Cgo, alokasi memori berada di luar jangkauan Go GC (arena heap C). Wajib memastikan destruktor C (seperti libraw_close) dipanggil melalui defer untuk mencegah kebocoran memori native.