Make the queue manager more robust at handling deceased queues.

Author: sodabrew

code/queue.rb

@@ -125,6 +125,8 @@ def self.log(path, mesg)
       File.open(path + '/queue.log', "a") do |f|
         f.write("#{Process.pid} - #{Time.now} - #{mesg}\n")
       end
+    rescue Exception
+      $stderr.write("Cannot log to #{path}: #{Process.pid} - #{Time.now} - #{mesg}\n")
     end
 
     def self.start_process(options)
@@ -180,6 +182,11 @@ def self.start_process(options)
       worker.num_restarts = 0
       worker.options = options
       worker
+
+    # If anything went wrong at all log it and return nil.
+    rescue Exception
+      self.log("Failed to start worker #{options.inspect}: #{$!}")
+      nil
     end
 
     def self.close_all_fds(exclude_fds)

code/queuemgr.rb

@@ -341,9 +341,11 @@ def run_loop
     #log(io_list.inspect)
     log('sleeping')
     begin
-      ready, _, _ = IO.select(io_list, nil, nil, 60)
-    rescue Errno::EAGAIN, Errno::EWOULDBLOCK, Errno::ECONNABORTED, Errno::EPROTO, Errno::EINTR
+      ready, _, _ = IO.select(io_list, nil, io_list, 60)
+    rescue SystemCallError # This is the parent for all Errno::EFOO exceptions
       log("error on SELECT #{$!}")
+      closed_sockets = io_list.delete_if { |i| i.closed? }
+      log("removing closed sockets #{closed_sockets.inspect} from io_list")
       retry
     end
 
@@ -392,8 +394,10 @@ def run_loop
                 log("FAILED [ #{worker.name} - too many restarts. Not restarting ]")
               else
                 new_worker = RQ::Queue.start_process(worker.options)
-                log("STARTED [ #{new_worker.name} - #{new_worker.pid} ]")
-                worker = new_worker
+                if new_worker
+                  log("STARTED [ #{new_worker.name} - #{new_worker.pid} ]")
+                  worker = new_worker
+                end
               end
             elsif worker.status == "DELETE"
               RQ::Queue.delete(worker.name)