Class: Cosmo::Processor

Inherits:
Object
  • Object
show all
Defined in:
lib/cosmo/processor.rb,
sig/cosmo/processor.rbs

Direct Known Subclasses

Job::Processor, Stream::Processor

Constant Summary collapse

STREAM_PAUSED_RECHECK_TTL =

Seconds a stream's paused state is cached before re-checking (override via COSMO_STREAM_PAUSED_RECHECK_TTL)

Returns:

  • (Float) 
5.0
STREAMS_PAUSED_IDLE_SLEEP =

Seconds to sleep when every stream is paused, preventing a tight CPU spin (override via COSMO_STREAMS_PAUSED_IDLE_SLEEP)

Returns:

  • (Float) 
1.0
STREAM_EMPTY_BACKOFF_MAX =

Max seconds to sleep between empty fetches (override via COSMO_STREAM_EMPTY_BACKOFF_MAX)

Returns:

  • (Float) 
5.0

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(pool, running, options) ⇒ Processor

Returns a new instance of Processor.

Parameters:



15
16
17
18
19
20
21
22
 
# File 'lib/cosmo/processor.rb', line 15

def initialize(pool, running, options)
  @pool = pool
  @running = running
  @options = options
  @threads = []
  @consumers = []
  @cache = Utils::TTLCache.new
end

Instance Attribute Details

#consumers ⇒ Array[untyped] (readonly)

Returns the value of attribute consumers.

Returns:

  • (Array[untyped]) 


13
14
15
 
# File 'lib/cosmo/processor.rb', line 13

def consumers
  @consumers
end

Class Method Details

.run ⇒ Processor

Parameters:

  • (Object)

Returns:



9
10
11
 
# File 'lib/cosmo/processor.rb', line 9

def self.run(...)
  new(...).tap(&:run)
end

Instance Method Details

#client ⇒ Client

Returns:



153
154
155
 
# File 'lib/cosmo/processor.rb', line 153

def client
  Client.instance
end

#consumer_state ⇒ Object

Returns:

  • (Object) 


166
167
168
 
# File 'lib/cosmo/processor.rb', line 166

def consumer_state
  @consumer_state ||= Concurrent::Map.new
end

#fetch(subscription, batch_size:, timeout:) ⇒ void

This method returns an undefined value.

Parameters:

  • subscription (Object)
  • batch_size: (Integer)
  • timeout: (Float) 


140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/cosmo/processor.rb', line 140

def fetch(subscription, batch_size:, timeout:)
  subscription.fetch(batch_size, timeout:)
rescue NATS::Timeout
  # No messages, continue
rescue StandardError => e
  Logger.error "Snap! Error just happened"
  Logger.error "#{e.class}: #{e.message}\n#{e.backtrace.join("\n")}"

  backoff = ENV.fetch("COSMO_STREAMS_FETCH_BACKOFF", 5).to_f
  sleep([timeout, backoff].max) # backoff before retry
  nil
end

#fetch_subjects ⇒ Object

Parameters:

  • config (Hash[Symbol, untyped])

Returns:

  • (Object) 


42
 
# File 'sig/cosmo/processor.rbs', line 42

def fetch_subjects: (Hash[Symbol, untyped] config) -> untyped

#fetch_timeout ⇒ Float

Parameters:

  • config (Hash[Symbol, untyped])

Returns:

  • (Float) 


40
 
# File 'sig/cosmo/processor.rbs', line 40

def fetch_timeout: (Hash[Symbol, untyped] config) -> Float

#lock(stream_name) { ... } ⇒ void

This method returns an undefined value.

Parameters:

  • stream_name (::String)

Yields:

Yield Returns:

  • (void) 


161
162
163
164
 
# File 'lib/cosmo/processor.rb', line 161

def lock(stream_name, &)
  @locks ||= Hash.new { |h, k| h[k] = Mutex.new }
  @locks[stream_name].synchronize(&)
end

#process ⇒ void

This method returns an undefined value.

Parameters:

  • (Object)

Raises:



128
129
130
 
# File 'lib/cosmo/processor.rb', line 128

def process(...)
  raise NotImplementedError
end

#run ⇒ void

This method returns an undefined value.



24
25
26
27
28
29
30
 
# File 'lib/cosmo/processor.rb', line 24

def run
  setup
  return unless @consumers.any?

  @running.make_true
  run_loop
end

#run_loop ⇒ void

This method returns an undefined value.



44
45
46
47
 
# File 'lib/cosmo/processor.rb', line 44

