diff --git a/lib/datadog/statsd/sender.rb b/lib/datadog/statsd/sender.rb index 9f50cc8..8ff7f91 100644 --- a/lib/datadog/statsd/sender.rb +++ b/lib/datadog/statsd/sender.rb @@ -50,6 +50,8 @@ def flush(sync: false) def rendez_vous # could happen if #start hasn't be called return unless message_queue + # no companion thread to answer, e.g. it could not be re-created + return unless sender_thread && sender_thread.alive? # Initialize and get the thread's sync queue queue = (@thread_class.current[:statsd_sync_queue] ||= @queue_class.new) @@ -81,6 +83,9 @@ def add(message) @message_queue = nil message_buffer.reset start + # the companion thread could not be re-created, don't queue + # messages that nothing will ever send + return if @done @flush_timer.start if @flush_timer && @flush_timer.stop? } end @@ -106,7 +111,13 @@ def start @sender_thread.name = "Statsd Sender" unless Gem::Version.new(RUBY_VERSION) < Gem::Version.new('2.3') rescue ThreadError => e @logger.debug { "Statsd: Failed to start sender thread: #{e.message}" } if @logger - @mx.synchronize { @done = true } + # #add calls #start while already holding @mx, and Ruby mutexes + # are not re-entrant + if @mx.owned? + @done = true + else + @mx.synchronize { @done = true } + end end @flush_timer.start if @flush_timer diff --git a/spec/statsd/sender_spec.rb b/spec/statsd/sender_spec.rb index 95c1f01..f830269 100644 --- a/spec/statsd/sender_spec.rb +++ b/spec/statsd/sender_spec.rb @@ -239,6 +239,35 @@ end end end + + context 'when the sender thread died and a new one cannot be started' do + let(:thread_class) { class_double(Thread) } + let(:dead_thread) { Thread.new {}.tap(&:join) } + + before do + allow(thread_class).to receive(:new).and_return(dead_thread) + subject.start + allow(thread_class).to receive(:new).and_raise(ThreadError, "can't create Thread") + allow(message_buffer).to receive(:reset) + end + + after do + subject.stop + end + + it 'drops the messages without raising' do + expect do + subject.add('message') + subject.add('message') + end.not_to raise_error + expect(subject.send(:message_queue).length).to eq 0 + end + + it 'does not wait for the missing companion thread in rendez_vous' do + subject.add('message') + Timeout.timeout(1) { subject.rendez_vous } + end + end end describe '#flush' do