Examples

Compatibility Facade Notice

lockfreequeues is maintained strictly as a zero-overhead backwards-compatibility facade around canonical lockfree. For modern examples using CSP channels and modern queue primitives, see lockfree Documentation.

Practical examples demonstrating lock-free queue patterns.

Running Examples

nimble examples

Or run individual examples:

nim c --threads:on -r examples/audio_buffer.nim
nim c --threads:on -r examples/task_fanout.nim
nim c --threads:on -r examples/event_collector.nim
nim c --threads:on -r examples/job_scheduler.nim

---

Audio Buffer (Bounded SPSC)

Real-time audio processing with fixed latency. The producer captures audio samples, the consumer plays them.

Key properties:

• Fixed latency determined by buffer size
• No allocation during operation
• Wait-free operations for predictable timing

Use cases: Audio engines, DSP pipelines, real-time signal processing

## Audio Buffer Example
##
## Demonstrates using a bounded Sipsic (SPSC) queue for real-time audio processing.
## The producer thread captures audio samples, the consumer thread plays them.
##
## Key properties:
## - Fixed latency: buffer size determines latency (64 samples @ 44.1kHz = 1.45ms)
## - No allocation: ring buffer operations never allocate
## - Wait-free: both push and pop complete in bounded time
##
## This pattern is essential for audio applications where:
## - Latency must be predictable and minimal
## - Glitches from GC pauses or allocation are unacceptable
## - Sample rate is fixed and known

import lockfreequeues/atomic_dsl
import math
import os
import options

import lockfreequeues

const
  SampleRate = 44100
  BufferSize = 64 # ~1.45ms latency at 44.1kHz
  DurationMs = 100 # Simulate 100ms of audio

type AudioSample = object
  left: float32
  right: float32
  timestamp: int64

var
  queue = initSipsic[BufferSize, AudioSample]()
  running: Atomic[bool]
  samplesProduced: Atomic[int]
  samplesConsumed: Atomic[int]
  underruns: Atomic[int] # Consumer needed data but queue was empty
  overruns: Atomic[int] # Producer couldn't push because queue was full

proc captureThread() {.thread.} =
  ## Simulates audio capture hardware filling the buffer.
  ## In a real application, this would be driven by hardware interrupts.
  var sampleIndex = 0
  let totalSamples = (SampleRate * DurationMs) div 1000

  while sampleIndex < totalSamples:
    # Generate a simple sine wave (440Hz tone)
    let t = float32(sampleIndex) / float32(SampleRate)
    let value = sin(t * 440.0 * 2.0 * PI).float32 * 0.5

    let sample = AudioSample(left: value, right: value, timestamp: sampleIndex)

    if queue.push(sample):
      discard samplesProduced.fetchAdd(1, moRelaxed)
      inc sampleIndex
    else:
      # Buffer full - would cause overrun in real audio
      discard overruns.fetchAdd(1, moRelaxed)
      # In real audio, we'd drop the sample or wait for hardware timing
      sleep(0) # Yield to let consumer catch up

  running.store(false, moRelease)

proc playbackThread() {.thread.} =
  ## Simulates audio playback hardware draining the buffer.
  ## In a real application, this would be driven by hardware interrupts.

  while running.load(moAcquire) or queue.pop().isSome:
    let sample = queue.pop()

    if sample.isSome:
      # In a real application, send to DAC
      discard samplesConsumed.fetchAdd(1, moRelaxed)
    else:
      # Buffer empty - would cause underrun (glitch) in real audio
      discard underruns.fetchAdd(1, moRelaxed)

    # Simulate playback timing (~22.7us per sample at 44.1kHz)
    # In reality, hardware timing drives this
    sleep(0)

