Fix/avoid oom accumulation in bufftok

logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.logstash.benchmark;
+
+import org.logstash.common.BufferedTokenizer;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.util.concurrent.TimeUnit;
+
+
+@Warmup(iterations = 3, time = 100, timeUnit = TimeUnit.MILLISECONDS)
+@Measurement(iterations = 10, time = 3000, timeUnit = TimeUnit.MILLISECONDS)
+@Fork(1)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@State(Scope.Thread)
+public class BufferedTokenizerBenchmark {
+
+    private BufferedTokenizer sut;
+    private String singleTokenPerFragment;
+    private String multipleTokensPerFragment;
+    private String multipleTokensSpreadMultipleFragments_1;
+    private String multipleTokensSpreadMultipleFragments_2;
+    private String multipleTokensSpreadMultipleFragments_3;
+
+    @Setup(Level.Invocation)
+    public void setUp() {
+        sut = new BufferedTokenizer();
+        singleTokenPerFragment = "a".repeat(512) + "\n";
+
+        multipleTokensPerFragment = "a".repeat(512) + "\n" + "b".repeat(512) + "\n" + "c".repeat(512) + "\n";
+
+        multipleTokensSpreadMultipleFragments_1 = "a".repeat(512) + "\n" + "b".repeat(512) + "\n" + "c".repeat(256);
+        multipleTokensSpreadMultipleFragments_2 = "c".repeat(256) + "\n" + "d".repeat(512) + "\n" + "e".repeat(256);
+        multipleTokensSpreadMultipleFragments_3 = "f".repeat(256) + "\n" + "g".repeat(512) + "\n" + "h".repeat(512) + "\n";
+    }
+
+    @Benchmark
+    public final void onlyOneTokenPerFragment(Blackhole blackhole) {
+        Iterable<String> tokens = sut.extract(singleTokenPerFragment);
+        tokens.forEach(blackhole::consume);
+        blackhole.consume(tokens);
+    }
+
+    @Benchmark
+    public final void multipleTokenPerFragment(Blackhole blackhole) {
+        Iterable<String> tokens = sut.extract(multipleTokensPerFragment);
+        tokens.forEach(blackhole::consume);
+        blackhole.consume(tokens);
+    }
+
+    @Benchmark
+    public final void multipleTokensCrossingMultipleFragments(Blackhole blackhole) {
+        Iterable<String> tokens = sut.extract(multipleTokensSpreadMultipleFragments_1);
+        tokens.forEach(t -> {});
+        blackhole.consume(tokens);
+
+        tokens = sut.extract(multipleTokensSpreadMultipleFragments_2);
+        tokens.forEach(t -> {});
+        blackhole.consume(tokens);
+
+        tokens = sut.extract(multipleTokensSpreadMultipleFragments_3);
+        tokens.forEach(blackhole::consume);
+        blackhole.consume(tokens);
+    }
+}

logstash-core/spec/logstash/util/buftok_spec.rb

@@ -20,27 +20,33 @@
 describe  FileWatch::BufferedTokenizer  do
   subject { FileWatch::BufferedTokenizer.new }
 
+  def to_list(iterator)
+    a = []
+    iterator.each { |v| a << v }
+    return a
+  end
+
   it "should tokenize a single token" do
-    expect(subject.extract("foo\n")).to eq(["foo"])
+    expect(to_list(subject.extract("foo\n"))).to eq(["foo"])
   end
 
   it "should merge multiple token" do
-    expect(subject.extract("foo")).to eq([])
-    expect(subject.extract("bar\n")).to eq(["foobar"])
+    expect(to_list(subject.extract("foo"))).to eq([])
+    expect(to_list(subject.extract("bar\n"))).to eq(["foobar"])
   end
 
   it "should tokenize multiple token" do
-    expect(subject.extract("foo\nbar\n")).to eq(["foo", "bar"])
+    expect(to_list(subject.extract("foo\nbar\n"))).to eq(["foo", "bar"])
   end
 
   it "should ignore empty payload" do
-    expect(subject.extract("")).to eq([])
-    expect(subject.extract("foo\nbar")).to eq(["foo"])
+    expect(to_list(subject.extract(""))).to eq([])
+    expect(to_list(subject.extract("foo\nbar"))).to eq(["foo"])
   end
 
   it "should tokenize empty payload with newline" do
-    expect(subject.extract("\n")).to eq([""])
-    expect(subject.extract("\n\n\n")).to eq(["", "", ""])
+    expect(to_list(subject.extract("\n"))).to eq([""])
+    expect(to_list(subject.extract("\n\n\n"))).to eq(["", "", ""])
   end
 
   describe 'flush' do
@@ -83,12 +89,12 @@
     let(:delimiter) { "||" }
 
     it "should tokenize multiple token" do
-      expect(subject.extract("foo||b|r||")).to eq(["foo", "b|r"])
+      expect(to_list(subject.extract("foo||b|r||"))).to eq(["foo", "b|r"])
     end
 
     it "should ignore empty payload" do
-      expect(subject.extract("")).to eq([])
-      expect(subject.extract("foo||bar")).to eq(["foo"])
+      expect(to_list(subject.extract(""))).to eq([])
+      expect(to_list(subject.extract("foo||bar"))).to eq(["foo"])
     end
   end
 end

logstash-core/src/main/java/org/logstash/RubyUtil.java

@@ -34,6 +34,7 @@
 import org.logstash.ackedqueue.ext.JRubyWrappedAckedQueueExt;
 import org.logstash.common.AbstractDeadLetterQueueWriterExt;
 import org.logstash.common.BufferedTokenizerExt;
+import org.logstash.common.BufferedTokenizer;
 import org.logstash.config.ir.compiler.AbstractFilterDelegatorExt;
 import org.logstash.config.ir.compiler.AbstractOutputDelegatorExt;
 import org.logstash.config.ir.compiler.FilterDelegatorExt;