Fix a bug that BufferChunkOverflowError cause a whole data loss

[kenhys]

Which issue(s) this PR fixes:

Fixes #

What this PR does / why we need it:

In the previous version, if BufferChunkOverflowError occurs,
whole data is not sent because of exception.
It is not appropriate because the problem exists only a partial split
data.

In this commit, detect BufferChunkOverflowError and postpone to raise
it as an exception.
So a valid partial split is processed as intended.

Docs Changes:

N/A

Release Note:

N/A

[kenhys]

TODO: need to fix error.

[kenhys]

TODO: need to fix error.

Resolved.

[kenhys]

micro bench:

It has slightly performance drawbacks.

Before:

  Calculating -------------------------------------
  no BufferChunkOverflowError
                          116.903k (± 1.4%) i/s -      1.172M in  10.030026s
  raise BufferChunkOverflowError in every 100 calls
                           40.906k (± 4.1%) i/s -    411.060k in  10.065067s

  Comparison:
  no BufferChunkOverflowError:   116903.2 i/s
  raise BufferChunkOverflowError in every 100 calls:    40905.8 i/s - 2.86x  (± 0.00) slower

After:

  Calculating -------------------------------------
  no BufferChunkOverflowError
                          116.276k (± 1.1%) i/s -      1.163M in  10.007341s
  raise BufferChunkOverflowError in every 100 calls
                           40.179k (± 4.9%) i/s -    401.472k in  10.016561s

  Comparison:
  no BufferChunkOverflowError:   116276.1 i/s
  raise BufferChunkOverflowError in every 100 calls:    40179.0 i/s - 2.89x  (± 0.00) slower

+    test '#write BufferChunkOverflowError micro benchmark' do
+      require 'benchmark/ips'
+      es = Fluent::ArrayEventStream.new([ [event_time('2016-04-11 16:00:02 +0000'), {"message" => "x"}] ])
+      large = Fluent::ArrayEventStream.new([ [event_time('2016-04-11 16:00:02 +0000'), {"message" => "x" * 1_280_000}] ])
+      Benchmark.ips do |x|
+        x.config(:time => 10, :warmup => 2)
+        x.report("no BufferChunkOverflowError") do
+          @p.write({@dm0 => es}, format: @format)
+        end
+        x.report("raise BufferChunkOverflowError in every 100 calls") do |n|
+          n.times do |i|
+            if i % 100 == 0
+              assert_raise Fluent::Plugin::Buffer::BufferChunkOverflowError do
+                @p.write({@dm0 => large}, format: @format)
+              end
+            else
+              @p.write({@dm0 => es}, format: @format)
+            end
+          end
+        end
+        x.compare!
+      end
+    end

[cosmo0920]

I believe almost this implementation is very awesome, but we should consider that buffer should be handled by multiple output threads.
Thus, we should use synchronize blocks to block processing by other output threads while updating values with substitutions.

[kenhys]

Fixed and squashed commit.

[cosmo0920]

Looks reasonable for me.
We should wait for @ashie's review.

[ashie]

Please consider not to use an instance variable @delay_buffer_chunk_overflow_error.
I'm not sure but I think the feature can be realized without bothering synchronization.
For example, store errors into a local variable in write_step_by_step and pass it to write_once or write via 3rd argument of given blocks.

[kenhys]

I've considered two-approaches to fix again.

• propagate error via #write_once block
• propagate error by raising #write_step_by_step

The first one is confirmed that it shows better performance.

[ashie]

It seems good, but I don't yet fully understand your patch.
Please wait for a while.

[ashie]

@kenhys Could you take a look #1849 too?
Although it's a problem that is different from the one we are going to solve (whole size of a chunk vs 1 message's size), but a very near issue.

[kenhys]

@kenhys Could you take a look #1849 too?

okay, I'll check it.

[kenhys]

With extending memory chunk approach (PoC), it seems that it overcome the performance regression:

 Before:

      Warming up --------------------------------------
      no BufferChunkOverflowError
                              11.488k i/100ms
      raise BufferChunkOverflowError in every 100 calls
                               4.110k i/100ms
      Calculating -------------------------------------
      no BufferChunkOverflowError
                              113.353k (± 1.0%) i/s -      1.137M in  10.034300s
      raise BufferChunkOverflowError in every 100 calls
                               40.599k (± 3.5%) i/s -    406.890k in  10.034076s

      Comparison:
      no BufferChunkOverflowError:   113353.4 i/s
      raise BufferChunkOverflowError in every 100 calls:    40599.5 i/s - 2.79x  (± 0.00) slower

    After:

      Warming up --------------------------------------
      no BufferChunkOverflowError
                              11.399k i/100ms
      raise BufferChunkOverflowError in every 100 calls
                               4.301k i/100ms
      Calculating -------------------------------------
      no BufferChunkOverflowError
                              113.855k (± 1.1%) i/s -      1.140M in  10.013090s
      raise BufferChunkOverflowError in every 100 calls
                               43.683k (± 4.0%) i/s -    438.702k in  10.059251s

      Comparison:
      no BufferChunkOverflowError:   113854.6 i/s
      raise BufferChunkOverflowError in every 100 calls:    43683.3 i/s - 2.61x  (± 0.00) slower

[kenhys]

TODO: also do the same thing for file chunk.

[ashie]

Waiting CI...

[ashie]

Thanks for your work!
TODO: Additional work is required for #1849

[cosmo0920]

🎉