when isMainModule:
  echo "Audio Buffer Example"
  echo "===================="
  echo "Buffer size: ", BufferSize, " samples"
  echo "Latency: ", (BufferSize * 1000) div SampleRate, "ms"
  echo "Duration: ", DurationMs, "ms"
  echo ""

  running.store(true, moRelease)
  samplesProduced.store(0, moRelaxed)
  samplesConsumed.store(0, moRelaxed)
  underruns.store(0, moRelaxed)
  overruns.store(0, moRelaxed)

  var threads: array[2, Thread[void]]
  threads[0].createThread(captureThread)
  threads[1].createThread(playbackThread)

  joinThreads(threads)

  echo "Results:"
  echo "  Samples produced: ", samplesProduced.load(moRelaxed)
  echo "  Samples consumed: ", samplesConsumed.load(moRelaxed)
  echo "  Underruns: ", underruns.load(moRelaxed)
  echo "  Overruns: ", overruns.load(moRelaxed)

  let produced = samplesProduced.load(moRelaxed)
  let consumed = samplesConsumed.load(moRelaxed)
  if produced == consumed and underruns.load(moRelaxed) == 0:
    echo ""
    echo "Clean audio stream - no glitches!"
  else:
    echo ""
    echo "Note: Some underruns/overruns expected in simulation"
    echo "(Real audio uses hardware timing, not sleep())"

---

Task Fan-Out (Bounded SPMC)

Work distribution from a single dispatcher to multiple workers.

Key properties:

• Work-stealing via CAS coordination
• Natural load balancing (faster workers get more tasks)
• Bounded memory provides backpressure

Use cases: HTTP request routing, image processing, game engine jobs

## Task Fan-Out Example
##
## Demonstrates using a bounded Sipmuc (SPMC) queue for distributing work
## from a single producer to multiple consumer workers.
##
## Pattern: Single dispatcher → Multiple workers
##
## Key properties:
## - Work-stealing: consumers compete for tasks via CAS
## - Load balancing: faster workers naturally get more tasks
## - Bounded memory: fixed queue size provides backpressure
## - Wait-free push: dispatcher never blocks
##
## Use cases:
## - HTTP request routing to worker pool
## - Image processing pipeline
## - Game engine job distribution

import lockfreequeues/atomic_dsl
import os
import options
import std/monotimes
import times

import lockfreequeues

const
  QueueCapacity = 128
  NumWorkers = 4
  NumTasks = 1000

type
  TaskKind = enum
    tkCompute # CPU-bound work
    tkIO # Simulated I/O
    tkFast # Quick task

  Task = object
    id: int
    kind: TaskKind
    payload: int

var
  queue = initSipmuc[QueueCapacity, NumWorkers, Task]()
  done: Atomic[bool]
  tasksCompleted: array[NumWorkers, Atomic[int]]
  totalLatency: array[NumWorkers, Atomic[int64]]

proc simulateWork(task: Task) =
  ## Simulate different types of work
  case task.kind
  of tkCompute:
    # Simulate CPU work with busy loop
    var sum = 0
    for i in 0 ..< task.payload:
      sum += i
    discard sum
  of tkIO:
    # Simulate I/O wait
    sleep(task.payload)
  of tkFast:
    # Minimal work
    discard

proc workerThread(idx: int) {.thread.} =
  ## Worker consumes tasks from the queue until shutdown.
  let consumer = queue.getConsumer(idx)
  var completed = 0
  var latencySum: int64 = 0

  while true:
    let task = consumer.pop()

    if task.isSome:
      let start = getMonoTime()
      simulateWork(task.get)
      let elapsed = (getMonoTime() - start).inMicroseconds
      latencySum += elapsed
      inc completed
    elif done.load(moAcquire):
      break
    else:
      # No work available, brief pause before retry
      sleep(0)

  tasksCompleted[idx].store(completed, moRelease)
  totalLatency[idx].store(latencySum, moRelease)

proc dispatcherThread() {.thread.} =
  ## Dispatcher generates and distributes tasks.
  var taskId = 0

  for i in 0 ..< NumTasks:
    # Create varied task mix
    let kind =
      case i mod 10
      of 0 .. 2: tkFast
      of 3 .. 6: tkCompute
      else: tkIO

    let payload =
      case kind
      of tkFast:
        0
      of tkCompute:
        1000 + (i mod 500)
      of tkIO:
        1 + (i mod 3)

    let task = Task(id: taskId, kind: kind, payload: payload)
    inc taskId

    # Push with backpressure - wait if queue is full
    while not queue.push(task):
      sleep(0)

  done.store(true, moRelease)

