Always lock chunks first to avoid deadlock

[abicky]

This PR resolves #1549.

For example, deadlock occurs by the following steps before this commit:

1. input plugin thread receives multiple events
   (metadata_and_data.size > 1)
2. input plugin thread processes the first metadata_and_data element
   and aquires the lock of chunk1 (chunk.mon_exit in Buffer#write)
3. enqueue thread aquires the lock of buffer
   (synchronize in Buffer#enqueue_chunk)
4. enqueue thread tries to aquire the lock of chunk1
   (chunk.synchronize in Buffer#enqueue_chunk)
5. input plugin thread processes the second metadata_and_data element
   and tries to aquire the lock of buffer (synchronize in Buffer#write_once)

cf. #1549 (comment)

[abicky]

@repeatedly Could you review the change? I want to try the latest version (v0.14.22.rc2?) which includes this change in our production environment if it looks good to you.

[repeatedly]

Thanks! I will review it!

[abicky]

I found there were errors like unexpected error while checking flushed chunks. ignored. error_class=RuntimeError error="can't enqueue buffer file: path = /fluentd/log/..., error = 'closed stream'" and I'll investigate the cause 🙇

[abicky]

I fixed "closed stream" error in 1a6f16b.

@repeatedly Please review. I've checked the behavior in our development environment and it works fine. I want your review before I try it in our production environment.
Some tests failed in Travis CI, but these tests seem to sometimes fails also in master branch, aren't they?

[repeatedly]

No need synchronized guard here?

[repeatedly]

Hmm... deleting metadata from @stage happens in only enqueue_chunk so it seems safe in CRuby.

[abicky]

I think data race doesn't occur in CRuby thanks to GIL and synchronized is not necessary, but I'll change the code like below for other Ruby implementations:

diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb
index 337b3b79..61a904ed 100644
--- a/lib/fluent/plugin/buffer.rb
+++ b/lib/fluent/plugin/buffer.rb
@@ -351,11 +351,15 @@ module Fluent
       def enqueue_all
         log.trace "enqueueing all chunks in buffer", instance: self.object_id
         if block_given?
-          synchronize{ @stage.keys }.each do |metadata|
-            chunk = @stage[metadata]
-            next unless chunk
-            v = yield metadata, chunk
-            enqueue_chunk(metadata) if v
+          metadata_array = []
+          synchronize do
+            @stage.each do |metadata, chunk|
+              v = yield metadata, chunk
+              metadata_array << metadata if v
+            end
+          end
+          metadata_array.each do |metadata|
+            enqueue_chunk(metadata)
           end
         else
           synchronize{ @stage.keys }.each do |metadata|

[repeatedly]

Yes and currently fluentd focus on CRuby so I think code change is not needed for now.
But adding comment is better for the future.

[abicky]

Okey, I added a comment 2261428

[repeatedly]

I re-run failed tests and all tests are passed.
So your patch doesn't break existing expected behaviour.