From f7b758a7b063368745b710511e8a41a16b615e91 Mon Sep 17 00:00:00 2001 From: Jaroslav Bachorik Date: Fri, 14 Aug 2026 15:42:22 +0200 Subject: [PATCH] feat: add interruptible native matching --- .../runtime/InterruptibleCharSequence.java | 34 +++ .../runtime/LinearTokenSequenceMatcher.java | 48 +++++ .../reggie/runtime/ReggieCompiledPattern.java | 15 +- .../reggie/runtime/ReggieMatchState.java | 11 + .../InterruptibleCharSequenceTest.java | 196 ++++++++++++++++++ ...arTokenSequenceMatcherConcurrencyTest.java | 48 ++--- 6 files changed, 315 insertions(+), 37 deletions(-) create mode 100644 reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/InterruptibleCharSequence.java create mode 100644 reggie-runtime/src/test/java/com/datadoghq/reggie/runtime/InterruptibleCharSequenceTest.java diff --git a/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/InterruptibleCharSequence.java b/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/InterruptibleCharSequence.java new file mode 100644 index 00000000..6224e822 --- /dev/null +++ b/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/InterruptibleCharSequence.java @@ -0,0 +1,34 @@ +/* + * Copyright 2026-Present Datadog, Inc. + * + * 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 com.datadoghq.reggie.runtime; + +/** + * A character sequence that can cooperatively interrupt a native Reggie match. + * + *

{@link #checkInterrupted()} is invoked synchronously on the matching caller thread. The + * matching state ({@link ReggieMatchState}) is single-thread-confined and must not be shared across + * threads; the {@link InterruptibleCharSequence} itself may be freely reused across independent + * match calls. + */ +public interface InterruptibleCharSequence extends CharSequence { + /** + * Checks whether the current match should stop. + * + *

Implementations may throw an unchecked cancellation or deadline exception. A sequence is + * owned by its caller; a {@link ReggieMatchState} never retains it after {@code matches} returns. + */ + void checkInterrupted(); +} diff --git a/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/LinearTokenSequenceMatcher.java b/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/LinearTokenSequenceMatcher.java index 53a9f454..9f20f36e 100644 --- a/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/LinearTokenSequenceMatcher.java +++ b/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/LinearTokenSequenceMatcher.java @@ -167,6 +167,18 @@ boolean matchIntoBounded( return true; } + boolean matchIntoBoundedInterruptibly( + InterruptibleCharSequence input, + int start, + int end, + int[] groupStarts, + int[] groupEnds, + MatchWorkspace workspace) { + input.checkInterrupted(); + return matchIntoBounded( + new CheckpointingCharSequence(input), start, end, groupStarts, groupEnds, workspace); + } + private MatchResult toMatchResult(String input, MatchWorkspace workspace) { if (!nameToIndex.isEmpty()) { return new NamedMatchResultImpl( @@ -531,6 +543,42 @@ static final class MatchWorkspace { } } + private static final class CheckpointingCharSequence implements CharSequence { + private static final int CHECK_INTERVAL = 256; + + private final InterruptibleCharSequence delegate; + private int charactersUntilCheck = CHECK_INTERVAL; + + private CheckpointingCharSequence(InterruptibleCharSequence delegate) { + this.delegate = Objects.requireNonNull(delegate, "input"); + } + + @Override + public int length() { + return delegate.length(); + } + + @Override + public char charAt(int index) { + char value = delegate.charAt(index); + if (--charactersUntilCheck == 0) { + delegate.checkInterrupted(); + charactersUntilCheck = CHECK_INTERVAL; + } + return value; + } + + @Override + public CharSequence subSequence(int start, int end) { + throw new AssertionError("native matching must not materialize a subsequence"); + } + + @Override + public String toString() { + throw new AssertionError("native matching must not materialize input"); + } + } + private static int skipWhitespace(CharSequence input, int pos, int regionEnd) { int start = pos; while (pos < regionEnd && isJdkWhitespace(input.charAt(pos))) pos++; diff --git a/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/ReggieCompiledPattern.java b/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/ReggieCompiledPattern.java index a9617f05..7add4e70 100644 --- a/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/ReggieCompiledPattern.java +++ b/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/ReggieCompiledPattern.java @@ -40,23 +40,12 @@ public static ReggieCompilationResult tryCompile(ReggieCompileRequest request) { if (compilation.matcher() != null) { return ReggieCompilationResult.admitted(new ReggieCompiledPattern(compilation.matcher())); } - return ReggieCompilationResult.rejected(mapRejection(compilation.rejection())); + return ReggieCompilationResult.rejected( + ReggieCompilationRejection.valueOf(compilation.rejection().name())); } /** Creates a new single-thread-confined state object for matching this immutable pattern. */ public ReggieMatchState newState() { return new ReggieMatchState(matcher); } - - private static ReggieCompilationRejection mapRejection( - RuntimeCompiler.NamedOnlyLtsRejection rejection) { - return switch (rejection) { - case UNSUPPORTED_FLAGS -> ReggieCompilationRejection.UNSUPPORTED_FLAGS; - case SOURCE_INLINE_MODIFIER -> ReggieCompilationRejection.SOURCE_INLINE_MODIFIER; - case PARSE_FAILURE -> ReggieCompilationRejection.PARSE_FAILURE; - case PLAN_UNAVAILABLE -> ReggieCompilationRejection.PLAN_UNAVAILABLE; - case MISSING_NAMED_CAPTURE -> ReggieCompilationRejection.MISSING_NAMED_CAPTURE; - case PROFILE_INELIGIBLE -> ReggieCompilationRejection.PROFILE_INELIGIBLE; - }; - } } diff --git a/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/ReggieMatchState.java b/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/ReggieMatchState.java index 01a7ce16..f19b5f06 100644 --- a/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/ReggieMatchState.java +++ b/reggie-runtime/src/main/java/com/datadoghq/reggie/runtime/ReggieMatchState.java @@ -29,6 +29,7 @@ public final class ReggieMatchState { private final int[] starts; private final int[] ends; private boolean matched; + private boolean usedInterruptibleExecution; ReggieMatchState(LinearTokenSequenceMatcher matcher) { this.matcher = Objects.requireNonNull(matcher, "matcher"); @@ -45,9 +46,18 @@ public final class ReggieMatchState { */ public boolean matches(CharSequence input, int start, int end) { clear(); + if (input instanceof InterruptibleCharSequence interruptible) { + usedInterruptibleExecution = true; + return matched = + matcher.matchIntoBoundedInterruptibly(interruptible, start, end, starts, ends, workspace); + } return matched = matcher.matchIntoBounded(input, start, end, starts, ends, workspace); } + boolean usedInterruptibleExecution() { + return usedInterruptibleExecution; + } + public int start() { requireMatched(); return starts[0]; @@ -72,6 +82,7 @@ private void clear() { Arrays.fill(starts, -1); Arrays.fill(ends, -1); matched = false; + usedInterruptibleExecution = false; } private void requireMatched() { diff --git a/reggie-runtime/src/test/java/com/datadoghq/reggie/runtime/InterruptibleCharSequenceTest.java b/reggie-runtime/src/test/java/com/datadoghq/reggie/runtime/InterruptibleCharSequenceTest.java new file mode 100644 index 00000000..fa884fe9 --- /dev/null +++ b/reggie-runtime/src/test/java/com/datadoghq/reggie/runtime/InterruptibleCharSequenceTest.java @@ -0,0 +1,196 @@ +/* + * Copyright 2026-Present Datadog, Inc. + * + * 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 com.datadoghq.reggie.runtime; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; + +class InterruptibleCharSequenceTest { + + @Test + void checksBeforeInputValidationAndPropagatesTheOriginalException() { + ReggieMatchState state = stateFor("(?\\S+)"); + Cancellation cancellation = new Cancellation(); + InterruptibleCharSequence input = + new Probe("value", 1) { + @Override + public int length() { + if (checks == 0) throw new AssertionError("length accessed before interruption check"); + return super.length(); + } + + @Override + void check() { + throw cancellation; + } + }; + + assertSame(cancellation, assertThrows(Cancellation.class, () -> state.matches(input, -1, 5))); + assertThrows(IllegalStateException.class, state::start); + } + + @Test + void checksLongNativeScansAtMostEvery256CharactersAndClearsPriorState() { + ReggieMatchState state = stateFor("(?\\S+)"); + assertTrue(state.matches("prior", 0, 5)); + Probe input = new Probe("a".repeat(1024), 2); + + assertThrows(Cancellation.class, () -> state.matches(input, 0, input.length())); + assertTrue(input.maximumGap <= 256, () -> "maximum checkpoint gap: " + input.maximumGap); + assertThrows(IllegalStateException.class, () -> state.start("value")); + } + + @Test + void onlyInterruptibleInputsUseTheCheckpointPath() { + ReggieMatchState plainState = stateFor("(?\\S+)"); + assertTrue(plainState.matches("value", 0, 5)); + assertFalse(plainState.usedInterruptibleExecution()); + + ReggieMatchState interruptibleState = stateFor("(?\\S+)"); + Probe input = new Probe("value", Integer.MAX_VALUE); + assertTrue(interruptibleState.matches(input, 0, input.length())); + assertTrue(interruptibleState.usedInterruptibleExecution()); + assertEquals(0, interruptibleState.start("value")); + assertEquals(5, interruptibleState.end("value")); + } + + @Test + void checkpointsOptionalHttpAndIpOrHostValidationPaths() { + String target = "a".repeat(600); + ReggieMatchState optionalHttp = + stateFor("\"(?\\b\\w+\\b) (?\\S+)(?: HTTP/(?\\d+\\.\\d+)|)\""); + Probe optionalInput = new Probe("\"GET " + target + " HTTP/1.1\"", 4); + assertThrows( + Cancellation.class, () -> optionalHttp.matches(optionalInput, 0, optionalInput.length())); + assertTrue(optionalInput.maximumGap <= 256); + + Probe decimalInput = new Probe("\"GET " + target + " HTTP/" + "1".repeat(6000) + ".1\"", 60); + assertThrows( + Cancellation.class, () -> optionalHttp.matches(decimalInput, 0, decimalInput.length())); + assertTrue(decimalInput.maximumGap <= 256); + + ReggieMatchState ipOrHost = + stateFor("(?(?:[0-9]{1,3}\\.){3}[0-9]{1,3}|[A-Za-z0-9.-]+)"); + Probe hostInput = new Probe(target, 4); + assertThrows(Cancellation.class, () -> ipOrHost.matches(hostInput, 0, hostInput.length())); + assertTrue(hostInput.maximumGap <= 256); + } + + @Test + void checkpointsLiteralQuotedBracketTailAndTailConsumptionScans() { + String longValue = "a".repeat(600); + assertCancels("literal".repeat(100) + "(?\\S+)", "literal".repeat(100) + "x"); + assertCancels("ref=\"(?[^\"]*)\"", "ref=\"" + longValue + "\""); + assertCancels( + ".* \\[(?\\b\\w+\\b)\\] .*", + "prefix [" + longValue + "] tail", + ReggieCompileFlag.DOTALL); + assertCancels("(?\\S+) .*", "x " + longValue, ReggieCompileFlag.DOTALL); + } + + @Test + void nonCancellingInterruptibleInputMatchesPlainInputWithoutCacheMutation() { + int patterns = RuntimeCompiler.cacheSize(); + int structures = RuntimeCompiler.structuralCacheSize(); + ReggieMatchState plain = stateFor("host=(?\\S+)"); + ReggieMatchState interruptible = stateFor("host=(?\\S+)"); + String source = "xxhost=api"; + assertTrue(plain.matches(source, 2, source.length())); + Probe input = new Probe(source, Integer.MAX_VALUE); + assertTrue(interruptible.matches(input, 2, input.length())); + assertEquals(plain.start(), interruptible.start()); + assertEquals(plain.end(), interruptible.end()); + assertEquals(plain.start("host"), interruptible.start("host")); + assertEquals(plain.end("host"), interruptible.end("host")); + assertEquals(patterns, RuntimeCompiler.cacheSize()); + assertEquals(structures, RuntimeCompiler.structuralCacheSize()); + } + + private static void assertCancels(String pattern, String source) { + assertCancels(pattern, source, ReggieCompileFlag.NONE); + } + + private static void assertCancels(String pattern, String source, ReggieCompileFlag flag) { + ReggieMatchState state = stateFor(pattern, flag); + Probe input = new Probe(source, 2); + assertThrows(Cancellation.class, () -> state.matches(input, 0, input.length())); + assertTrue(input.maximumGap <= 256); + } + + private static ReggieMatchState stateFor(String source) { + return stateFor(source, ReggieCompileFlag.NONE); + } + + private static ReggieMatchState stateFor(String source, ReggieCompileFlag flag) { + ReggieCompilationResult result = + ReggieCompiledPattern.tryCompile(new ReggieCompileRequest(source, flag)); + assertTrue(result.isAdmitted()); + return result.pattern().newState(); + } + + private static class Probe implements InterruptibleCharSequence { + private final String value; + private final int throwOnCheck; + private int charactersSinceCheck; + protected int checks; + int maximumGap; + + Probe(String value, int throwOnCheck) { + this.value = value; + this.throwOnCheck = throwOnCheck; + } + + @Override + public int length() { + return value.length(); + } + + @Override + public char charAt(int index) { + charactersSinceCheck++; + return value.charAt(index); + } + + @Override + public CharSequence subSequence(int start, int end) { + throw new AssertionError("matching must not materialize a subsequence"); + } + + @Override + public String toString() { + throw new AssertionError("matching must not materialize input"); + } + + @Override + public final void checkInterrupted() { + checks++; + maximumGap = Math.max(maximumGap, charactersSinceCheck); + charactersSinceCheck = 0; + if (checks == throwOnCheck) check(); + } + + void check() { + throw new Cancellation(); + } + } + + private static final class Cancellation extends RuntimeException {} +} diff --git a/reggie-runtime/src/test/java/com/datadoghq/reggie/runtime/LinearTokenSequenceMatcherConcurrencyTest.java b/reggie-runtime/src/test/java/com/datadoghq/reggie/runtime/LinearTokenSequenceMatcherConcurrencyTest.java index 289ac6f2..5eddb02d 100644 --- a/reggie-runtime/src/test/java/com/datadoghq/reggie/runtime/LinearTokenSequenceMatcherConcurrencyTest.java +++ b/reggie-runtime/src/test/java/com/datadoghq/reggie/runtime/LinearTokenSequenceMatcherConcurrencyTest.java @@ -67,37 +67,37 @@ void cachedNamedOnlyLinearTokenSequenceMatcherIsSafeForConcurrentUse() throws Ex int threads = 16; int iterations = 1_000; + ExecutorService executor = Executors.newFixedThreadPool(threads); CountDownLatch ready = new CountDownLatch(threads); CountDownLatch start = new CountDownLatch(1); CountDownLatch done = new CountDownLatch(threads); ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); - try (ExecutorService executor = Executors.newFixedThreadPool(threads)) { - for (int thread = 0; thread < threads; thread++) { - executor.execute( - () -> { - ready.countDown(); - try { - start.await(); - for (int iteration = 0; iteration < iterations; iteration++) { - assertMatchWithOptionalCaptures(shared, groupNumbers); - assertMatchWithoutOptionalCaptures(shared, groupNumbers); - assertFailedMatchLeavesArraysUntouched(shared, groupCount); - assertTrue(shared.find("noise " + WITH_OPTIONAL_CAPTURES)); - assertFalse(shared.find("noise malformed access log")); - } - } catch (Throwable failure) { - failures.add(failure); - } finally { - done.countDown(); + for (int thread = 0; thread < threads; thread++) { + executor.execute( + () -> { + ready.countDown(); + try { + start.await(); + for (int iteration = 0; iteration < iterations; iteration++) { + assertMatchWithOptionalCaptures(shared, groupNumbers); + assertMatchWithoutOptionalCaptures(shared, groupNumbers); + assertFailedMatchLeavesArraysUntouched(shared, groupCount); + assertTrue(shared.find("noise " + WITH_OPTIONAL_CAPTURES)); + assertFalse(shared.find("noise malformed access log")); } - }); - } - - assertTrue(ready.await(10, TimeUnit.SECONDS), "workers did not become ready"); - start.countDown(); - assertTrue(done.await(30, TimeUnit.SECONDS), "workers did not finish"); + } catch (Throwable failure) { + failures.add(failure); + } finally { + done.countDown(); + } + }); } + + assertTrue(ready.await(10, TimeUnit.SECONDS), "workers did not become ready"); + start.countDown(); + assertTrue(done.await(30, TimeUnit.SECONDS), "workers did not finish"); + executor.shutdownNow(); assertTrue(failures.isEmpty(), () -> "concurrent LTS failure: " + failures.peek()); }