when isMainModule:
  echo "Task Fan-Out Example"
  echo "===================="
  echo "Queue capacity: ", QueueCapacity
  echo "Workers: ", NumWorkers
  echo "Tasks: ", NumTasks
  echo ""

  done.store(false, moRelaxed)
  for i in 0 ..< NumWorkers:
    tasksCompleted[i].store(0, moRelaxed)
    totalLatency[i].store(0, moRelaxed)

  let startTime = getMonoTime()

  # Start workers
  var workers: array[NumWorkers, Thread[int]]
  for i in 0 ..< NumWorkers:
    createThread(workers[i], workerThread, i)

  # Start dispatcher
  var dispatcher: Thread[void]
  createThread(dispatcher, dispatcherThread)

  # Wait for completion
  joinThread(dispatcher)
  sleep(50) # Let workers drain queue
  for i in 0 ..< NumWorkers:
    joinThread(workers[i])

  let totalTime = (getMonoTime() - startTime).inMilliseconds

  # Report results
  echo "Results:"
  var totalCompleted = 0
  for i in 0 ..< NumWorkers:
    let completed = tasksCompleted[i].load(moAcquire)
    let latency = totalLatency[i].load(moAcquire)
    let avgLatency =
      if completed > 0:
        latency div completed
      else:
        0
    echo "  Worker ", i, ": ", completed, " tasks, avg ", avgLatency, "us/task"
    totalCompleted += completed

  echo ""
  echo "Total completed: ", totalCompleted, "/", NumTasks
  echo "Total time: ", totalTime, "ms"
  echo "Throughput: ", (NumTasks * 1000) div max(1, totalTime.int), " tasks/sec"

  if totalCompleted == NumTasks:
    echo ""
    echo "All tasks processed successfully!"

---

Event Collector (Unbounded MPSC)

Collecting events from multiple sources into a single processing pipeline.

Key properties:

• Handles traffic bursts (queue grows during spikes)
• Never drops events
• Lock-free producers don't block each other

Use cases: Log aggregation, metrics collection, network packet capture

## Event Collector Example
##
## Demonstrates using an unbounded Mupsic (MPSC) queue for collecting events
## from multiple sources into a single processing pipeline.
##
## Pattern: Multiple event sources → Single processor
##
## Key properties:
## - Handles bursts: queue grows during traffic spikes
## - Never drops: events are queued even during overload
## - Memory reclamation: epoch-based cleanup when safe
## - Lock-free producers: sources don't block each other
##
## Use cases:
## - Log aggregation from multiple services
## - Metrics collection from distributed sensors
## - Network packet capture from multiple interfaces
## - User activity tracking across browser tabs

import lockfreequeues/atomic_dsl
import os
import options
import random
import std/monotimes
import times

import debra
import lockfreequeues

const
  SegmentSize = 64
  NumSources = 4
  DurationMs = 200
  MaxThreads = NumSources + 4 # sources + main/consumer + slack

type
  EventKind = enum
    ekClick
    ekPageView
    ekError
    ekMetric

  Event = object
    sourceId: int
    kind: EventKind
    timestamp: int64
    value: int

  SourceContext = object
    queue: ptr UnboundedMupsic[SegmentSize, Event, MaxThreads]
    manager: ptr DebraManager[MaxThreads]
    sourceId: int
    startTime: MonoTime

  ProcessorContext = object
    queue: ptr UnboundedMupsic[SegmentSize, Event, MaxThreads]
    manager: ptr DebraManager[MaxThreads]
    consumerHandle: ThreadHandle[MaxThreads]

var
  manager = initDebraManager[MaxThreads]()
  consumerHandle = registerThread(manager)
  queue =
    newUnboundedMupsic[SegmentSize, Event, MaxThreads](addr manager, consumerHandle)
  running: Atomic[bool]
  eventsProduced: array[NumSources, Atomic[int]]
  eventsConsumed: Atomic[int]
  maxQueueDepth: Atomic[int]

