Merge pull request #965 from fluent/fix-lifecycle-control

Fix lifecycle control

lib/fluent/engine.rb

@@ -46,6 +46,8 @@ def initialize
       @system_config = SystemConfig.new
     end
 
+    MAINLOOP_SLEEP_INTERVAL = 0.3
+
     MATCH_CACHE_SIZE = 1024
     LOG_EMIT_INTERVAL = 0.1
 
@@ -177,38 +179,24 @@ def run
           @log_emit_thread = Thread.new(&method(:log_event_loop))
         end
 
-        unless @engine_stopped
-          # for empty loop
-          @default_loop = Coolio::Loop.default
-          @default_loop.attach Coolio::TimerWatcher.new(1, true)
-          # TODO attach async watch for thread pool
-          @default_loop.run
-        end
-
-        if @engine_stopped and @default_loop
-          @default_loop.stop
-          @default_loop = nil
-        end
+        sleep MAINLOOP_SLEEP_INTERVAL until @engine_stopped
 
-      rescue => e
+      rescue Exception => e
         $log.error "unexpected error", error: e
         $log.error_backtrace
-      ensure
-        $log.info "shutting down fluentd"
-        shutdown
-        if @log_emit_thread
-          @log_event_loop_stop = true
-          @log_emit_thread.join
-        end
+        raise
+      end
+
+      $log.info "shutting down fluentd"
+      shutdown
+      if @log_emit_thread
+        @log_event_loop_stop = true
+        @log_emit_thread.join
       end
     end
 
     def stop
       @engine_stopped = true
-      if @default_loop
-        @default_loop.stop
-        @default_loop = nil
-      end
       nil
     end
 

lib/fluent/plugin/in_debug_agent.rb

@@ -42,6 +42,8 @@ def configure(conf)
     end
 
     def start
+      super
+
       if @unix_path
         require 'drb/unix'
         uri = "drbunix:#{@unix_path}"
@@ -55,6 +57,8 @@ def start
 
     def shutdown
       @server.stop_service if @server
+
+      super
     end
   end
 end

lib/fluent/plugin/in_dummy.rb

@@ -64,6 +64,7 @@ def start
     def shutdown
       @running = false
       @thread.join
+      super
     end
 
     def run

lib/fluent/plugin/in_exec.rb

@@ -103,6 +103,8 @@ def setup_parser(conf)
     end
 
     def start
+      super
+
       if @run_interval
         @finished = false
         @thread = Thread.new(&method(:run_periodic))
@@ -134,6 +136,8 @@ def shutdown
         end
         @thread.join
       end
+
+      super
     end
 
     def run

lib/fluent/plugin/in_forward.rb

@@ -55,6 +55,8 @@ def configure(conf)
     end
 
     def start
+      super
+
       @loop = Coolio::Loop.new
 
       socket_manager_path = ENV['SERVERENGINE_SOCKETMANAGER_PATH']
@@ -89,6 +91,8 @@ def shutdown
       @usock.close
       @thread.join
       @lsock.close
+
+      super
     end
 
     def listen(client)

lib/fluent/plugin/in_gc_stat.rb

@@ -50,6 +50,8 @@ def configure(conf)
     end
 
     def start
+      super
+
       @loop = Coolio::Loop.new
       @timer = TimerWatcher.new(@emit_interval, true, log, &method(:on_timer))
       @loop.attach(@timer)
@@ -60,6 +62,8 @@ def shutdown
       @loop.watchers.each {|w| w.detach }
       @loop.stop
       @thread.join
+
+      super
     end
 
     def run

lib/fluent/plugin/in_http.rb

@@ -112,7 +112,6 @@ def start
       detach_multi_process do
         super
         @km = KeepaliveManager.new(@keepalive_timeout)
-        #@lsock = Coolio::TCPServer.new(@bind, @port, Handler, @km, method(:on_request), @body_size_limit)
         @lsock = Coolio::TCPServer.new(lsock, nil, Handler, @km, method(:on_request),
                                        @body_size_limit, @format, log,
                                        @cors_allow_origins)
@@ -131,6 +130,8 @@ def shutdown
       @loop.stop
       @lsock.close
       @thread.join
+
+      super
     end
 
     def run

lib/fluent/plugin/in_monitor_agent.rb

@@ -235,6 +235,8 @@ def on_timer
     end
 
     def start
