From b199e0a5b384ebdcf578f000651788694069dd6a Mon Sep 17 00:00:00 2001 From: Markus Frohme Date: Thu, 9 Jul 2026 17:12:33 +0200 Subject: [PATCH 1/5] initial experimentation --- examples/pom.xml | 10 ++ .../example/reactive/JavaRXExample.java | 93 ++++++++++++++++++ .../example/reactive/ReactorExample.java | 95 +++++++++++++++++++ examples/src/main/java/module-info.java | 4 + 4 files changed, 202 insertions(+) create mode 100644 examples/src/main/java/de/learnlib/example/reactive/JavaRXExample.java create mode 100644 examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java diff --git a/examples/pom.xml b/examples/pom.xml index 4781159c3..2cab0b9ad 100644 --- a/examples/pom.xml +++ b/examples/pom.xml @@ -141,6 +141,16 @@ limitations under the License. runtime + + io.projectreactor + reactor-core + + + io.reactivex.rxjava3 + rxjava + 3.1.12 + + net.automatalib automata-api diff --git a/examples/src/main/java/de/learnlib/example/reactive/JavaRXExample.java b/examples/src/main/java/de/learnlib/example/reactive/JavaRXExample.java new file mode 100644 index 000000000..ff3ca7463 --- /dev/null +++ b/examples/src/main/java/de/learnlib/example/reactive/JavaRXExample.java @@ -0,0 +1,93 @@ +package de.learnlib.example.reactive; + +import java.util.Objects; +import java.util.Random; + +import de.learnlib.algorithm.LearningAlgorithm; +import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy; +import de.learnlib.driver.simulator.MealySimulatorSUL; +import de.learnlib.oracle.MembershipOracle; +import de.learnlib.oracle.equivalence.KWayStateCoverEQOracle; +import de.learnlib.oracle.equivalence.RandomWMethodEQOracle; +import de.learnlib.oracle.parallelism.ParallelOracleBuilders; +import de.learnlib.query.DefaultQuery; +import io.reactivex.rxjava3.core.Flowable; +import io.reactivex.rxjava3.functions.Action; +import io.reactivex.rxjava3.functions.Consumer; +import io.reactivex.rxjava3.schedulers.Schedulers; +import net.automatalib.alphabet.impl.Alphabets; +import net.automatalib.util.automaton.Automata; +import net.automatalib.util.automaton.conformance.KWayStateCoverTestsIterator.CombinationMethod; +import net.automatalib.util.automaton.random.RandomAutomata; +import net.automatalib.word.Word; + +public class JavaRXExample { + + public static void main(String[] args) { + // setup symbols + var inputs = Alphabets.integers(0, 4); + var outputs = Alphabets.characters('a', 'd'); + + // setup membership oracle + var mealy = RandomAutomata.randomMealy(new Random(42), 10, inputs, outputs); + var sul = new MealySimulatorSUL<>(mealy); + // make sure to use parallel-aware oracle (with thread local instances) + // because it will be called from different threads in the reactive environment + var mqo = ParallelOracleBuilders.newDynamicParallelOracle(sul).create(); + + // setup equivalence oracles + var eqo = new RandomWMethodEQOracle<>(mqo, 2, 4); + var eqo2 = new KWayStateCoverEQOracle<>(mqo, new Random(42), 2, 4, CombinationMethod.COMBINATIONS, 2); + + // setup learner + var learner = new TTTLearnerMealy<>(inputs, mqo); + var tracker = new ProgressTracker<>(learner, mqo); + + // learning loop + learner.startLearning(); + var hyp = learner.getHypothesisModel(); + + while (!tracker.hasFinished()) { + final var finalHyp = hyp; + Flowable.fromStream(eqo.generateTestWords(finalHyp, inputs)) + .take(100) + .mergeWith(Flowable.fromStream(eqo2.generateTestWords(finalHyp, inputs))) + .parallel() + .runOn(Schedulers.computation()) + .filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w))) + .sequential() + .firstElement() + .subscribe(tracker, System.out::println, tracker); + } + + hyp = learner.getHypothesisModel(); + + System.out.println(Automata.testEquivalence(mealy, hyp, inputs)); + } + + private static class ProgressTracker implements Consumer>, Action { + + private boolean finished; + private final LearningAlgorithm learner; + private final MembershipOracle oracle; + + private ProgressTracker(LearningAlgorithm learner, MembershipOracle oracle) { + this.learner = learner; + this.oracle = oracle; + } + + @Override + public void accept(Word w) { + this.finished = learner.refineHypothesis(new DefaultQuery<>(w, oracle.answerQuery(w))); + } + + public boolean hasFinished() { + return finished; + } + + @Override + public void run() throws Throwable { + this.finished = true; + } + } +} diff --git a/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java b/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java new file mode 100644 index 000000000..d54e2c321 --- /dev/null +++ b/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java @@ -0,0 +1,95 @@ +package de.learnlib.example.reactive; + +import java.util.Objects; +import java.util.Random; +import java.util.function.Consumer; + +import de.learnlib.algorithm.LearningAlgorithm; +import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy; +import de.learnlib.driver.simulator.MealySimulatorSUL; +import de.learnlib.oracle.MembershipOracle; +import de.learnlib.oracle.equivalence.KWayStateCoverEQOracle; +import de.learnlib.oracle.equivalence.RandomWMethodEQOracle; +import de.learnlib.oracle.equivalence.SimulatorEQOracle; +import de.learnlib.oracle.parallelism.ParallelOracleBuilders; +import de.learnlib.query.DefaultQuery; +import net.automatalib.alphabet.impl.Alphabets; +import net.automatalib.util.automaton.Automata; +import net.automatalib.util.automaton.conformance.KWayStateCoverTestsIterator.CombinationMethod; +import net.automatalib.util.automaton.random.RandomAutomata; +import net.automatalib.word.Word; +import reactor.core.publisher.Flux; +import reactor.core.scheduler.Schedulers; + +public class ReactorExample { + + public static void main(String[] args) { + // setup symbols + var inputs = Alphabets.integers(0, 4); + var outputs = Alphabets.characters('a', 'd'); + + // setup membership oracle + var mealy = RandomAutomata.randomMealy(new Random(42), 10, inputs, outputs); + var sul = new MealySimulatorSUL<>(mealy); + // make sure to use a parallel-aware oracle (with thread local instances) + // because it will be called from different threads in the reactive environment + var mqo = ParallelOracleBuilders.newDynamicParallelOracle(sul).create(); + + // setup equivalence oracles + var eqo = new RandomWMethodEQOracle<>(mqo, 2, 4); + var eqo2 = new KWayStateCoverEQOracle<>(mqo, new Random(42), 2, 4, CombinationMethod.COMBINATIONS, 2); + + // setup learner + var learner = new TTTLearnerMealy<>(inputs, mqo); + + // learning loop + learner.startLearning(); + var hyp = learner.getHypothesisModel(); + var tracker = new ProgressTracker<>(learner, mqo); + + while (!tracker.hasFinished()) { + final var finalHyp = hyp; + Flux.fromStream(eqo.generateTestWords(finalHyp, inputs)) + .take(1000) + .mergeWith(Flux.fromStream(eqo2.generateTestWords(finalHyp, inputs))) + .parallel() + .runOn(Schedulers.parallel()) + .filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w))) + .sequential() + .next() + .subscribe(tracker, System.out::println, tracker) + .dispose(); + hyp = learner.getHypothesisModel(); + } + + hyp = learner.getHypothesisModel(); + + System.out.println(Automata.testEquivalence(mealy, hyp, inputs)); + } + + private static class ProgressTracker implements Consumer>, Runnable { + + private boolean finished; + private final LearningAlgorithm learner; + private final MembershipOracle oracle; + + private ProgressTracker(LearningAlgorithm learner, MembershipOracle oracle) { + this.learner = learner; + this.oracle = oracle; + } + + @Override + public void accept(Word w) { + this.finished = learner.refineHypothesis(new DefaultQuery<>(w, oracle.answerQuery(w))); + } + + public boolean hasFinished() { + return finished; + } + + @Override + public void run() { + this.finished = true; + } + } +} diff --git a/examples/src/main/java/module-info.java b/examples/src/main/java/module-info.java index 7b4b99096..cce24b171 100644 --- a/examples/src/main/java/module-info.java +++ b/examples/src/main/java/module-info.java @@ -56,12 +56,16 @@ // annotations are 'provided'-scoped and do not need to be loaded at runtime requires static org.checkerframework.checker.qual; + requires reactor.core; + requires org.reactivestreams; + requires io.reactivex.rxjava3; exports de.learnlib.example; exports de.learnlib.example.aaar; exports de.learnlib.example.bbc; exports de.learnlib.example.parallelism; exports de.learnlib.example.passive; + exports de.learnlib.example.reactive; exports de.learnlib.example.resumable; exports de.learnlib.example.sli; } From 524d16f4bab809bf6662bef36890264d8eb2e8cd Mon Sep 17 00:00:00 2001 From: Markus Frohme Date: Tue, 14 Jul 2026 01:41:21 +0200 Subject: [PATCH 2/5] some more progress, Project Reactor still borked --- examples/pom.xml | 15 +- .../example/reactive/JavaRXExample.java | 93 -------- .../example/reactive/RXJavaExample.java | 158 ++++++++++++ .../example/reactive/ReactorExample.java | 225 +++++++++++++++--- .../example/reactive/SmallRyeExample.java | 121 ++++++++++ examples/src/main/java/module-info.java | 7 +- .../de/learnlib/example/ExamplesTest.java | 10 + pom.xml | 20 ++ 8 files changed, 511 insertions(+), 138 deletions(-) delete mode 100644 examples/src/main/java/de/learnlib/example/reactive/JavaRXExample.java create mode 100644 examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java create mode 100644 examples/src/main/java/de/learnlib/example/reactive/SmallRyeExample.java diff --git a/examples/pom.xml b/examples/pom.xml index 2cab0b9ad..c005dc56d 100644 --- a/examples/pom.xml +++ b/examples/pom.xml @@ -146,11 +146,20 @@ limitations under the License. reactor-core - io.reactivex.rxjava3 - rxjava - 3.1.12 + io.reactivex.rxjava3 + rxjava + + + io.smallrye.reactive + mutiny + 3.3.0 + + org.reactivestreams + reactive-streams + + net.automatalib automata-api diff --git a/examples/src/main/java/de/learnlib/example/reactive/JavaRXExample.java b/examples/src/main/java/de/learnlib/example/reactive/JavaRXExample.java deleted file mode 100644 index ff3ca7463..000000000 --- a/examples/src/main/java/de/learnlib/example/reactive/JavaRXExample.java +++ /dev/null @@ -1,93 +0,0 @@ -package de.learnlib.example.reactive; - -import java.util.Objects; -import java.util.Random; - -import de.learnlib.algorithm.LearningAlgorithm; -import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy; -import de.learnlib.driver.simulator.MealySimulatorSUL; -import de.learnlib.oracle.MembershipOracle; -import de.learnlib.oracle.equivalence.KWayStateCoverEQOracle; -import de.learnlib.oracle.equivalence.RandomWMethodEQOracle; -import de.learnlib.oracle.parallelism.ParallelOracleBuilders; -import de.learnlib.query.DefaultQuery; -import io.reactivex.rxjava3.core.Flowable; -import io.reactivex.rxjava3.functions.Action; -import io.reactivex.rxjava3.functions.Consumer; -import io.reactivex.rxjava3.schedulers.Schedulers; -import net.automatalib.alphabet.impl.Alphabets; -import net.automatalib.util.automaton.Automata; -import net.automatalib.util.automaton.conformance.KWayStateCoverTestsIterator.CombinationMethod; -import net.automatalib.util.automaton.random.RandomAutomata; -import net.automatalib.word.Word; - -public class JavaRXExample { - - public static void main(String[] args) { - // setup symbols - var inputs = Alphabets.integers(0, 4); - var outputs = Alphabets.characters('a', 'd'); - - // setup membership oracle - var mealy = RandomAutomata.randomMealy(new Random(42), 10, inputs, outputs); - var sul = new MealySimulatorSUL<>(mealy); - // make sure to use parallel-aware oracle (with thread local instances) - // because it will be called from different threads in the reactive environment - var mqo = ParallelOracleBuilders.newDynamicParallelOracle(sul).create(); - - // setup equivalence oracles - var eqo = new RandomWMethodEQOracle<>(mqo, 2, 4); - var eqo2 = new KWayStateCoverEQOracle<>(mqo, new Random(42), 2, 4, CombinationMethod.COMBINATIONS, 2); - - // setup learner - var learner = new TTTLearnerMealy<>(inputs, mqo); - var tracker = new ProgressTracker<>(learner, mqo); - - // learning loop - learner.startLearning(); - var hyp = learner.getHypothesisModel(); - - while (!tracker.hasFinished()) { - final var finalHyp = hyp; - Flowable.fromStream(eqo.generateTestWords(finalHyp, inputs)) - .take(100) - .mergeWith(Flowable.fromStream(eqo2.generateTestWords(finalHyp, inputs))) - .parallel() - .runOn(Schedulers.computation()) - .filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w))) - .sequential() - .firstElement() - .subscribe(tracker, System.out::println, tracker); - } - - hyp = learner.getHypothesisModel(); - - System.out.println(Automata.testEquivalence(mealy, hyp, inputs)); - } - - private static class ProgressTracker implements Consumer>, Action { - - private boolean finished; - private final LearningAlgorithm learner; - private final MembershipOracle oracle; - - private ProgressTracker(LearningAlgorithm learner, MembershipOracle oracle) { - this.learner = learner; - this.oracle = oracle; - } - - @Override - public void accept(Word w) { - this.finished = learner.refineHypothesis(new DefaultQuery<>(w, oracle.answerQuery(w))); - } - - public boolean hasFinished() { - return finished; - } - - @Override - public void run() throws Throwable { - this.finished = true; - } - } -} diff --git a/examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java b/examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java new file mode 100644 index 000000000..1c84b477f --- /dev/null +++ b/examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java @@ -0,0 +1,158 @@ +/* Copyright (C) 2013-2026 TU Dortmund University + * This file is part of LearnLib . + * + * 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 de.learnlib.example.reactive; + +import java.util.Objects; +import java.util.Random; +import java.util.concurrent.Executors; + +import de.learnlib.algorithm.LearningAlgorithm; +import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy; +import de.learnlib.driver.simulator.MealySimulatorSUL; +import de.learnlib.oracle.MembershipOracle; +import de.learnlib.oracle.equivalence.KWayStateCoverEQOracleBuilder; +import de.learnlib.oracle.equivalence.RandomWMethodEQOracle; +import de.learnlib.oracle.parallelism.ParallelOracleBuilders; +import de.learnlib.query.DefaultQuery; +import io.reactivex.rxjava3.core.Flowable; +import io.reactivex.rxjava3.functions.Action; +import io.reactivex.rxjava3.functions.Consumer; +import io.reactivex.rxjava3.schedulers.Schedulers; +import net.automatalib.alphabet.impl.Alphabets; +import net.automatalib.automaton.transducer.MealyMachine; +import net.automatalib.util.automaton.Automata; +import net.automatalib.util.automaton.random.RandomAutomata; +import net.automatalib.word.Word; + +/** + * An example of constructing a learn-loop using reactive streams (from RXJava) to compute counterexamples. + */ +@SuppressWarnings({"PMD.SystemPrintln", "PMD.UseExplicitTypes"}) // allow println and vars in examples +public final class RXJavaExample { + + private static final int SEED = 42; + private static final int SIZE = 10; + private static final int NUM_INPUTS = 4; + private static final int RND_LENGTH = 4; + private static final int LIMIT = 1000; + + private RXJavaExample() { + // prevent instantiation + } + + public static void main(String[] args) { + // setup symbols + var inputs = Alphabets.integers(0, NUM_INPUTS); + var outputs = Alphabets.characters('a', 'd'); + + // setup membership oracle + var mealy = RandomAutomata.randomMealy(new Random(SEED), SIZE, inputs, outputs); + var sul = new MealySimulatorSUL<>(mealy); + // IMPORTANT: make sure to use parallel-aware oracle (with thread local instances) + // because it will be called from different threads in the reactive environment + var mqo = ParallelOracleBuilders.newDynamicParallelOracle(sul).create(); + + // setup equivalence oracles + var eqo = new RandomWMethodEQOracle<>(mqo, SIZE / 2, RND_LENGTH); + var eqo2 = + new KWayStateCoverEQOracleBuilder, Integer, Word>().withOracle( + mqo).withRandom(new Random(SEED)).create(); + + // setup learner + var learner = new TTTLearnerMealy<>(inputs, mqo); + var tracker = new ProgressTracker<>(learner, mqo); + + // setup scheduler + // IMPORTANT: set interruptibleWorker to false in order to allow graceful oracle/SUL shutdown + var pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); + var scheduler = Schedulers.from(pool, false, true); + + // learning loop + learner.startLearning(); + var hyp = learner.getHypothesisModel(); + + while (!tracker.hasFinished()) { + // since we only access the hypothesis in read-only fashion it is fine to share the reference across threads + final var finalHyp = hyp; + Flowable // create a publisher from first quivalence oracle + .fromStream(eqo.generateTestWords(finalHyp, inputs)) + // limit to 1000 elements + .take(LIMIT) + // merge with elements from second equivalence oracle + .mergeWith(Flowable.fromStream(eqo2.generateTestWords(finalHyp, inputs))) + // process items in parallel + .parallel() + // run on our non-interruptible scheduler + .runOn(scheduler) + // filter for counterexamples + .filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w))) + // process them sequentially + .sequential() + // the first counterexample suffices for refinement + .firstElement() + // use a blocking subscribe to ensure that all pipelines are cleared for the next iteration + // alternatively, use blockingGet like in the ReactorExample + .blockingSubscribe(tracker, System.out::println, tracker); + } + + // cleanup + pool.shutdown(); + mqo.shutdown(); + scheduler.shutdown(); + + // process results + hyp = learner.getHypothesisModel(); + + System.out.println("Final hypothesis size " + hyp.size()); + System.out.println("Is equivalent? " + Automata.testEquivalence(mealy, hyp, inputs)); + } + + private static final class ProgressTracker implements Consumer>, Action { + + private boolean finished; + private final LearningAlgorithm learner; + private final MembershipOracle oracle; + + private ProgressTracker(LearningAlgorithm learner, MembershipOracle oracle) { + this.learner = learner; + this.oracle = oracle; + } + + /** + * The onSuccess handler. Upon receiving a counterexample, refine the hypothesis. + * + * @param w + * the counterexample + */ + @Override + public void accept(Word w) { + learner.refineHypothesis(new DefaultQuery<>(w, oracle.answerQuery(w))); + } + + /** + * The onComplete handler. When all items have been processed without finding a counterexample, terminate the + * learning loop. + */ + @Override + public void run() { + this.finished = true; + } + + private boolean hasFinished() { + return finished; + } + } +} diff --git a/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java b/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java index d54e2c321..f7a38f708 100644 --- a/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java +++ b/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java @@ -1,95 +1,242 @@ +/* Copyright (C) 2013-2026 TU Dortmund University + * This file is part of LearnLib . + * + * 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 de.learnlib.example.reactive; import java.util.Objects; import java.util.Random; -import java.util.function.Consumer; +import java.util.concurrent.Delayed; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; +import java.util.concurrent.RunnableFuture; +import java.util.concurrent.RunnableScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; -import de.learnlib.algorithm.LearningAlgorithm; import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy; import de.learnlib.driver.simulator.MealySimulatorSUL; -import de.learnlib.oracle.MembershipOracle; -import de.learnlib.oracle.equivalence.KWayStateCoverEQOracle; +import de.learnlib.oracle.equivalence.KWayStateCoverEQOracleBuilder; import de.learnlib.oracle.equivalence.RandomWMethodEQOracle; -import de.learnlib.oracle.equivalence.SimulatorEQOracle; import de.learnlib.oracle.parallelism.ParallelOracleBuilders; import de.learnlib.query.DefaultQuery; import net.automatalib.alphabet.impl.Alphabets; +import net.automatalib.automaton.transducer.MealyMachine; import net.automatalib.util.automaton.Automata; -import net.automatalib.util.automaton.conformance.KWayStateCoverTestsIterator.CombinationMethod; import net.automatalib.util.automaton.random.RandomAutomata; import net.automatalib.word.Word; +import reactor.core.Disposable; import reactor.core.publisher.Flux; +import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; -public class ReactorExample { +/** + * An example of constructing a learn-loop using reactive streams (from Project Reactor) to compute counterexamples. + */ +@SuppressWarnings({"PMD.SystemPrintln", "PMD.UseExplicitTypes"}) // allow println and vars in examples +public final class ReactorExample { + + private static final int SEED = 42; + private static final int SIZE = 10; + private static final int NUM_INPUTS = 4; + private static final int RND_LENGTH = 4; + private static final int LIMIT = 1000; + + private ReactorExample() { + // prevent instantiation + } public static void main(String[] args) { // setup symbols - var inputs = Alphabets.integers(0, 4); + var inputs = Alphabets.integers(0, NUM_INPUTS); var outputs = Alphabets.characters('a', 'd'); // setup membership oracle - var mealy = RandomAutomata.randomMealy(new Random(42), 10, inputs, outputs); + var mealy = RandomAutomata.randomMealy(new Random(SEED), SIZE, inputs, outputs); var sul = new MealySimulatorSUL<>(mealy); - // make sure to use a parallel-aware oracle (with thread local instances) + // IMPORTANT: make sure to use parallel-aware oracle (with thread local instances) // because it will be called from different threads in the reactive environment var mqo = ParallelOracleBuilders.newDynamicParallelOracle(sul).create(); // setup equivalence oracles - var eqo = new RandomWMethodEQOracle<>(mqo, 2, 4); - var eqo2 = new KWayStateCoverEQOracle<>(mqo, new Random(42), 2, 4, CombinationMethod.COMBINATIONS, 2); + var eqo = new RandomWMethodEQOracle<>(mqo, SIZE / 2, RND_LENGTH); + var eqo2 = + new KWayStateCoverEQOracleBuilder, Integer, Word>().withOracle( + mqo).withRandom(new Random(SEED)).create(); // setup learner var learner = new TTTLearnerMealy<>(inputs, mqo); + // setup scheduler + // IMPORTANT: set interruptibleWorker to false in order to allow graceful oracle/SUL shutdown + var pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); +// var scheduler = new PureNonInterruptingScheduler(Schedulers.parallel()); + // Decorate all BoundedElastic schedulers (or any standard pools) + Schedulers.addExecutorServiceDecorator("prevent-interrupts", (sched, executorService) -> + new ScheduledThreadPoolExecutor(1) { // Or wrap the existing executorService + @Override + protected RunnableScheduledFuture decorateTask( + Runnable runnable, RunnableScheduledFuture task) { + return new CustomScheduledFuture<>(task); + } + } + ); + // learning loop learner.startLearning(); var hyp = learner.getHypothesisModel(); - var tracker = new ProgressTracker<>(learner, mqo); - while (!tracker.hasFinished()) { + while (true) { final var finalHyp = hyp; - Flux.fromStream(eqo.generateTestWords(finalHyp, inputs)) - .take(1000) - .mergeWith(Flux.fromStream(eqo2.generateTestWords(finalHyp, inputs))) - .parallel() - .runOn(Schedulers.parallel()) - .filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w))) - .sequential() - .next() - .subscribe(tracker, System.out::println, tracker) - .dispose(); - hyp = learner.getHypothesisModel(); + var ce = Flux.fromStream(eqo.generateTestWords(finalHyp, inputs)) + .take(LIMIT) + .mergeWith(Flux.fromStream(eqo2.generateTestWords(finalHyp, inputs))) + .parallel() + .runOn(Schedulers.parallel()) + .filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w))) + .sequential() + .next() + .block(); + + if (ce != null) { + learner.refineHypothesis(new DefaultQuery<>(ce, mqo.answerQuery(ce))); + hyp = learner.getHypothesisModel(); + } else { + break; + } } + // cleanup + pool.shutdown(); + mqo.shutdown(); +// scheduler.dispose(); + + // process results hyp = learner.getHypothesisModel(); - System.out.println(Automata.testEquivalence(mealy, hyp, inputs)); + System.out.println("Final hypothesis size " + hyp.size()); + System.out.println("Is equivalent? " + Automata.testEquivalence(mealy, hyp, inputs)); } - private static class ProgressTracker implements Consumer>, Runnable { + private static class NonInterruptingScheduler implements Scheduler { - private boolean finished; - private final LearningAlgorithm learner; - private final MembershipOracle oracle; + private final Scheduler delegate; - private ProgressTracker(LearningAlgorithm learner, MembershipOracle oracle) { - this.learner = learner; - this.oracle = oracle; + public NonInterruptingScheduler(Scheduler delegate) { + this.delegate = delegate; } @Override - public void accept(Word w) { - this.finished = learner.refineHypothesis(new DefaultQuery<>(w, oracle.answerQuery(w))); + public Disposable schedule(Runnable task) { + // Wrap the task to clear or ignore the interrupted status if something upstream triggers it + Runnable safeTask = () -> { + try { + task.run(); + } catch (Exception e) { + // Handle or catch unexpected InterruptedExceptions + } finally { + // Clear the interrupted status flag just in case + Thread.interrupted(); + } + }; + return delegate.schedule(safeTask); } - public boolean hasFinished() { - return finished; + @Override + public Worker createWorker() { + return delegate.createWorker(); + } + + @Override + public void dispose() { + delegate.dispose(); + } + } + + public static class PureNonInterruptingScheduler implements Scheduler { + + private final Scheduler delegate; + private final ExecutorService executor = Executors.newFixedThreadPool(10); + + public PureNonInterruptingScheduler(Scheduler delegate) { + this.delegate = delegate; + } + + @Override + public Disposable schedule(Runnable task) { + return delegate.schedule(task); } @Override - public void run() { - this.finished = true; + public Worker createWorker() { + return new NonInterruptingWorker(executor); } + + @Override + public void dispose() { + executor.shutdown(); + delegate.dispose(); + } + + private static class NonInterruptingWorker implements Worker { + + private final ExecutorService executor; + + public NonInterruptingWorker(ExecutorService executor) { + this.executor = executor; + } + + @Override + public Disposable schedule(Runnable task) { + Future future = executor.submit(task); + return () -> future.cancel(false); // Overrides thread interruption logic + } + + @Override + public void dispose() { + // Keep empty if you do not want worker disposal to break the underlying pool + } + } + } + // A delegate wrapper that ignores the "mayInterruptIfRunning" flag + private static class CustomScheduledFuture implements RunnableScheduledFuture { + private final RunnableScheduledFuture delegate; + + public CustomScheduledFuture(RunnableScheduledFuture delegate) { + this.delegate = delegate; + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + // FORCE false to prevent Thread.interrupt() + return delegate.cancel(false); + } + + // Delegate all other standard interface methods... + @Override public boolean isCancelled() { return delegate.isCancelled(); } + @Override public boolean isDone() { return delegate.isDone(); } + @Override public V get() throws InterruptedException, ExecutionException { return delegate.get(); } + @Override public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, + TimeoutException { return delegate.get(timeout, unit); } + @Override public long getDelay(TimeUnit unit) { return delegate.getDelay(unit); } + @Override public int compareTo(Delayed o) { return delegate.compareTo(o); } + @Override public void run() { delegate.run(); } + @Override public boolean isPeriodic() { return delegate.isPeriodic(); } } } diff --git a/examples/src/main/java/de/learnlib/example/reactive/SmallRyeExample.java b/examples/src/main/java/de/learnlib/example/reactive/SmallRyeExample.java new file mode 100644 index 000000000..ac618afa2 --- /dev/null +++ b/examples/src/main/java/de/learnlib/example/reactive/SmallRyeExample.java @@ -0,0 +1,121 @@ +/* Copyright (C) 2013-2026 TU Dortmund University + * This file is part of LearnLib . + * + * 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 de.learnlib.example.reactive; + +import java.util.Objects; +import java.util.Random; +import java.util.concurrent.Executors; + +import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy; +import de.learnlib.driver.simulator.MealySimulatorSUL; +import de.learnlib.oracle.equivalence.KWayStateCoverEQOracleBuilder; +import de.learnlib.oracle.equivalence.RandomWMethodEQOracle; +import de.learnlib.oracle.parallelism.ParallelOracleBuilders; +import de.learnlib.query.DefaultQuery; +import io.smallrye.mutiny.Multi; +import net.automatalib.alphabet.impl.Alphabets; +import net.automatalib.automaton.transducer.MealyMachine; +import net.automatalib.util.automaton.Automata; +import net.automatalib.util.automaton.random.RandomAutomata; +import net.automatalib.word.Word; + +/** + * An example of constructing a learn-loop using reactive streams (from SmallRye) to compute counterexamples. + */ +@SuppressWarnings({"PMD.SystemPrintln", "PMD.UseExplicitTypes"}) // allow println and vars in examples +public final class SmallRyeExample { + + private static final int SEED = 42; + private static final int SIZE = 10; + private static final int NUM_INPUTS = 4; + private static final int RND_LENGTH = 4; + private static final int LIMIT = 1000; + + private SmallRyeExample() { + // prevent instantiation + } + + public static void main(String[] args) { + // setup symbols + var inputs = Alphabets.integers(0, NUM_INPUTS); + var outputs = Alphabets.characters('a', 'd'); + + // setup membership oracle + var mealy = RandomAutomata.randomMealy(new Random(SEED), SIZE, inputs, outputs); + var sul = new MealySimulatorSUL<>(mealy); + // IMPORTANT: make sure to use parallel-aware oracle (with thread local instances) + // because it will be called from different threads in the reactive environment + var mqo = ParallelOracleBuilders.newDynamicParallelOracle(sul).create(); + + // setup equivalence oracles + var eqo = new RandomWMethodEQOracle<>(mqo, SIZE / 2, RND_LENGTH); + var eqo2 = + new KWayStateCoverEQOracleBuilder, Integer, Word>().withOracle( + mqo).withRandom(new Random(SEED)).create(); + + // setup learner + var learner = new TTTLearnerMealy<>(inputs, mqo); + + // setup thread pool + var pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); + + // learning loop + learner.startLearning(); + var hyp = learner.getHypothesisModel(); + + while (true) { + // since we only access the hypothesis in read-only fashion it is fine to share the reference across threads + final var finalHyp = hyp; + var m1 = Multi.createFrom() + // create a publisher from first quivalence oracle + .items(eqo.generateTestWords(finalHyp, inputs)) + // limit to 1000 elements + .select().first(LIMIT); + var m2 = Multi.createFrom() + // create a publisher from second quivalence oracle + .items(eqo2.generateTestWords(finalHyp, inputs)); + + var ce = Multi.createBy() + // merge elements from both oracles in interleaving fashion + .merging().streams(m1, m2) + // run processing in parallel + .runSubscriptionOn(pool) + // filter for counterexamples + .filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w))) + // toUni implicitly fetches the first element + .toUni() + // use a blocking call to extract the final counterexample + .await().indefinitely(); + + if (ce != null) { + learner.refineHypothesis(new DefaultQuery<>(ce, mqo.answerQuery(ce))); + hyp = learner.getHypothesisModel(); + } else { + break; + } + } + + // cleanup + mqo.shutdown(); + pool.shutdown(); + + // process results + hyp = learner.getHypothesisModel(); + + System.out.println("Final hypothesis size " + hyp.size()); + System.out.println("Is equivalent? " + Automata.testEquivalence(mealy, hyp, inputs)); + } +} diff --git a/examples/src/main/java/module-info.java b/examples/src/main/java/module-info.java index cce24b171..e7c8cc0a3 100644 --- a/examples/src/main/java/module-info.java +++ b/examples/src/main/java/module-info.java @@ -44,6 +44,8 @@ requires de.learnlib.oracle.parallelism; requires de.learnlib.oracle.property; requires de.learnlib.testsupport.example; + requires io.reactivex.rxjava3; + requires io.smallrye.mutiny; requires net.automatalib.api; requires net.automatalib.common.util; requires net.automatalib.core; @@ -53,12 +55,11 @@ requires net.automatalib.serialization.dot; requires net.automatalib.visualization.dot; requires org.apache.fury.core; + requires org.reactivestreams; + requires reactor.core; // annotations are 'provided'-scoped and do not need to be loaded at runtime requires static org.checkerframework.checker.qual; - requires reactor.core; - requires org.reactivestreams; - requires io.reactivex.rxjava3; exports de.learnlib.example; exports de.learnlib.example.aaar; diff --git a/examples/src/test/java/de/learnlib/example/ExamplesTest.java b/examples/src/test/java/de/learnlib/example/ExamplesTest.java index 7e1e47e76..993214313 100644 --- a/examples/src/test/java/de/learnlib/example/ExamplesTest.java +++ b/examples/src/test/java/de/learnlib/example/ExamplesTest.java @@ -165,6 +165,16 @@ public void testPassiveExample1() { de.learnlib.example.passive.Example1.main(new String[0]); } + @Test + public void testReactiveReactorExample() { + de.learnlib.example.reactive.ReactorExample.main(new String[0]); + } + + @Test + public void testReactiveRXJavaExample() { + de.learnlib.example.reactive.RXJavaExample.main(new String[0]); + } + @Test public void testResumableExample() { de.learnlib.example.resumable.ResumableExample.main(new String[0]); diff --git a/pom.xml b/pom.xml index c52da4a5b..95da30dd5 100644 --- a/pom.xml +++ b/pom.xml @@ -280,6 +280,9 @@ limitations under the License. 1.11 5.20.0 7.22.0 + 1.0.4 + 3.8.6 + 3.1.12 2.0.16 7.10.2 @@ -668,6 +671,23 @@ limitations under the License. ${logback.version} + + + io.projectreactor + reactor-core + ${reactor.version} + + + io.reactivex.rxjava3 + rxjava + ${rxjava.version} + + + org.reactivestreams + reactive-streams + ${reactive-streams.version} + + From 4ada884aa8b87dc60ada4bfa85155af972e99d0e Mon Sep 17 00:00:00 2001 From: Markus Frohme Date: Tue, 14 Jul 2026 12:53:54 +0200 Subject: [PATCH 3/5] fix setups --- examples/pom.xml | 8 +- ...mallRyeExample.java => MutinyExample.java} | 26 ++- .../example/reactive/RXJavaExample.java | 115 +++++------ .../example/reactive/ReactorExample.java | 182 +++--------------- .../de/learnlib/example/ExamplesTest.java | 5 + pom.xml | 6 + 6 files changed, 106 insertions(+), 236 deletions(-) rename examples/src/main/java/de/learnlib/example/reactive/{SmallRyeExample.java => MutinyExample.java} (84%) diff --git a/examples/pom.xml b/examples/pom.xml index c005dc56d..d8b48718d 100644 --- a/examples/pom.xml +++ b/examples/pom.xml @@ -150,11 +150,9 @@ limitations under the License. rxjava - io.smallrye.reactive - mutiny - 3.3.0 - - + io.smallrye.reactive + mutiny + org.reactivestreams reactive-streams diff --git a/examples/src/main/java/de/learnlib/example/reactive/SmallRyeExample.java b/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java similarity index 84% rename from examples/src/main/java/de/learnlib/example/reactive/SmallRyeExample.java rename to examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java index ac618afa2..da48871ff 100644 --- a/examples/src/main/java/de/learnlib/example/reactive/SmallRyeExample.java +++ b/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java @@ -36,7 +36,7 @@ * An example of constructing a learn-loop using reactive streams (from SmallRye) to compute counterexamples. */ @SuppressWarnings({"PMD.SystemPrintln", "PMD.UseExplicitTypes"}) // allow println and vars in examples -public final class SmallRyeExample { +public final class MutinyExample { private static final int SEED = 42; private static final int SIZE = 10; @@ -44,7 +44,7 @@ public final class SmallRyeExample { private static final int RND_LENGTH = 4; private static final int LIMIT = 1000; - private SmallRyeExample() { + private MutinyExample() { // prevent instantiation } @@ -69,8 +69,10 @@ public static void main(String[] args) { // setup learner var learner = new TTTLearnerMealy<>(inputs, mqo); - // setup thread pool - var pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); + // setup thread pools + var pool = Executors.newFixedThreadPool(1); + var pool2 = Executors.newFixedThreadPool(1); + var pool3 = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); // learning loop learner.startLearning(); @@ -80,19 +82,23 @@ public static void main(String[] args) { // since we only access the hypothesis in read-only fashion it is fine to share the reference across threads final var finalHyp = hyp; var m1 = Multi.createFrom() - // create a publisher from first quivalence oracle - .items(eqo.generateTestWords(finalHyp, inputs)) + // create a (cold) publisher from first quivalence oracle + .items(() -> eqo.generateTestWords(finalHyp, inputs)) + // sample on own thread + .runSubscriptionOn(pool) // limit to 1000 elements .select().first(LIMIT); var m2 = Multi.createFrom() - // create a publisher from second quivalence oracle - .items(eqo2.generateTestWords(finalHyp, inputs)); + // create a (cold) publisher from second quivalence oracle + .items(() -> eqo2.generateTestWords(finalHyp, inputs)) + // sample on own thread + .runSubscriptionOn(pool2); var ce = Multi.createBy() // merge elements from both oracles in interleaving fashion .merging().streams(m1, m2) // run processing in parallel - .runSubscriptionOn(pool) + .runSubscriptionOn(pool3) // filter for counterexamples .filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w))) // toUni implicitly fetches the first element @@ -111,6 +117,8 @@ public static void main(String[] args) { // cleanup mqo.shutdown(); pool.shutdown(); + pool2.shutdown(); + pool3.shutdown(); // process results hyp = learner.getHypothesisModel(); diff --git a/examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java b/examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java index 1c84b477f..985db13da 100644 --- a/examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java +++ b/examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java @@ -17,19 +17,14 @@ import java.util.Objects; import java.util.Random; -import java.util.concurrent.Executors; -import de.learnlib.algorithm.LearningAlgorithm; import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy; import de.learnlib.driver.simulator.MealySimulatorSUL; -import de.learnlib.oracle.MembershipOracle; import de.learnlib.oracle.equivalence.KWayStateCoverEQOracleBuilder; import de.learnlib.oracle.equivalence.RandomWMethodEQOracle; import de.learnlib.oracle.parallelism.ParallelOracleBuilders; import de.learnlib.query.DefaultQuery; import io.reactivex.rxjava3.core.Flowable; -import io.reactivex.rxjava3.functions.Action; -import io.reactivex.rxjava3.functions.Consumer; import io.reactivex.rxjava3.schedulers.Schedulers; import net.automatalib.alphabet.impl.Alphabets; import net.automatalib.automaton.transducer.MealyMachine; @@ -45,6 +40,7 @@ public final class RXJavaExample { private static final int SEED = 42; private static final int SIZE = 10; + private static final int BATCH_SIZE = 10; private static final int NUM_INPUTS = 4; private static final int RND_LENGTH = 4; private static final int LIMIT = 1000; @@ -61,9 +57,11 @@ public static void main(String[] args) { // setup membership oracle var mealy = RandomAutomata.randomMealy(new Random(SEED), SIZE, inputs, outputs); var sul = new MealySimulatorSUL<>(mealy); - // IMPORTANT: make sure to use parallel-aware oracle (with thread local instances) - // because it will be called from different threads in the reactive environment - var mqo = ParallelOracleBuilders.newDynamicParallelOracle(sul).create(); + // note that we can still use a parallel oracle to answer query batches in parallel + var mqo = ParallelOracleBuilders.newStaticParallelOracle(sul) + .withNumInstances(BATCH_SIZE) + .withMinBatchSize(1) + .create(); // setup equivalence oracles var eqo = new RandomWMethodEQOracle<>(mqo, SIZE / 2, RND_LENGTH); @@ -73,45 +71,54 @@ public static void main(String[] args) { // setup learner var learner = new TTTLearnerMealy<>(inputs, mqo); - var tracker = new ProgressTracker<>(learner, mqo); - - // setup scheduler - // IMPORTANT: set interruptibleWorker to false in order to allow graceful oracle/SUL shutdown - var pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); - var scheduler = Schedulers.from(pool, false, true); // learning loop learner.startLearning(); var hyp = learner.getHypothesisModel(); - while (!tracker.hasFinished()) { + while (true) { // since we only access the hypothesis in read-only fashion it is fine to share the reference across threads final var finalHyp = hyp; - Flowable // create a publisher from first quivalence oracle - .fromStream(eqo.generateTestWords(finalHyp, inputs)) - // limit to 1000 elements - .take(LIMIT) - // merge with elements from second equivalence oracle - .mergeWith(Flowable.fromStream(eqo2.generateTestWords(finalHyp, inputs))) - // process items in parallel - .parallel() - // run on our non-interruptible scheduler - .runOn(scheduler) - // filter for counterexamples - .filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w))) - // process them sequentially - .sequential() - // the first counterexample suffices for refinement - .firstElement() - // use a blocking subscribe to ensure that all pipelines are cleared for the next iteration - // alternatively, use blockingGet like in the ReactorExample - .blockingSubscribe(tracker, System.out::println, tracker); + var ce = Flowable // create a (cold) publisher from first quivalence oracle + .defer(() -> Flowable.fromStream(eqo.generateTestWords(finalHyp, inputs))) + // sample on own thread + .subscribeOn(Schedulers.computation()) + // limit to 1000 elements + .take(LIMIT) + // merge with elements from second equivalence oracle, also sampled in its own thread + .mergeWith(Flowable.defer(() -> Flowable.fromStream(eqo2.generateTestWords(finalHyp, + inputs))) + .subscribeOn(Schedulers.computation())) + // map to queries ... + .map(DefaultQuery>::new) + // ... create batches ... + .buffer(BATCH_SIZE) + // ... and process in bulk + // NOTE: this happens synchronously to allow the oracle/SUL to gracefully shutdown + // There exist ways to create schedulers that do not interrupt running threads + // (Schedulers#from) but I still experienced the occasional race condition ... + // However, the oracle can still answer the batch in parallel itself + .doOnNext(mqo::processQueries) + // flat to individual queries + .flatMapIterable(l -> l) + // filter for counterexamples + .filter(q -> !Objects.equals(q.getOutput(), + finalHyp.computeSuffixOutput(q.getPrefix(), q.getSuffix()))) + // the first counterexample suffices for refinement + .firstElement() + // use a blocking operation to ensure that all pipelines are cleared for the next iteration + .blockingGet(); + + if (ce != null) { + learner.refineHypothesis(ce); + hyp = learner.getHypothesisModel(); + } else { + break; + } } // cleanup - pool.shutdown(); mqo.shutdown(); - scheduler.shutdown(); // process results hyp = learner.getHypothesisModel(); @@ -119,40 +126,4 @@ public static void main(String[] args) { System.out.println("Final hypothesis size " + hyp.size()); System.out.println("Is equivalent? " + Automata.testEquivalence(mealy, hyp, inputs)); } - - private static final class ProgressTracker implements Consumer>, Action { - - private boolean finished; - private final LearningAlgorithm learner; - private final MembershipOracle oracle; - - private ProgressTracker(LearningAlgorithm learner, MembershipOracle oracle) { - this.learner = learner; - this.oracle = oracle; - } - - /** - * The onSuccess handler. Upon receiving a counterexample, refine the hypothesis. - * - * @param w - * the counterexample - */ - @Override - public void accept(Word w) { - learner.refineHypothesis(new DefaultQuery<>(w, oracle.answerQuery(w))); - } - - /** - * The onComplete handler. When all items have been processed without finding a counterexample, terminate the - * learning loop. - */ - @Override - public void run() { - this.finished = true; - } - - private boolean hasFinished() { - return finished; - } - } } diff --git a/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java b/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java index f7a38f708..0c9990a46 100644 --- a/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java +++ b/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java @@ -17,18 +17,6 @@ import java.util.Objects; import java.util.Random; -import java.util.concurrent.Delayed; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; -import java.util.concurrent.FutureTask; -import java.util.concurrent.RunnableFuture; -import java.util.concurrent.RunnableScheduledFuture; -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy; import de.learnlib.driver.simulator.MealySimulatorSUL; @@ -41,9 +29,7 @@ import net.automatalib.util.automaton.Automata; import net.automatalib.util.automaton.random.RandomAutomata; import net.automatalib.word.Word; -import reactor.core.Disposable; import reactor.core.publisher.Flux; -import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; /** @@ -54,6 +40,7 @@ public final class ReactorExample { private static final int SEED = 42; private static final int SIZE = 10; + private static final int BATCH_SIZE = 10; private static final int NUM_INPUTS = 4; private static final int RND_LENGTH = 4; private static final int LIMIT = 1000; @@ -70,9 +57,11 @@ public static void main(String[] args) { // setup membership oracle var mealy = RandomAutomata.randomMealy(new Random(SEED), SIZE, inputs, outputs); var sul = new MealySimulatorSUL<>(mealy); - // IMPORTANT: make sure to use parallel-aware oracle (with thread local instances) - // because it will be called from different threads in the reactive environment - var mqo = ParallelOracleBuilders.newDynamicParallelOracle(sul).create(); + // note that we can still use a parallel oracle to answer query batches in parallel + var mqo = ParallelOracleBuilders.newStaticParallelOracle(sul) + .withNumInstances(BATCH_SIZE) + .withMinBatchSize(1) + .create(); // setup equivalence oracles var eqo = new RandomWMethodEQOracle<>(mqo, SIZE / 2, RND_LENGTH); @@ -83,39 +72,40 @@ public static void main(String[] args) { // setup learner var learner = new TTTLearnerMealy<>(inputs, mqo); - // setup scheduler - // IMPORTANT: set interruptibleWorker to false in order to allow graceful oracle/SUL shutdown - var pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); -// var scheduler = new PureNonInterruptingScheduler(Schedulers.parallel()); - // Decorate all BoundedElastic schedulers (or any standard pools) - Schedulers.addExecutorServiceDecorator("prevent-interrupts", (sched, executorService) -> - new ScheduledThreadPoolExecutor(1) { // Or wrap the existing executorService - @Override - protected RunnableScheduledFuture decorateTask( - Runnable runnable, RunnableScheduledFuture task) { - return new CustomScheduledFuture<>(task); - } - } - ); - // learning loop learner.startLearning(); var hyp = learner.getHypothesisModel(); while (true) { final var finalHyp = hyp; - var ce = Flux.fromStream(eqo.generateTestWords(finalHyp, inputs)) - .take(LIMIT) - .mergeWith(Flux.fromStream(eqo2.generateTestWords(finalHyp, inputs))) - .parallel() - .runOn(Schedulers.parallel()) - .filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w))) - .sequential() - .next() - .block(); + var ce = Flux + // create a (cold) publisher from first quivalence oracle + .fromStream(() -> eqo.generateTestWords(finalHyp, inputs)) + // sample on own thread + .subscribeOn(Schedulers.boundedElastic()) + // limit to 1000 elements + .take(LIMIT) + // merge with elements from second equivalence oracle, also sampled in its own thread + .mergeWith(Flux.fromStream(() -> eqo2.generateTestWords(finalHyp, inputs)) + .subscribeOn(Schedulers.boundedElastic())) + // map to queries ... + .map(DefaultQuery>::new) + // ... create batches ... + .buffer(BATCH_SIZE) + // ... and process in bulk + // NOTE: this happens synchronously to allow the oracle/SUL to gracefully shutdown + // I have not found a way to tell Reactor to not forcefully interrupt canceled threads ... + // However, the oracle can still answer the batch in parallel itself + .doOnNext(mqo::processQueries) + // flat to individual queries + .flatMapIterable(l -> l) + // filter for counterexamples + .filter(q -> !Objects.equals(q.getOutput(), finalHyp.computeSuffixOutput(q.getPrefix(), q.getSuffix()))) + // use a blocking call to extract the final counterexample + .blockFirst(); if (ce != null) { - learner.refineHypothesis(new DefaultQuery<>(ce, mqo.answerQuery(ce))); + learner.refineHypothesis(ce); hyp = learner.getHypothesisModel(); } else { break; @@ -123,9 +113,7 @@ protected RunnableScheduledFuture decorateTask( } // cleanup - pool.shutdown(); mqo.shutdown(); -// scheduler.dispose(); // process results hyp = learner.getHypothesisModel(); @@ -133,110 +121,4 @@ protected RunnableScheduledFuture decorateTask( System.out.println("Final hypothesis size " + hyp.size()); System.out.println("Is equivalent? " + Automata.testEquivalence(mealy, hyp, inputs)); } - - private static class NonInterruptingScheduler implements Scheduler { - - private final Scheduler delegate; - - public NonInterruptingScheduler(Scheduler delegate) { - this.delegate = delegate; - } - - @Override - public Disposable schedule(Runnable task) { - // Wrap the task to clear or ignore the interrupted status if something upstream triggers it - Runnable safeTask = () -> { - try { - task.run(); - } catch (Exception e) { - // Handle or catch unexpected InterruptedExceptions - } finally { - // Clear the interrupted status flag just in case - Thread.interrupted(); - } - }; - return delegate.schedule(safeTask); - } - - @Override - public Worker createWorker() { - return delegate.createWorker(); - } - - @Override - public void dispose() { - delegate.dispose(); - } - } - - public static class PureNonInterruptingScheduler implements Scheduler { - - private final Scheduler delegate; - private final ExecutorService executor = Executors.newFixedThreadPool(10); - - public PureNonInterruptingScheduler(Scheduler delegate) { - this.delegate = delegate; - } - - @Override - public Disposable schedule(Runnable task) { - return delegate.schedule(task); - } - - @Override - public Worker createWorker() { - return new NonInterruptingWorker(executor); - } - - @Override - public void dispose() { - executor.shutdown(); - delegate.dispose(); - } - - private static class NonInterruptingWorker implements Worker { - - private final ExecutorService executor; - - public NonInterruptingWorker(ExecutorService executor) { - this.executor = executor; - } - - @Override - public Disposable schedule(Runnable task) { - Future future = executor.submit(task); - return () -> future.cancel(false); // Overrides thread interruption logic - } - - @Override - public void dispose() { - // Keep empty if you do not want worker disposal to break the underlying pool - } - } - } - // A delegate wrapper that ignores the "mayInterruptIfRunning" flag - private static class CustomScheduledFuture implements RunnableScheduledFuture { - private final RunnableScheduledFuture delegate; - - public CustomScheduledFuture(RunnableScheduledFuture delegate) { - this.delegate = delegate; - } - - @Override - public boolean cancel(boolean mayInterruptIfRunning) { - // FORCE false to prevent Thread.interrupt() - return delegate.cancel(false); - } - - // Delegate all other standard interface methods... - @Override public boolean isCancelled() { return delegate.isCancelled(); } - @Override public boolean isDone() { return delegate.isDone(); } - @Override public V get() throws InterruptedException, ExecutionException { return delegate.get(); } - @Override public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, - TimeoutException { return delegate.get(timeout, unit); } - @Override public long getDelay(TimeUnit unit) { return delegate.getDelay(unit); } - @Override public int compareTo(Delayed o) { return delegate.compareTo(o); } - @Override public void run() { delegate.run(); } - @Override public boolean isPeriodic() { return delegate.isPeriodic(); } - } } diff --git a/examples/src/test/java/de/learnlib/example/ExamplesTest.java b/examples/src/test/java/de/learnlib/example/ExamplesTest.java index 993214313..3611c519b 100644 --- a/examples/src/test/java/de/learnlib/example/ExamplesTest.java +++ b/examples/src/test/java/de/learnlib/example/ExamplesTest.java @@ -175,6 +175,11 @@ public void testReactiveRXJavaExample() { de.learnlib.example.reactive.RXJavaExample.main(new String[0]); } + @Test + public void testReactiveMutinyExample() { + de.learnlib.example.reactive.MutinyExample.main(new String[0]); + } + @Test public void testResumableExample() { de.learnlib.example.resumable.ResumableExample.main(new String[0]); diff --git a/pom.xml b/pom.xml index 95da30dd5..1c8d6042e 100644 --- a/pom.xml +++ b/pom.xml @@ -279,6 +279,7 @@ limitations under the License. 1.5.35 1.11 5.20.0 + 3.3.0 7.22.0 1.0.4 3.8.6 @@ -682,6 +683,11 @@ limitations under the License. rxjava ${rxjava.version} + + io.smallrye.reactive + mutiny + ${mutiny.version} + org.reactivestreams reactive-streams From 9013ff436877ad5bcbc0d491563d9b197e00ba89 Mon Sep 17 00:00:00 2001 From: Markus Frohme Date: Tue, 14 Jul 2026 16:04:57 +0200 Subject: [PATCH 4/5] supress warning that cannot be fixed on Java 17 --- .../main/java/de/learnlib/example/reactive/MutinyExample.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java b/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java index da48871ff..0f225e7f3 100644 --- a/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java +++ b/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java @@ -35,7 +35,9 @@ /** * An example of constructing a learn-loop using reactive streams (from SmallRye) to compute counterexamples. */ -@SuppressWarnings({"PMD.SystemPrintln", "PMD.UseExplicitTypes"}) // allow println and vars in examples +// allow println and vars in examples +// ExecutorService does not implement AutoClosable on Java 17 +@SuppressWarnings({"PMD.SystemPrintln", "PMD.UseExplicitTypes", "PMD.CloseResource"}) public final class MutinyExample { private static final int SEED = 42; From 5930dc17aa716f9c6b900bb4be3f6455f60259e4 Mon Sep 17 00:00:00 2001 From: Markus Frohme Date: Tue, 14 Jul 2026 16:49:20 +0200 Subject: [PATCH 5/5] rigorously suppress all PMD warnings --- .../java/de/learnlib/example/reactive/MutinyExample.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java b/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java index 0f225e7f3..77a3ed18d 100644 --- a/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java +++ b/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java @@ -35,9 +35,8 @@ /** * An example of constructing a learn-loop using reactive streams (from SmallRye) to compute counterexamples. */ -// allow println and vars in examples -// ExecutorService does not implement AutoClosable on Java 17 -@SuppressWarnings({"PMD.SystemPrintln", "PMD.UseExplicitTypes", "PMD.CloseResource"}) +// allow println and vars in examples, ExecutorService does not implement AutoClosable until Java 19+ +@SuppressWarnings("PMD") public final class MutinyExample { private static final int SEED = 42;