proc eventSourceThread(ctx: ptr SourceContext) {.thread.} =
  ## Simulates an event source generating events at variable rates.
  ## Sources occasionally burst to simulate real traffic patterns.
  {.cast(gcsafe).}:
    let handle = registerThread(ctx.manager[])
    var producer = ctx.queue[].getProducer(handle)
    var produced = 0

    while running.load(moAcquire):
      # Simulate variable event rate with occasional bursts
      let burstMode = rand(100) < 10 # 10% chance of burst
      let eventCount =
        if burstMode:
          rand(10 .. 20)
        else:
          rand(1 .. 3)

      for _ in 0 ..< eventCount:
        let event = Event(
          sourceId: ctx.sourceId,
          kind: EventKind(rand(ord(EventKind.high))),
          timestamp: (getMonoTime() - ctx.startTime).inMicroseconds,
          value: rand(1000),
        )

        producer.push(event)
        inc produced

      # Variable delay between event batches
      let delayMs =
        if burstMode:
          1
        else:
          rand(5 .. 15)
      sleep(delayMs)

    eventsProduced[ctx.sourceId].store(produced, moRelease)

proc processorThread(ctx: ptr ProcessorContext) {.thread.} =
  ## Single processor consumes and processes all events.
  ## Periodically triggers memory reclamation.
  {.cast(gcsafe).}:
    var consumed = 0
    var maxDepth = 0
    var lastReclaim = getMonoTime()
    var eventCounts: array[EventKind, int]

    while running.load(moAcquire) or ctx.queue[].len() > 0:
      let event = ctx.queue[].pop()

      if event.isSome:
        # Process the event
        inc eventCounts[event.get.kind]
        inc consumed

        # Track queue depth
        let depth = ctx.queue[].len()
        if depth > maxDepth:
          maxDepth = depth

        # Periodic memory reclamation (every 10ms)
        if (getMonoTime() - lastReclaim).inMilliseconds > 10:
          ctx.manager[].advance()
          discard reclaimNow(ctx.consumerHandle)
          lastReclaim = getMonoTime()
      else:
        # No events, brief pause
        sleep(1)

    eventsConsumed.store(consumed, moRelease)
    maxQueueDepth.store(maxDepth, moRelease)

    echo ""
    echo "Event breakdown:"
    for kind in EventKind:
      echo "  ", kind, ": ", eventCounts[kind]

when isMainModule:
  randomize()

  echo "Event Collector Example"
  echo "======================="
  echo "Segment size: ", SegmentSize
  echo "Event sources: ", NumSources
  echo "Duration: ", DurationMs, "ms"
  echo ""

  running.store(true, moRelease)
  eventsConsumed.store(0, moRelaxed)
  maxQueueDepth.store(0, moRelaxed)
  for i in 0 ..< NumSources:
    eventsProduced[i].store(0, moRelaxed)

  let startTime = getMonoTime()

  # Start processor
  var procCtx = ProcessorContext(
    queue: addr queue, manager: addr manager, consumerHandle: consumerHandle
  )
  var processor: Thread[ptr ProcessorContext]
  createThread(processor, processorThread, addr procCtx)

  # Start event sources
  var sourceContexts: array[NumSources, SourceContext]
  for i in 0 ..< NumSources:
    sourceContexts[i] = SourceContext(
      queue: addr queue, manager: addr manager, sourceId: i, startTime: startTime
    )

  var sources: array[NumSources, Thread[ptr SourceContext]]
  for i in 0 ..< NumSources:
    createThread(sources[i], eventSourceThread, addr sourceContexts[i])

  # Run for specified duration
  sleep(DurationMs)
  running.store(false, moRelease)

  # Wait for completion
  for i in 0 ..< NumSources:
    joinThread(sources[i])
  sleep(50) # Let processor drain
  joinThread(processor)

  let totalTime = (getMonoTime() - startTime).inMilliseconds

  # Report results
  echo ""
  echo "Results:"
  var totalProduced = 0
  for i in 0 ..< NumSources:
    let produced = eventsProduced[i].load(moAcquire)
    echo "  Source ", i, ": ", produced, " events"
    totalProduced += produced

  let consumed = eventsConsumed.load(moAcquire)
  let maxDepth = maxQueueDepth.load(moAcquire)

  echo ""
  echo "Total produced: ", totalProduced
  echo "Total consumed: ", consumed
  echo "Max queue depth: ", maxDepth
  echo "Final segments: ", queue.segmentCount()
  echo "Total time: ", totalTime, "ms"
  echo "Throughput: ", (consumed * 1000) div max(1, totalTime.int), " events/sec"

  if totalProduced == consumed:
    echo ""
    echo "All events processed - no data loss!"