def run_loop
  @threads << Thread.new { work_loop }
  @threads << Thread.new { schedule_loop } if scheduler?
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


132
133
134
 
# File 'lib/cosmo/processor.rb', line 132

def running?
  @running.true?
end

#schedule_loop ⇒ void

This method returns an undefined value.

Raises:



120
121
122
 
# File 'lib/cosmo/processor.rb', line 120

def schedule_loop
  raise NotImplementedError
end

#scheduler? ⇒ Boolean

Returns:

  • (Boolean) 


136
137
138
 
# File 'lib/cosmo/processor.rb', line 136

def scheduler?
  false
end

#setup ⇒ void

This method returns an undefined value.

Raises:



124
125
126
 
# File 'lib/cosmo/processor.rb', line 124

def setup
  raise NotImplementedError
end

#stop(timeout = ) ⇒ void

This method returns an undefined value.

Parameters:

  • timeout (::Integer, ::Float) (defaults to: ) 


32
33
34
35
36
37
38
39
40
 
# File 'lib/cosmo/processor.rb', line 32

def stop(timeout = Config[:timeout])
  @running.make_false
  @pool.shutdown
  @consumers.each { |(s, _)| s.unsubscribe rescue nil }
  @pool.wait_for_termination(timeout)
  @threads.compact.each { _1.join(timeout) || _1.kill }
  @consumers.clear
  @threads.clear
end

#stopwatch ⇒ Utils::Stopwatch

Returns:



157
158
159
 
# File 'lib/cosmo/processor.rb', line 157

def stopwatch
  Utils::Stopwatch.new
end

#work_loop ⇒ void

This method returns an undefined value.

rubocop:disable Metrics/AbcSize, Metrics/MethodLength, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/cosmo/processor.rb', line 49

def work_loop # rubocop:disable Metrics/AbcSize, Metrics/MethodLength, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
  shutdown = false

  while running?
    break if shutdown

    all_empty = true  # every stream is empty
    all_paused = true # every stream is paused
    consumers.each do |(subscription, config, processor)| # rubocop:disable Metrics/BlockLength
      break unless running?

      stream_name = config[:stream].to_s
      ttl = ENV.fetch("COSMO_STREAM_PAUSED_RECHECK_TTL", STREAM_PAUSED_RECHECK_TTL).to_f
      if @cache.fetch(stream_name, ttl:) { API::Stream.new(stream_name).paused? }
        Logger.debug "stream #{stream_name} is paused, skipping fetch"
        next
      end
      all_paused = false

      _, skip_t = consumer_state[stream_name]
      if skip_t && Time.now < skip_t
        Logger.debug "stream #{stream_name} is empty, backing off"
        next
      end
      all_empty = false

      begin
        @pool.post do
          # Re-check, after possibly being blocked on post to thread pool
          _, skip_t = consumer_state[stream_name]
          next if skip_t && Time.now < skip_t

          timeout = fetch_timeout(config)
          Logger.debug "fetching #{fetch_subjects(config).inspect}, timeout=#{timeout}"
          messages = lock(stream_name) { fetch(subscription, batch_size: config[:batch_size], timeout:) }
          Logger.debug "fetched (#{messages&.size.to_i}) messages"
          if messages&.any?
            consumer_state.delete(stream_name)
            process(messages, processor)
          else
            max_backoff = ENV.fetch("COSMO_STREAM_EMPTY_BACKOFF_MAX", STREAM_EMPTY_BACKOFF_MAX).to_f
            consumer_state.compute(stream_name) do |current|
              count = (current&.first || 0) + 1
              backoff = [timeout * (2**(count - 1)), max_backoff].min
              [count, Time.now + backoff]
            end
          end
        end
      rescue Concurrent::RejectedExecutionError
        shutdown = true
        break # pool doesn't accept new jobs, we are shutting down
      end
    end

    break unless running?

    if all_paused
      period = ENV.fetch("COSMO_STREAMS_PAUSED_IDLE_SLEEP", STREAMS_PAUSED_IDLE_SLEEP).to_f
      Logger.debug "all streams paused, sleep=#{period}"
      sleep(period)
    elsif all_empty
      next_wake = consumer_state.values.filter_map { |_, t| t }.min
      next unless next_wake # entry was deleted concurrently (messages arrived), re-loop immediately

      remaining = [next_wake - Time.now, 0.01].max
      Logger.debug "all streams empty, sleep=#{remaining}"
      sleep(remaining)
    end
  end
end