Class: LogStash::Util::WrappedAckedQueue::ReadClient
- Inherits:
-
Object
- Object
- LogStash::Util::WrappedAckedQueue::ReadClient
- Defined in:
- lib/logstash/util/wrapped_acked_queue.rb
Instance Method Summary collapse
- #add_filtered_metrics(batch) ⇒ Object
- #add_output_metrics(batch) ⇒ Object
- #add_starting_metrics(batch) ⇒ Object
- #close ⇒ Object
- #close_batch(batch) ⇒ Object
- #current_inflight_batch ⇒ Object
- #inflight_batches ⇒ Object
-
#initialize(queue, batch_size = 125, wait_for = 250) ⇒ ReadClient
constructor
We generally only want one thread at a time able to access pop/take/poll operations from this queue.
- #set_batch_dimensions(batch_size, wait_for) ⇒ Object
- #set_current_thread_inflight_batch(batch) ⇒ Object
- #set_events_metric(metric) ⇒ Object
- #set_pipeline_metric(metric) ⇒ Object
- #start_clock ⇒ Object
- #stop_clock ⇒ Object
- #take_batch ⇒ Object
Constructor Details
#initialize(queue, batch_size = 125, wait_for = 250) ⇒ ReadClient
We generally only want one thread at a time able to access pop/take/poll operations from this queue. We also depend on this to be able to block consumers while we snapshot in-flight buffers
110 111 112 113 114 115 116 117 118 119 120 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 110 def initialize(queue, batch_size = 125, wait_for = 250) @queue = queue @mutex = Mutex.new # Note that @inflight_batches as a central mechanism for tracking inflight # batches will fail if we have multiple read clients in the pipeline. @inflight_batches = {} # allow the worker thread to report the execution time of the filter + output @inflight_clocks = {} @batch_size = batch_size @wait_for = wait_for end |
Instance Method Details
#add_filtered_metrics(batch) ⇒ Object
192 193 194 195 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 192 def add_filtered_metrics(batch) @event_metric.increment(:filtered, batch.filtered_size) @pipeline_metric.increment(:filtered, batch.filtered_size) end |
#add_output_metrics(batch) ⇒ Object
197 198 199 200 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 197 def add_output_metrics(batch) @event_metric.increment(:out, batch.filtered_size) @pipeline_metric.increment(:out, batch.filtered_size) end |
#add_starting_metrics(batch) ⇒ Object
186 187 188 189 190 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 186 def add_starting_metrics(batch) return if @event_metric.nil? || @pipeline_metric.nil? @event_metric.increment(:in, batch.starting_size) @pipeline_metric.increment(:in, batch.starting_size) end |
#close ⇒ Object
122 123 124 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 122 def close @queue.close end |
#close_batch(batch) ⇒ Object
166 167 168 169 170 171 172 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 166 def close_batch(batch) @mutex.synchronize do batch.close @inflight_batches.delete(Thread.current) stop_clock end end |
#current_inflight_batch ⇒ Object
145 146 147 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 145 def current_inflight_batch @inflight_batches.fetch(Thread.current, []) end |
#inflight_batches ⇒ Object
139 140 141 142 143 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 139 def inflight_batches @mutex.synchronize do yield(@inflight_batches) end end |
#set_batch_dimensions(batch_size, wait_for) ⇒ Object
126 127 128 129 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 126 def set_batch_dimensions(batch_size, wait_for) @batch_size = batch_size @wait_for = wait_for end |
#set_current_thread_inflight_batch(batch) ⇒ Object
162 163 164 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 162 def set_current_thread_inflight_batch(batch) @inflight_batches[Thread.current] = batch end |
#set_events_metric(metric) ⇒ Object
131 132 133 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 131 def set_events_metric(metric) @event_metric = metric end |
#set_pipeline_metric(metric) ⇒ Object
135 136 137 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 135 def set_pipeline_metric(metric) @pipeline_metric = metric end |
#start_clock ⇒ Object
174 175 176 177 178 179 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 174 def start_clock @inflight_clocks[Thread.current] = [ @event_metric.time(:duration_in_millis), @pipeline_metric.time(:duration_in_millis) ] end |
#stop_clock ⇒ Object
181 182 183 184 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 181 def stop_clock @inflight_clocks[Thread.current].each(&:stop) @inflight_clocks.delete(Thread.current) end |
#take_batch ⇒ Object
149 150 151 152 153 154 155 156 157 158 159 160 |
# File 'lib/logstash/util/wrapped_acked_queue.rb', line 149 def take_batch if @queue.closed? raise QueueClosedError.new("Attempt to take a batch from a closed AckedQueue") end @mutex.synchronize do batch = ReadBatch.new(@queue, @batch_size, @wait_for) add_starting_metrics(batch) set_current_thread_inflight_batch(batch) start_clock batch end end |