Merge branch 'main' into fix/merge_non_empty_results

Author: javanna

benchmarks/README.md

@@ -82,19 +82,21 @@ To get realistic results, you should exercise care when running benchmarks. Here
 NOTE: Linux only. Sorry Mac and Windows.
 
 Disassembling is fun! Maybe not always useful, but always fun! Generally, you'll want to install `perf` and the JDK's `hsdis`.
-`perf` is generally available via `apg-get install perf` or `pacman -S perf`. `hsdis` you'll want to compile from source. is a little more involved. This worked
+`perf` is generally available via `apg-get install perf` or `pacman -S perf linux-tools`. `hsdis` you'll want to compile from source. is a little more involved. This worked
 on 2020-08-01:
 
 ```
 git clone [email protected]:openjdk/jdk.git
 cd jdk
-git checkout jdk-17-ga
-cd src/utils/hsdis
+git checkout jdk-24-ga
 # Get a known good binutils
 wget https://ftp.gnu.org/gnu/binutils/binutils-2.35.tar.gz
 tar xf binutils-2.35.tar.gz
-make BINUTILS=binutils-2.35 ARCH=amd64
-sudo cp build/linux-amd64/hsdis-amd64.so /usr/lib/jvm/java-17-openjdk/lib/server/
+bash configure --with-hsdis=binutils --with-binutils-src=binutils-2.35 \
+    --with-boot-jdk=~/.gradle/jdks/oracle_corporation-24-amd64-linux.2
+make build-hsdis
+cp ./build/linux-x86_64-server-release/jdk/lib/hsdis-amd64.so \
+    ~/.gradle/jdks/oracle_corporation-24-amd64-linux.2/lib/hsdis.so
 ```
 
 If you want to disassemble a single method do something like this:
@@ -105,6 +107,30 @@ gradlew -p benchmarks run --args ' MemoryStatsBenchmark -jvmArgs "-XX:+UnlockDia
 
 If you want `perf` to find the hot methods for you, then do add `-prof perfasm`.
 
