2.x: Upgrade to Gradle 4.3.1, add TakeUntilPerf (ReactiveX#6029)

build.gradle

@@ -9,7 +9,7 @@ buildscript {
   dependencies {
     classpath "ru.vyarus:gradle-animalsniffer-plugin:1.2.0"
     classpath "gradle.plugin.nl.javadude.gradle.plugins:license-gradle-plugin:0.13.1"
-    classpath "me.champeau.gradle:jmh-gradle-plugin:0.4.4"
+    classpath "me.champeau.gradle:jmh-gradle-plugin:0.4.5"
     classpath "com.jfrog.bintray.gradle:gradle-bintray-plugin:1.7.3"
     classpath "org.jfrog.buildinfo:build-info-extractor-gradle:4.5.2"
   }
@@ -201,6 +201,7 @@ jmh {
     jmhVersion = jmhLibVersion
     humanOutputFile = null
     includeTests = false
+    jvmArgs = ["-Djmh.ignoreLock=true"]
     jvmArgsAppend = ["-Djmh.separateClasspathJAR=true"]
 
     if (project.hasProperty("jmh")) {

gradle/wrapper/gradle-wrapper.properties

@@ -2,4 +2,4 @@ distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.2.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.3.1-bin.zip

src/jmh/java/io/reactivex/TakeUntilPerf.java

@@ -0,0 +1,95 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed 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 io.reactivex;
+
+import java.util.concurrent.*;
+
+import org.openjdk.jmh.annotations.*;
+
+import io.reactivex.functions.*;
+import io.reactivex.internal.functions.Functions;
+import io.reactivex.schedulers.Schedulers;
+
+@BenchmarkMode(Mode.Throughput)
+@Warmup(iterations = 5)
+@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Fork(value = 1)
+@State(Scope.Thread)
+@AuxCounters
+public class TakeUntilPerf implements Consumer<Integer> {
+
+    public volatile int items;
+
+    static final int count = 10000;
+
+    Flowable<Integer> flowable;
+
+    Observable<Integer> observable;
+
+    @Override
+    public void accept(Integer t) throws Exception {
+        items++;
+    }
+
+    @Setup
+    public void setup() {
+
+        flowable = Flowable.range(1, 1000 * 1000).takeUntil(Flowable.fromCallable(new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+                int c = count;
+                while (items < c) { }
+                return 1;
+            }
+        }).subscribeOn(Schedulers.single()));
+
+        observable = Observable.range(1, 1000 * 1000).takeUntil(Observable.fromCallable(new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+                int c = count;
+                while (items < c) { }
+                return 1;
+            }
+        }).subscribeOn(Schedulers.single()));
+    }
+
+    @Benchmark
+    public void flowable() {
+        final CountDownLatch cdl = new CountDownLatch(1);
+
+        flowable.subscribe(this, Functions.emptyConsumer(), new Action() {
+            @Override
+            public void run() throws Exception {
+                cdl.countDown();
+            }
+        });
+
+        while (cdl.getCount() != 0) { }
+    }
+
+    @Benchmark
+    public void observable() {
+        final CountDownLatch cdl = new CountDownLatch(1);
+
+        observable.subscribe(this, Functions.emptyConsumer(), new Action() {
+            @Override
+            public void run() throws Exception {
+                cdl.countDown();
+            }
+        });
+
+        while (cdl.getCount() != 0) { }
+    }
+}