Merge pull request #2564 from ganmacs/opt-queue

Add option to check queued? loosely

lib/fluent/plugin/buffer.rb

@@ -366,13 +366,12 @@ def queued_records
         synchronize { @queue.reduce(0){|r, chunk| r + chunk.size } }
       end
 
-      def queued?(metadata=nil)
-        synchronize do
-          if metadata
-            n = @queued_num[metadata]
-            n && n.nonzero?
-          else
-            !@queue.empty?
+      def queued?(metadata = nil, optimistic: false)
+        if optimistic
+          optimistic_queued?(metadata)
+        else
+          synchronize do
+            optimistic_queued?(metadata)
           end
         end
       end
@@ -739,6 +738,15 @@ def statistics
 
       private
 
+      def optimistic_queued?(metadata = nil)
+        if metadata
+          n = @queued_num[metadata]
+          n && n.nonzero?
+        else
+          !@queue.empty?
+        end
+      end
+
       def add_timekey(t)
         @mutex.synchronize do
           @timekeys[t] += 1

lib/fluent/plugin/output.rb

@@ -795,7 +795,7 @@ def emit_buffered(tag, es)
         @counter_mutex.synchronize{ @emit_count += 1 }
         begin
           execute_chunking(tag, es, enqueue: (@flush_mode == :immediate))
-          if !@retry && @buffer.queued?
+          if !@retry && @buffer.queued?(nil, optimistic: true)
             submit_flush_once
           end
         rescue