+NOTE: `perfasm` will need more access:
+```
+sudo bash
+echo -1 > /proc/sys/kernel/perf_event_paranoid
+exit
+```
+
+If you get warnings like:
+```
+The perf event count is suspiciously low (0).
+```
+then check if you are bumping into [this](https://man.archlinux.org/man/perf-stat.1.en#INTEL_HYBRID_SUPPORT)
+by running:
+```
+perf stat -B dd if=/dev/zero of=/dev/null count=1000000
+```
+
+If you see lines like:
+```
+         765019980      cpu_atom/cycles/                 #    1.728 GHz                         (0.60%)
+        2258845959      cpu_core/cycles/                 #    5.103 GHz                         (99.18%)
+```
+then `perf` is just not going to work for you.
+
 ## Async Profiler
 
 Note: Linux and Mac only. Sorry Windows.

benchmarks/src/main/java/org/elasticsearch/benchmark/compute/operator/ParseIpBenchmark.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.benchmark.compute.operator;
+
+import org.apache.lucene.document.InetAddressPoint;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.breaker.NoopCircuitBreaker;
+import org.elasticsearch.common.network.InetAddresses;
+import org.elasticsearch.compute.operator.BreakingBytesRefBuilder;
+import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ParseIp;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+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.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+import java.net.InetAddress;
+import java.util.concurrent.TimeUnit;
+
+@Warmup(iterations = 5)
+@Measurement(iterations = 7)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@State(Scope.Thread)
+@Fork(1)
+public class ParseIpBenchmark {
+    private final BytesRef ip = new BytesRef("192.168.0.1");
+    private final BreakingBytesRefBuilder scratch = ParseIp.buildScratch(new NoopCircuitBreaker("request"));
+
+    @Benchmark
+    public BytesRef leadingZerosRejected() {
+        return ParseIp.leadingZerosRejected(ip, scratch);
+    }
+
+    @Benchmark
+    public BytesRef leadingZerosAreDecimal() {
+        return ParseIp.leadingZerosAreDecimal(ip, scratch);
+    }
+
+    @Benchmark
+    public BytesRef leadingZerosAreOctal() {
+        return ParseIp.leadingZerosAreOctal(ip, scratch);
+    }
+
+    @Benchmark
+    public BytesRef original() {
+        InetAddress inetAddress = InetAddresses.forString(ip.utf8ToString());
+        return new BytesRef(InetAddressPoint.encode(inetAddress));
+    }
+}

docs/changelog/126338.yaml

@@ -0,0 +1,5 @@
+pr: 126338
+summary: Speed up TO_IP
+area: ES|QL
+type: enhancement
+issues: []

server/src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java

@@ -11,6 +11,8 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRunnable;
+import org.elasticsearch.action.support.SubscribableListener;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.Processors;
@@ -25,9 +27,12 @@
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.elasticsearch.common.util.concurrent.EsExecutors.TaskTrackingConfig.DEFAULT;
@@ -295,6 +300,7 @@ public void run() {
             }
             try {
                 executor.execute(new Runnable() {
+
                     @Override
                     public void run() {
                         // Doesn't matter is going to be rejected
@@ -757,7 +763,7 @@ public void onRejection(Exception e) {
     }
 
     public void testScalingWithEmptyCore() {
-        testScalingWithEmptyCore(
+        testScalingWithEmptyCoreAndMaxSingleThread(
             EsExecutors.newScaling(
                 getTestName(),
                 0,
@@ -772,7 +778,7 @@ public void testScalingWithEmptyCore() {
     }
 
     public void testScalingWithEmptyCoreAndKeepAlive() {
-        testScalingWithEmptyCore(
+        testScalingWithEmptyCoreAndMaxSingleThread(
             EsExecutors.newScaling(
                 getTestName(),
                 0,
@@ -787,9 +793,7 @@ public void testScalingWithEmptyCoreAndKeepAlive() {
     }
 
     public void testScalingWithEmptyCoreAndLargerMaxSize() {
-        // TODO currently the reproduction of the starvation bug does not work if max pool size > 1
-        // https://github.com/elastic/elasticsearch/issues/124867
-        testScalingWithEmptyCore(
+        testScalingWithEmptyCoreAndMaxMultipleThreads(
             EsExecutors.newScaling(
                 getTestName(),
                 0,
@@ -804,9 +808,7 @@ public void testScalingWithEmptyCoreAndLargerMaxSize() {
     }
 
     public void testScalingWithEmptyCoreAndKeepAliveAndLargerMaxSize() {
-        // TODO currently the reproduction of the starvation bug does not work if max pool size > 1
-        // https://github.com/elastic/elasticsearch/issues/124867
-        testScalingWithEmptyCore(
+        testScalingWithEmptyCoreAndMaxMultipleThreads(
             EsExecutors.newScaling(
                 getTestName(),
                 0,
@@ -821,10 +823,8 @@ public void testScalingWithEmptyCoreAndKeepAliveAndLargerMaxSize() {
     }
 
     public void testScalingWithEmptyCoreAndWorkerPoolProbing() {
-        // https://github.com/elastic/elasticsearch/issues/124667 is difficult to reproduce if max pool size > 1.
-        // if probing mitigates the bug for max pool size = 1, we're good for larger pool sizes as well.
         // the executor is created directly here, newScaling doesn't use ExecutorScalingQueue & probing if max pool size = 1.
-        testScalingWithEmptyCore(
+        testScalingWithEmptyCoreAndMaxSingleThread(
             new EsThreadPoolExecutor(
                 getTestName(),
                 0,
@@ -840,10 +840,8 @@ public void testScalingWithEmptyCoreAndWorkerPoolProbing() {
     }
 
     public void testScalingWithEmptyCoreAndKeepAliveAndWorkerPoolProbing() {
-        // https://github.com/elastic/elasticsearch/issues/124667 is difficult to reproduce if max pool size > 1.
-        // if probing mitigates the bug for max pool size = 1, we're good for larger pool sizes as well.
         // the executor is created directly here, newScaling doesn't use ExecutorScalingQueue & probing if max pool size = 1.
-        testScalingWithEmptyCore(
+        testScalingWithEmptyCoreAndMaxSingleThread(
             new EsThreadPoolExecutor(
                 getTestName(),
                 0,
@@ -858,11 +856,13 @@ public void testScalingWithEmptyCoreAndKeepAliveAndWorkerPoolProbing() {
         );
     }
 
-    private void testScalingWithEmptyCore(EsThreadPoolExecutor executor) {
+    private void testScalingWithEmptyCoreAndMaxSingleThread(EsThreadPoolExecutor testSubject) {
         try {
+            final var keepAliveNanos = testSubject.getKeepAliveTime(TimeUnit.NANOSECONDS);
+
             class Task extends AbstractRunnable {
-                private int remaining;
                 private final CountDownLatch doneLatch;
+                private int remaining;
 
                 Task(int iterations, CountDownLatch doneLatch) {
                     this.remaining = iterations;
@@ -879,29 +879,108 @@ protected void doRun() {
                     if (--remaining == 0) {
                         doneLatch.countDown();
                     } else {
-                        logger.trace("--> remaining [{}]", remaining);
-                        final long keepAliveNanos = executor.getKeepAliveTime(TimeUnit.NANOSECONDS);
                         new Thread(() -> {
                             if (keepAliveNanos > 0) {
-                                final var targetNanoTime = System.nanoTime() + keepAliveNanos + between(-10_000, 10_000);
-                                while (System.nanoTime() < targetNanoTime) {
-                                    Thread.yield();
-                                }
+                                waitUntilKeepAliveTime(keepAliveNanos);
                             }
-                            executor.execute(Task.this);
+                            testSubject.execute(Task.this);
                         }).start();
                     }
                 }
             }
 
             for (int i = 0; i < 20; i++) {
-                logger.trace("--> attempt [{}]", i);
                 final var doneLatch = new CountDownLatch(1);
-                executor.execute(new Task(between(1, 500), doneLatch));
+                testSubject.execute(new Task(between(1, 500), doneLatch));
                 safeAwait(doneLatch, TimeValue.ONE_MINUTE);
             }
         } finally {
-            ThreadPool.terminate(executor, 1, TimeUnit.SECONDS);
+            ThreadPool.terminate(testSubject, 1, TimeUnit.SECONDS);
+        }
+    }
+
+    private void testScalingWithEmptyCoreAndMaxMultipleThreads(EsThreadPoolExecutor testSubject) {
+        final var keepAliveNanos = testSubject.getKeepAliveTime(TimeUnit.NANOSECONDS);
+        // Use max pool size with one additional scheduler task if a keep alive time is set.
+        final var schedulerTasks = testSubject.getMaximumPoolSize() + (keepAliveNanos > 0 ? 1 : 0);
+
+        class TaskScheduler {
+            final SubscribableListener<Void> result = new SubscribableListener<>();
+            final ExecutorService scheduler;
+            final CyclicBarrier cyclicBarrier;
+            final Semaphore taskCompletions;
+            private int remaining;
+
+            TaskScheduler(ExecutorService scheduler, int iterations) {
+                this.scheduler = scheduler;
+                this.taskCompletions = new Semaphore(0);
+                this.cyclicBarrier = new CyclicBarrier(schedulerTasks, () -> remaining--);
+                this.remaining = iterations;
+            }
+
+            public void start() {
+                // The scheduler tasks are running on the dedicated scheduler thread pool. Each task submits
+                // a test task on the EsThreadPoolExecutor (`testSubject`) releasing one `taskCompletions` permit.
+                final Runnable schedulerTask = () -> {
+                    try {
+                        while (remaining > 0) {
+                            // Wait for all scheduler threads to be ready for the next attempt.
+                            var first = cyclicBarrier.await(SAFE_AWAIT_TIMEOUT.millis(), TimeUnit.MILLISECONDS) == schedulerTasks - 1;
+                            if (first && keepAliveNanos > 0) {
+                                // The task submitted by the first scheduler task (after reaching the keep alive time) is the task
+                                // that might starve without any worker available unless an additional worker probe is submitted.
+                                waitUntilKeepAliveTime(keepAliveNanos);
+                            }
+                            // Test EsThreadPoolExecutor by submitting a task that releases one permit.
+                            testSubject.execute(taskCompletions::release);
+                            if (first) {
+                                // Let the first scheduler task (by arrival on the barrier) wait for all permits.
+                                var success = taskCompletions.tryAcquire(
+                                    schedulerTasks,
+                                    SAFE_AWAIT_TIMEOUT.millis(),
+                                    TimeUnit.MILLISECONDS
+                                );
+                                if (success == false) {
+                                    var msg = Strings.format(
+                                        "timed out waiting for [%s] of [%s] tasks to complete [queue size: %s, workers: %s] ",
+                                        schedulerTasks - taskCompletions.availablePermits(),
+                                        schedulerTasks,
+                                        testSubject.getQueue().size(),
+                                        testSubject.getPoolSize()
+                                    );
+                                    result.onFailure(new TimeoutException(msg));
+                                    return;
+                                }
+                            }
+                        }
+                    } catch (Exception e) {
+                        result.onFailure(e);
+                        return;
+                    }
+                    result.onResponse(null);
+                };
+                // Run scheduler tasks on the dedicated scheduler thread pool.
+                for (int i = 0; i < schedulerTasks; i++) {
+                    scheduler.execute(schedulerTask);
+                }
+            }
+        }
+
+        try (var scheduler = Executors.newFixedThreadPool(schedulerTasks)) {
+            for (int i = 0; i < 100; i++) {
+                TaskScheduler taskScheduler = new TaskScheduler(scheduler, between(10, 200));
+                taskScheduler.start();
+                safeAwait(taskScheduler.result);
+            }
+        } finally {
+            ThreadPool.terminate(testSubject, 1, TimeUnit.SECONDS);
+        }
+    }
+
+    private void waitUntilKeepAliveTime(long keepAliveNanos) {
+        var targetNanoTime = System.nanoTime() + keepAliveNanos + between(-1_000, 1_000);
+        while (System.nanoTime() < targetNanoTime) {
+            Thread.yield();
         }
     }
 }