---

Job Scheduler (Unbounded MPMC)

Dynamic job scheduling with multiple submitters and workers.

Key properties:

• Elastic capacity grows with pending work
• Lock-free operations for high concurrency
• Dynamic scaling of producers and consumers

Use cases: Background job processing, build systems, database query scheduling

## Job Scheduler Example
##
## Demonstrates using an unbounded Mupmuc (MPMC) queue for a dynamic job
## scheduling system with multiple producers and consumers.
##
## Pattern: Multiple submitters → Job queue → Multiple workers
##
## Key properties:
## - Elastic capacity: queue grows with pending work
## - Lock-free operations: submitters and workers don't block each other
## - Dynamic scaling: workers can be added/removed at runtime
## - Fair scheduling: committed flag ensures FIFO within segments
##
## Use cases:
## - Web server request handling
## - Background job processing (like Sidekiq/Celery)
## - Build system task execution
## - Database query scheduling

import lockfreequeues/atomic_dsl
import os
import options
import random
import std/monotimes
import strutils
import times

import debra
import lockfreequeues

const
  SegmentSize = 32
  NumSubmitters = 3
  NumWorkers = 4
  JobsPerSubmitter = 50
  MaxThreads = NumSubmitters + NumWorkers + 4 # producers + consumers + main + slack

type
  Priority = enum
    pLow
    pNormal
    pHigh

  Job = object
    id: int
    submitterId: int
    priority: Priority
    workMs: int # Simulated work duration

  SubmitterContext = object
    queue: ptr UnboundedMupmuc[SegmentSize, ptr Job, MaxThreads]
    manager: ptr DebraManager[MaxThreads]
    submitterId: int

  WorkerContext = object
    queue: ptr UnboundedMupmuc[SegmentSize, ptr Job, MaxThreads]
    manager: ptr DebraManager[MaxThreads]
    workerId: int

var
  manager = initDebraManager[MaxThreads]()
  queue = newUnboundedMupmuc[SegmentSize, ptr Job, MaxThreads](addr manager)
  running: Atomic[bool]
  jobsSubmitted: array[NumSubmitters, Atomic[int]]
  jobsCompleted: array[NumWorkers, Atomic[int]]
  nextJobId: Atomic[int]

proc submitterThread(ctx: ptr SubmitterContext) {.thread.} =
  ## Job submitter - creates and enqueues jobs.
  {.cast(gcsafe).}:
    let handle = registerThread(ctx.manager[])
    var producer = ctx.queue[].getProducer(handle)
    var submitted = 0

    for i in 0 ..< JobsPerSubmitter:
      let jobId = nextJobId.fetchAdd(1, moRelaxed)

      # Varied job characteristics
      let priority =
        case rand(10)
        of 0: pHigh
        of 1 .. 3: pNormal
        else: pLow

      let workMs =
        case priority
        of pHigh:
          rand(1 .. 5)
        of pNormal:
          rand(5 .. 15)
        of pLow:
          rand(10 .. 30)

      let job = create(Job)
      job[] =
        Job(id: jobId, submitterId: ctx.submitterId, priority: priority, workMs: workMs)

      producer.push(job)
      inc submitted

      # Variable submission rate
      sleep(rand(1 .. 10))

    jobsSubmitted[ctx.submitterId].store(submitted, moRelease)

