The waiter will use the main thread queue for signals and actions

Ticket: https://fatsoma.atlassian.net/browse/ENG-549

This is part of an ongoing investigation into delivery acknowledgement timeout errors that we are intermittently experiencing in production (particularly in the ticket service). This change modifies the waiter, so that a single main thread queue is used to handle process signals and actions, rather than a queue and IO.pipes.

Author: brentfatsoma

lib/hutch/waiter.rb

@@ -26,23 +26,36 @@ def initialize(broker)
     end
 
     def register_handlers
-      self.sig_read, self.sig_write = IO.pipe
+      Thread.main[:event_queue] = Queue.new
       register_signal_handlers
-
-      self.action_read, self.action_write = IO.pipe
-      Thread.main[:action_queue] = Queue.new
     end
 
     def wait_until_signaled
-      loop do
-        read_pipes = wait_for_signal.first
-        break unless read_pipes.all? { |pipe| read_pipe(pipe) }
+      # Block and wait for messages
+      while (event = event_queue.pop)
+        type, data = event
+
+        case type
+        when :signal
+          # Return false breaks the loop for graceful shutdown
+          break unless handle_signal(data)
+        when :action
+          handle_action(data)
+        else
+          raise "Assertion failed - unhandled event: #{type}"
+        end
       end
     end
 
+    # Consumer threads call this to push work to the main thread
     def push_action(action, delivery_info, properties, ex)
-      Thread.main[:action_queue] << [action, delivery_info, properties, ex]
-      action_write.write("#{delivery_info.delivery_tag}\n")
+      event_queue << [:action, {
+        action: action,
+        delivery_info: delivery_info,
+        properties: properties,
+        ex: ex,
+        pushed_at: Time.now.to_f
+      }]
     end
 
     # return true to continue processing
@@ -65,46 +78,42 @@ def handle_user_signal(sig)
 
     def handle_shutdown_signal(sig)
       logger.info "caught SIG#{sig}, stopping hutch..."
+      drain_actions
       false
     end
 
-    # return true to continue processing
-    def handle_action(_delivery_tag)
-      action, delivery_info, properties, ex = Thread.main[:action_queue].pop
-      # TODO: check delivery_tag ??
-      case action
-      when :ack then broker.ack(delivery_info.delivery_tag)
-      when :nack then acknowledge_error(delivery_info, properties, ex)
-      else raise "Assertion failed - unhandled action: #{action.inspect}"
+    def handle_action(data)
+      latency = (Time.now.to_f - data[:pushed_at]) * 1000
+
+      if latency > 5000
+        logger.warn "Queue latency exceeded 5 seconds"
       end
-      true
+
+      case data[:action]
+      when :ack then broker.ack(data[:delivery_info].delivery_tag)
+      when :nack then acknowledge_error(data[:delivery_info], data[:properties], data[:ex])
+      else raise "Assertion failed - unhandled action: #{action}"
+      end
+    rescue => e
+      logger.error "Error during #{data[:action]}: #{e.message}"
+      raise e
     end
 
     def acknowledge_error(delivery_info, properties, ex)
       acks = error_acknowledgements +
-        [Hutch::Acknowledgements::NackOnAllFailures.new]
+             [Hutch::Acknowledgements::NackOnAllFailures.new]
       acks.find do |backend|
         backend.handle(delivery_info, properties, broker, ex)
       end
     end
 
     private
 
-    def read_pipe(pipe)
-      case pipe
-      when sig_read
-        sig = sig_read.gets.chomp
-        handle_signal(sig)
-      when action_read
-        delivery_tag = action_read.gets.chomp
-        handle_action(delivery_tag)
-      end
-    end
-
     def log_thread_backtraces
       logger.info 'Requested a VM-wide thread stack trace dump...'
       Thread.list.each do |thread|
-        logger.info "Thread TID-#{thread.object_id.to_s(36)} #{thread['label']}"
+        main_label = thread == Thread.main ? 'main' : ''
+        logger.info "Thread TID-#{thread.object_id.to_s(36)} #{thread['label']} #{main_label}"
         logger.info backtrace_for(thread)
       end
     end
@@ -118,18 +127,11 @@ def backtrace_for(thread)
     end
 
     attr_reader :broker
-    attr_accessor :sig_read, :sig_write, :action_read, :action_write
-
-    def wait_for_signal
-      IO.select([sig_read, action_read])
-    end
 
     def register_signal_handlers
       REGISTERED_SIGNALS.each do |sig|
-        # This needs to be reentrant, so we queue up signals to be handled
-        # in the run loop, rather than acting on signals here
         trap(sig) do
-          sig_write.puts(sig)
+          event_queue << [:signal, sig]
         end
       end
     end
@@ -139,7 +141,27 @@ def user_signal?(sig)
     end
 
     def error_acknowledgements
-      Hutch::Config[:error_acknowledgements]
+      Hutch::Config[:error_acknowledgements] || []
+    end
+
+    def event_queue
+      queue = Thread.main[:event_queue]
+      raise 'Undefined main thread queue' unless queue
+      queue
+    end
+
+    # Drain the queue during shutdown
+    def drain_actions
+      queue = event_queue
+
+      until queue.empty?
+        begin
+          type, data = queue.pop(true)
+          handle_action(data) if type == :action
+        rescue ThreadError
+          break
+        end
+      end
     end
   end
 end

spec/hutch/worker_spec.rb

@@ -73,7 +73,7 @@
   describe '#handle_message' do
     subject do
       worker.handle_message(consumer, delivery_info, properties, payload)
-      waiter.handle_action(delivery_info.delivery_tag)
+      waiter.handle_action
     end
     let(:payload) { '{}' }
     let(:consumer_instance) { double('Consumer instance') }