Class: Magick::Adapters::AsyncWriter

Inherits:
Object
  • Object
show all
Defined in:
lib/magick/adapters/async_writer.rb

Overview

Serialized, bounded drain for asynchronous adapter writes.

With async_updates enabled: true the registry used to spawn one thread per write. That had two failure modes:

1. **No ordering.** Two writes to the same feature raced each other, so
 Redis could keep the OLDER value while memory kept the newer one —
 a permanent divergence, made worse by the trailing Pub/Sub publish
 telling every other process to load the stale value.
2. **No cap.** An Admin UI bulk toggle or a boot-time DSL apply spawned
 one thread (and one Redis connection) per write, which can exhaust
 the Redis connection pool.

This class fixes both: exactly ONE worker thread drains a bounded FIFO queue, so writes reach Redis in the order they were submitted and a burst of N writes costs one thread and one connection regardless of N.

Backpressure policy: block, then drop with a log line

When the queue is full, #submit BLOCKS the caller for up to enqueue_timeout seconds (backpressure — the queue drains while the caller waits). If it is still full when that expires, the write is DROPPED and a warning is emitted with the running drop total.

Dropping is deliberate. The alternatives are worse for a feature-flag library whose contract is "never take the host application down":

* Blocking forever turns a wedged Redis into an application-wide stall.
* Running the write inline on the caller's thread would let it overtake
the writes already queued for the same feature — reintroducing exactly
the out-of-order divergence this class exists to prevent.

A full queue means ~queue_limit writes are already backed up, i.e. Redis is down or unreachably slow; in that state the write would most likely have failed anyway, memory (the read path) still holds the correct value, and the drop is loud rather than silent.

Constant Summary collapse

DEFAULT_QUEUE_LIMIT =

Pending writes held before backpressure kicks in.

1_000
DEFAULT_ENQUEUE_TIMEOUT =

How long #submit blocks on a full queue before dropping the write.

5.0
DEFAULT_DRAIN_TIMEOUT =

How long #shutdown lets the worker drain before abandoning the rest.

5
DROP_WARN_INTERVAL =

Cap on how often a drop is logged, so a sustained outage cannot flood the log pipeline. The cumulative count is included in every line.

1.0
IDLE_WAIT =

Bounded wait in the worker's idle loop. Work is signalled explicitly; the timeout only guarantees the worker re-checks its stop flag even if a signal is ever missed.

1.0

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(queue_limit: nil, enqueue_timeout: nil, name: 'magick-async-writer', on_drop: nil) ⇒ AsyncWriter

on_drop is called as (label, total_dropped) when a write is dropped, rate-limited by DROP_WARN_INTERVAL. The host passes one so the drop goes through its own failed-write reporting (a dropped write is a write that never landed); without one the writer warns on $stderr by itself.



66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/magick/adapters/async_writer.rb', line 66

def initialize(queue_limit: nil, enqueue_timeout: nil, name: 'magick-async-writer', on_drop: nil)
  @queue_limit = positive_integer(queue_limit, DEFAULT_QUEUE_LIMIT)
  @enqueue_timeout = positive_float(enqueue_timeout, DEFAULT_ENQUEUE_TIMEOUT)
  @name = name
  @on_drop = on_drop

  @mutex = Mutex.new
  @work_available = ConditionVariable.new
  @space_available = ConditionVariable.new

  @queue = []
  @thread = nil
  @owner_pid = Process.pid
  @stopped = false

  @completed = 0
  @failed = 0
  @dropped = 0
  @last_drop_warn_at = nil
end

Instance Attribute Details

#enqueue_timeout ⇒ Object (readonly)

Returns the value of attribute enqueue_timeout.



60
61
62
 
# File 'lib/magick/adapters/async_writer.rb', line 60

def enqueue_timeout
  @enqueue_timeout
end

#queue_limit ⇒ Object (readonly)

Returns the value of attribute queue_limit.



60
61
62
 
# File 'lib/magick/adapters/async_writer.rb', line 60

def queue_limit
  @queue_limit
end

Instance Method Details

#pending ⇒ Object

Number of writes accepted but not yet sent to Redis.



144
145
146
 
# File 'lib/magick/adapters/async_writer.rb', line 144

def pending
  @mutex.synchronize { @queue.size }
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


148
149
150
151
 
# File 'lib/magick/adapters/async_writer.rb', line 148

def running?
  thread = @mutex.synchronize { @thread }
  thread ? thread.alive? : false
end

#shutdown(timeout: DEFAULT_DRAIN_TIMEOUT) ⇒ Object

Stop accepting work, drain what is already queued, and make sure no worker thread outlives this call.

Deterministic by construction: the worker gets timeout seconds to finish the backlog; anything still queued when that expires is abandoned (and reported), the thread is killed, and the queue is cleared. Producers blocked on a full queue are woken with :rejected rather than left waiting. Idempotent.

Returns a stats hash including :abandoned.



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/magick/adapters/async_writer.rb', line 122

def shutdown(timeout: DEFAULT_DRAIN_TIMEOUT)
  thread = @mutex.synchronize do
    @stopped = true
    @work_available.broadcast
    @space_available.broadcast
    @thread
  end

  if thread && thread != Thread.current && !join_quietly(thread, timeout)
    thread.kill
    join_quietly(thread, 1)
  end

  @mutex.synchronize do
    abandoned = @queue.size
    @queue.clear
    @thread = nil
    stats_snapshot(abandoned: abandoned)
  end
end

#stats ⇒ Object

Counters for observability and specs: completed / failed / dropped.



158
159
160
 
# File 'lib/magick/adapters/async_writer.rb', line 158

def stats
  @mutex.synchronize { stats_snapshot }
end

#stopped? ⇒ Boolean

Returns:

  • (Boolean) 


153
154
155
 
# File 'lib/magick/adapters/async_writer.rb', line 153

def stopped?
  @mutex.synchronize { @stopped }
end

#submit(label = nil, &job) ⇒ Object

Hand a write to the worker thread. Returns:

:queued   — accepted; the worker will run it in submission order.
:dropped  — queue stayed full for `enqueue_timeout`; not run (logged).
:rejected — the writer is shut down; the caller must run it inline.

The worker thread is started on first submit (and restarted after a fork, or if a job killed it), so an idle registry costs no threads.

Raises:

  • (ArgumentError) 


95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/magick/adapters/async_writer.rb', line 95

def submit(label = nil, &job)
  raise ArgumentError, 'AsyncWriter#submit requires a block' unless job

  @mutex.synchronize do
    reset_after_fork
    return :rejected if @stopped

    start_worker
    admission = wait_for_space(label)
    return admission unless admission == :queued

    @queue << [label, job]
    @work_available.signal
    :queued
  end
end