+      super
+
       log.debug "listening monitoring http server on http://#{@bind}:#{@port}/api/plugins"
       @srv = WEBrick::HTTPServer.new({
           BindAddress: @bind,
@@ -290,6 +292,8 @@ def shutdown
         @thread_for_emit.join
         @thread_for_emit = nil
       end
+
+      super
     end
 
     MONITOR_INFO = {

lib/fluent/plugin/in_object_space.rb

@@ -52,6 +52,8 @@ def configure(conf)
     end
 
     def start
+      super
+
       @loop = Coolio::Loop.new
       @timer = TimerWatcher.new(@emit_interval, true, log, &method(:on_timer))
       @loop.attach(@timer)
@@ -62,6 +64,8 @@ def shutdown
       @loop.watchers.each {|w| w.detach }
       @loop.stop
       @thread.join
+
+      super
     end
 
     def run

lib/fluent/plugin/in_stream.rb

@@ -35,6 +35,8 @@ def initialize
     end
 
     def start
+      super
+
       @loop = Coolio::Loop.new
       @lsock = listen
       @loop.attach(@lsock)
@@ -46,6 +48,8 @@ def shutdown
       @loop.stop
       @lsock.close
       @thread.join
+
+      super
     end
 
     #def listen

lib/fluent/plugin/in_syslog.rb

@@ -110,6 +110,8 @@ def configure(conf)
     end
 
     def start
+      super
+
       callback = if @use_default
                    method(:receive_data)
                  else
@@ -128,6 +130,8 @@ def shutdown
       @loop.stop
       @handler.close
       @thread.join
+
+      super
     end
 
     def run

lib/fluent/plugin/in_tail.rb

@@ -118,6 +118,8 @@ def configure_tag
     end
 
     def start
+      super
+
       if @pos_file
         @pf_file = File.open(@pos_file, File::RDWR|File::CREAT|File::BINARY, @file_perm)
         @pf_file.sync = true
@@ -139,6 +141,8 @@ def shutdown
       @loop.stop rescue nil # when all watchers are detached, `stop` raises RuntimeError. We can ignore this exception.
       @thread.join
       @pf_file.close if @pf_file
+
+      super
     end
 
     def expand_paths

lib/fluent/plugin/out_copy.rb

@@ -51,15 +51,19 @@ def configure(conf)
     end
 
     def start
-      @outputs.each {|o|
-        o.start
-      }
+      super
+
+      @outputs.each do |o|
+        o.start unless o.started?
+      end
     end
 
     def shutdown
-      @outputs.each {|o|
-        o.shutdown
-      }
+      @outputs.each do |o|
+        o.shutdown unless o.shutdown?
+      end
+
+      super
     end
 
     def emit(tag, es, chain)

lib/fluent/plugin/out_exec_filter.rb

@@ -220,21 +220,22 @@ def start
     end
 
     def before_shutdown
-      super
       log.debug "out_exec_filter#before_shutdown called"
       @children.each {|c|
         c.finished = true
       }
       sleep 0.5  # TODO wait time before killing child process
-    end
 
-    def shutdown
       super
+    end
 
+    def shutdown
       @children.reject! {|c|
         c.shutdown
         true
       }
+
+      super
     end
 
     def format_stream(tag, es)

lib/fluent/plugin/out_forward.rb

@@ -193,6 +193,8 @@ def shutdown
       end
       @thread.join if @thread
       @usock.close if @usock
+
+      super
     end
 
     def run

lib/fluent/plugin/out_null.rb

@@ -20,20 +20,6 @@ module Fluent
   class NullOutput < Output
     Plugin.register_output('null', self)
 
-    def initialize
-      super
-    end
-
-    def configure(conf)
-      super
-    end
-
-    def start
-    end
-
-    def shutdown
-    end
-
     def emit(tag, es, chain)
       chain.next
     end

lib/fluent/plugin/out_roundrobin.rb

@@ -57,17 +57,21 @@ def configure(conf)
     end
 
     def start
+      super
+
       rebuild_weight_array
 
-      @outputs.each {|o|
-        o.start
-      }
+      @outputs.each do |o|
+        o.start unless o.started?
+      end
     end
 
     def shutdown
-      @outputs.each {|o|
-        o.shutdown
-      }
+      @outputs.each do |o|
+        o.shutdown unless o.shutdown?
+      end
+
+      super
     end
 
     def emit(tag, es, chain)