proc workerThread(ctx: ptr WorkerContext) {.thread.} =
  ## Worker - fetches and executes jobs.
  {.cast(gcsafe).}:
    let handle = registerThread(ctx.manager[])
    var consumer = ctx.queue[].getConsumer(handle)
    var completed = 0
    var workTime: int64 = 0

    while running.load(moAcquire) or ctx.queue[].len() > 0:
      let job = consumer.pop()

      if job.isSome:
        let jp = job.get

        # Simulate work
        let start = getMonoTime()
        sleep(jp.workMs)
        workTime += (getMonoTime() - start).inMilliseconds

        dealloc(jp)
        inc completed

        # Periodic reclamation
        if completed mod 10 == 0:
          ctx.manager[].advance()
          discard reclaimNow(handle)
      else:
        sleep(1)

    jobsCompleted[ctx.workerId].store(completed, moRelease)
    echo "Worker ",
      ctx.workerId, " completed ", completed, " jobs (", workTime, "ms work)"

when isMainModule:
  randomize()

  echo "Job Scheduler Example"
  echo "====================="
  echo "Segment size: ", SegmentSize
  echo "Submitters: ", NumSubmitters
  echo "Workers: ", NumWorkers
  echo "Jobs per submitter: ", JobsPerSubmitter
  echo "Total jobs: ", NumSubmitters * JobsPerSubmitter
  echo ""

  running.store(true, moRelease)
  nextJobId.store(0, moRelaxed)
  for i in 0 ..< NumSubmitters:
    jobsSubmitted[i].store(0, moRelaxed)
  for i in 0 ..< NumWorkers:
    jobsCompleted[i].store(0, moRelaxed)

  let startTime = getMonoTime()

  # Start workers
  var workerContexts: array[NumWorkers, WorkerContext]
  for i in 0 ..< NumWorkers:
    workerContexts[i] =
      WorkerContext(queue: addr queue, manager: addr manager, workerId: i)

  var workers: array[NumWorkers, Thread[ptr WorkerContext]]
  for i in 0 ..< NumWorkers:
    createThread(workers[i], workerThread, addr workerContexts[i])

  # Start submitters
  var submitterContexts: array[NumSubmitters, SubmitterContext]
  for i in 0 ..< NumSubmitters:
    submitterContexts[i] =
      SubmitterContext(queue: addr queue, manager: addr manager, submitterId: i)

  var submitters: array[NumSubmitters, Thread[ptr SubmitterContext]]
  for i in 0 ..< NumSubmitters:
    createThread(submitters[i], submitterThread, addr submitterContexts[i])

  # Wait for submitters to finish
  for i in 0 ..< NumSubmitters:
    joinThread(submitters[i])

  echo "All jobs submitted, waiting for workers..."
  echo ""

  # Signal shutdown and wait for workers
  sleep(100) # Let workers drain
  running.store(false, moRelease)

  for i in 0 ..< NumWorkers:
    joinThread(workers[i])

  let totalTime = (getMonoTime() - startTime).inMilliseconds

  # Report results
  echo ""
  echo "Submission summary:"
  var totalSubmitted = 0
  for i in 0 ..< NumSubmitters:
    let submitted = jobsSubmitted[i].load(moAcquire)
    echo "  Submitter ", i, ": ", submitted, " jobs"
    totalSubmitted += submitted

  echo ""
  echo "Completion summary:"
  var totalCompleted = 0
  for i in 0 ..< NumWorkers:
    let completed = jobsCompleted[i].load(moAcquire)
    totalCompleted += completed

  echo ""
  echo "Total submitted: ", totalSubmitted
  echo "Total completed: ", totalCompleted
  echo "Queue remaining: ", queue.len()
  echo "Segments used: ", queue.segmentCount()
  echo "Total time: ", totalTime, "ms"

  if totalSubmitted == totalCompleted:
    echo ""
    echo "All jobs completed successfully!"

---

Pattern Guide

Pattern	Queue Type	Example
Real-time audio/video	Bounded Sipsic	Audio buffer
Work distribution	Bounded Sipmuc	Task fan-out
Event aggregation	Unbounded Mupsic	Event collector
Job scheduling	Unbounded Mupmuc	Job scheduler
Sensor data	Bounded Sipsic/Mupsic	-
Request routing	Bounded Sipmuc	Task fan-out
Log collection	Unbounded Mupsic	Event collector
Thread pool	Unbounded Mupmuc	Job scheduler