diff --git a/examples/pom.xml b/examples/pom.xml index 4781159c3..d8b48718d 100644 --- a/examples/pom.xml +++ b/examples/pom.xml @@ -141,6 +141,23 @@ limitations under the License. runtime + + io.projectreactor + reactor-core + + + io.reactivex.rxjava3 + rxjava + + + io.smallrye.reactive + mutiny + + + org.reactivestreams + reactive-streams + + net.automatalib automata-api diff --git a/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java b/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java new file mode 100644 index 000000000..77a3ed18d --- /dev/null +++ b/examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java @@ -0,0 +1,130 @@ +/* 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. + */ +// 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; + 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 MutinyExample() { + // 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 pools + var pool = Executors.newFixedThreadPool(1); + var pool2 = Executors.newFixedThreadPool(1); + var pool3 = 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 (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 (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(pool3) + // 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(); + pool2.shutdown(); + pool3.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/de/learnlib/example/reactive/RXJavaExample.java b/examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java new file mode 100644 index 000000000..985db13da --- /dev/null +++ b/examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java @@ -0,0 +1,129 @@ +/* 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 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.reactivex.rxjava3.core.Flowable; +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 BATCH_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); + // 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); + var eqo2 = + new KWayStateCoverEQOracleBuilder, Integer, Word>().withOracle( + mqo).withRandom(new Random(SEED)).create(); + + // setup learner + var learner = new TTTLearnerMealy<>(inputs, mqo); + + // 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 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 + mqo.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/de/learnlib/example/reactive/ReactorExample.java b/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java new file mode 100644 index 000000000..0c9990a46 --- /dev/null +++ b/examples/src/main/java/de/learnlib/example/reactive/ReactorExample.java @@ -0,0 +1,124 @@ +/* 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 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 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; +import reactor.core.publisher.Flux; +import reactor.core.scheduler.Schedulers; + +/** + * 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 BATCH_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, 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); + // 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); + var eqo2 = + new KWayStateCoverEQOracleBuilder, Integer, Word>().withOracle( + mqo).withRandom(new Random(SEED)).create(); + + // setup learner + var learner = new TTTLearnerMealy<>(inputs, mqo); + + // learning loop + learner.startLearning(); + var hyp = learner.getHypothesisModel(); + + while (true) { + final var finalHyp = hyp; + 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(ce); + hyp = learner.getHypothesisModel(); + } else { + break; + } + } + + // cleanup + mqo.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 7b4b99096..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,6 +55,8 @@ 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; @@ -62,6 +66,7 @@ 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; } diff --git a/examples/src/test/java/de/learnlib/example/ExamplesTest.java b/examples/src/test/java/de/learnlib/example/ExamplesTest.java index 7e1e47e76..3611c519b 100644 --- a/examples/src/test/java/de/learnlib/example/ExamplesTest.java +++ b/examples/src/test/java/de/learnlib/example/ExamplesTest.java @@ -165,6 +165,21 @@ 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 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 c52da4a5b..1c8d6042e 100644 --- a/pom.xml +++ b/pom.xml @@ -279,7 +279,11 @@ limitations under the License. 1.5.35 1.11 5.20.0 + 3.3.0 7.22.0 + 1.0.4 + 3.8.6 + 3.1.12 2.0.16 7.10.2 @@ -668,6 +672,28 @@ limitations under the License. ${logback.version} + + + io.projectreactor + reactor-core + ${reactor.version} + + + io.reactivex.rxjava3 + rxjava + ${rxjava.version} + + + io.smallrye.reactive + mutiny + ${mutiny.version} + + + org.reactivestreams + reactive-streams + ${reactive-streams.version} + +