Class: Wurk::Scheduled::Poller

Inherits:
Object
  • Object
show all
Includes:
Component
Defined in:
lib/wurk/scheduled.rb

Overview

Single thread that wakes on a randomized interval, drains both ZSETs, then sleeps again. Random spread prevents the cluster from dogpiling Redis at the top of each cadence.

Constant Summary collapse

INITIAL_WAIT = 
10

Constants included from Component

Component::DEFAULT_THREAD_PRIORITY, Component::LEADER_CACHE_TTL_MS, Component::PROCESS_NONCE

Instance Attribute Summary collapse

Attributes included from Component

#config

Instance Method Summary collapse

Methods included from Component

#default_tag, #fire_event, #handle_exception, #hostname, #identity, #leader?, #logger, #mono_ms, #process_nonce, #real_ms, #redis, #safe_thread, #tid, #watchdog

Constructor Details

#initialize(config) ⇒ Poller

Returns a new instance of Poller.



145
146
147
148
149
150
151
152
153
154
 
# File 'lib/wurk/scheduled.rb', line 145

def initialize(config)
  @config = config
  @enq = (config[:scheduled_enq] || Enq).new(config)
  @done = false
  @mutex = ::Mutex.new
  @sleeper = ::ConditionVariable.new
  @thread = nil
  @rnd = ::Random.new
  @last_cleanup_ms = 0
end

Instance Attribute Details

#rnd ⇒ Object

Returns the value of attribute rnd.



143
144
145
 
# File 'lib/wurk/scheduled.rb', line 143

def rnd
  @rnd
end

Instance Method Details

#enqueue ⇒ Object

Called on every wake. Any raise inside the Enq is reported and the loop continues — a transient Redis blip must not kill the scheduler.



183
184
185
186
187
 
# File 'lib/wurk/scheduled.rb', line 183

def enqueue
  @enq.enqueue_jobs
rescue StandardError => e
  handle_exception(e, { context: 'scheduler' })
end

#start ⇒ Object

Spawns the scheduler thread. INITIAL_WAIT delays the first sweep so a fleet-wide deploy doesn't have every freshly-booted process hit Redis simultaneously.



159
160
161
162
163
164
165
166
167
168
 
# File 'lib/wurk/scheduled.rb', line 159

def start
  @thread ||= safe_thread('scheduler') do # rubocop:disable Naming/MemoizedInstanceVariableName
    initial_wait
    until @done
      enqueue
      wait
    end
    logger.info('Scheduler exiting...')
  end
end

#terminate ⇒ Object

Idempotent. Wakes the sleeping thread so it observes @done and exits. Also propagates the stop signal to @enq so any in-flight drain loop short-circuits instead of running to completion.



173
174
175
176
177
178
179
 
# File 'lib/wurk/scheduled.rb', line 173

def terminate
  @mutex.synchronize do
    @done = true
    @enq.terminate
    @sleeper.signal
  end
end