Class: StatsD::Instrument::BatchedUDPSink::Dispatcher

Inherits:
Object
  • Object
show all
Defined in:
lib/statsd/instrument/batched_udp_sink.rb

Instance Method Summary collapse

Constructor Details

#initialize(host, port, buffer_capacity, thread_priority, max_packet_size, statistics_interval) ⇒ Dispatcher

Returns a new instance of Dispatcher.



146
147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/statsd/instrument/batched_udp_sink.rb', line 146

def initialize(host, port, buffer_capacity, thread_priority, max_packet_size, statistics_interval)
  @udp_sink = UDPSink.new(host, port)
  @interrupted = false
  @thread_priority = thread_priority
  @max_packet_size = max_packet_size
  @buffer_capacity = buffer_capacity
  @buffer = Buffer.new(buffer_capacity)
  @dispatcher_thread = Thread.new { dispatch }
  @pid = Process.pid
  if statistics_interval > 0
    @statistics = DispatcherStats.new(statistics_interval)
  end
end

Instance Method Details

#<<(datagram) ⇒ Object 



160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/statsd/instrument/batched_udp_sink.rb', line 160

def <<(datagram)
  if !thread_healthcheck || !@buffer.push_nonblock(datagram)
    # The buffer is full or the thread can't be respawned,
    # we'll send the datagram synchronously
    @udp_sink << datagram

    @statistics&.increment_synchronous_sends
  end

  self
end

#flush(blocking:) ⇒ Object 



181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
 
# File 'lib/statsd/instrument/batched_udp_sink.rb', line 181

def flush(blocking:)
  packet = "".b
  next_datagram = nil
  until @buffer.closed? && @buffer.empty? && next_datagram.nil?
    if blocking
      next_datagram ||= @buffer.pop
      break if next_datagram.nil? # queue was closed
    else
      next_datagram ||= @buffer.pop_nonblock
      break if next_datagram.nil? # no datagram in buffer
    end
    buffer_len = @buffer.length + 1
    batch_len = 1

    packet << next_datagram
    next_datagram = nil
    if packet.bytesize <= @max_packet_size
      while (next_datagram = @buffer.pop_nonblock)
        if @max_packet_size - packet.bytesize - 1 > next_datagram.bytesize
          packet << NEWLINE << next_datagram
          batch_len += 1
        else
          break
        end
      end
    end

    packet_size = packet.bytesize
    @udp_sink << packet
    packet.clear

    @statistics&.increment_batched_sends(buffer_len, packet_size, batch_len)
    @statistics&.maybe_flush!
  end
end

#shutdown(wait = 2) ⇒ Object 



172
173
174
175
176
177
178
179
 
# File 'lib/statsd/instrument/batched_udp_sink.rb', line 172

def shutdown(wait = 2)
  @interrupted = true
  @buffer.close
  if @dispatcher_thread&.alive?
    @dispatcher_thread.join(wait)
  end
  flush(blocking: false)
end