Skip to content

Commit 82ea553

Browse files
mtf90mtf90
authored
Add examples for reactive streams (#165)
* initial experimentation * some more progress, Project Reactor still borked * fix setups * supress warning that cannot be fixed on Java 17 * rigorously suppress all PMD warnings
1 parent e7aa897 commit 82ea553

7 files changed

Lines changed: 446 additions & 0 deletions

File tree

‎examples/pom.xml‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,23 @@ limitations under the License.
141141
<scope>runtime</scope>
142142
</dependency>
143143

144+
<dependency>
145+
<groupId>io.projectreactor</groupId>
146+
<artifactId>reactor-core</artifactId>
147+
</dependency>
148+
<dependency>
149+
<groupId>io.reactivex.rxjava3</groupId>
150+
<artifactId>rxjava</artifactId>
151+
</dependency>
152+
<dependency>
153+
<groupId>io.smallrye.reactive</groupId>
154+
<artifactId>mutiny</artifactId>
155+
</dependency>
156+
<dependency>
157+
<groupId>org.reactivestreams</groupId>
158+
<artifactId>reactive-streams</artifactId>
159+
</dependency>
160+
144161
<dependency>
145162
<groupId>net.automatalib</groupId>
146163
<artifactId>automata-api</artifactId>

‎examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/* Copyright (C) 2013-2026 TU Dortmund University
2+
* This file is part of LearnLib <https://learnlib.de>.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package de.learnlib.example.reactive;
17+
18+
import java.util.Objects;
19+
import java.util.Random;
20+
import java.util.concurrent.Executors;
21+
22+
import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy;
23+
import de.learnlib.driver.simulator.MealySimulatorSUL;
24+
import de.learnlib.oracle.equivalence.KWayStateCoverEQOracleBuilder;
25+
import de.learnlib.oracle.equivalence.RandomWMethodEQOracle;
26+
import de.learnlib.oracle.parallelism.ParallelOracleBuilders;
27+
import de.learnlib.query.DefaultQuery;
28+
import io.smallrye.mutiny.Multi;
29+
import net.automatalib.alphabet.impl.Alphabets;
30+
import net.automatalib.automaton.transducer.MealyMachine;
31+
import net.automatalib.util.automaton.Automata;
32+
import net.automatalib.util.automaton.random.RandomAutomata;
33+
import net.automatalib.word.Word;
34+
35+
/**
36+
* An example of constructing a learn-loop using reactive streams (from SmallRye) to compute counterexamples.
37+
*/
38+
// allow println and vars in examples, ExecutorService does not implement AutoClosable until Java 19+
39+
@SuppressWarnings("PMD")
40+
public final class MutinyExample {
41+
42+
private static final int SEED = 42;
43+
private static final int SIZE = 10;
44+
private static final int NUM_INPUTS = 4;
45+
private static final int RND_LENGTH = 4;
46+
private static final int LIMIT = 1000;
47+
48+
private MutinyExample() {
49+
// prevent instantiation
50+
}
51+
52+
public static void main(String[] args) {
53+
// setup symbols
54+
var inputs = Alphabets.integers(0, NUM_INPUTS);
55+
var outputs = Alphabets.characters('a', 'd');
56+
57+
// setup membership oracle
58+
var mealy = RandomAutomata.randomMealy(new Random(SEED), SIZE, inputs, outputs);
59+
var sul = new MealySimulatorSUL<>(mealy);
60+
// IMPORTANT: make sure to use parallel-aware oracle (with thread local instances)
61+
// because it will be called from different threads in the reactive environment
62+
var mqo = ParallelOracleBuilders.newDynamicParallelOracle(sul).create();
63+
64+
// setup equivalence oracles
65+
var eqo = new RandomWMethodEQOracle<>(mqo, SIZE / 2, RND_LENGTH);
66+
var eqo2 =
67+
new KWayStateCoverEQOracleBuilder<MealyMachine<?, Integer, ?, Character>, Integer, Word<Character>>().withOracle(
68+
mqo).withRandom(new Random(SEED)).create();
69+
70+
// setup learner
71+
var learner = new TTTLearnerMealy<>(inputs, mqo);
72+
73+
// setup thread pools
74+
var pool = Executors.newFixedThreadPool(1);
75+
var pool2 = Executors.newFixedThreadPool(1);
76+
var pool3 = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
77+
78+
// learning loop
79+
learner.startLearning();
80+
var hyp = learner.getHypothesisModel();
81+
82+
while (true) {
83+
// since we only access the hypothesis in read-only fashion it is fine to share the reference across threads
84+
final var finalHyp = hyp;
85+
var m1 = Multi.createFrom()
86+
// create a (cold) publisher from first quivalence oracle
87+
.items(() -> eqo.generateTestWords(finalHyp, inputs))
88+
// sample on own thread
89+
.runSubscriptionOn(pool)
90+
// limit to 1000 elements
91+
.select().first(LIMIT);
92+
var m2 = Multi.createFrom()
93+
// create a (cold) publisher from second quivalence oracle
94+
.items(() -> eqo2.generateTestWords(finalHyp, inputs))
95+
// sample on own thread
96+
.runSubscriptionOn(pool2);
97+
98+
var ce = Multi.createBy()
99+
// merge elements from both oracles in interleaving fashion
100+
.merging().streams(m1, m2)
101+
// run processing in parallel
102+
.runSubscriptionOn(pool3)
103+
// filter for counterexamples
104+
.filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w)))
105+
// toUni implicitly fetches the first element
106+
.toUni()
107+
// use a blocking call to extract the final counterexample
108+
.await().indefinitely();
109+
110+
if (ce != null) {
111+
learner.refineHypothesis(new DefaultQuery<>(ce, mqo.answerQuery(ce)));
112+
hyp = learner.getHypothesisModel();
113+
} else {
114+
break;
115+
}
116+
}
117+
118+
// cleanup
119+
mqo.shutdown();
120+
pool.shutdown();
121+
pool2.shutdown();
122+
pool3.shutdown();
123+
124+
// process results
125+
hyp = learner.getHypothesisModel();
126+
127+
System.out.println("Final hypothesis size " + hyp.size());
128+
System.out.println("Is equivalent? " + Automata.testEquivalence(mealy, hyp, inputs));
129+
}
130+
}

‎examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
1+
/* Copyright (C) 2013-2026 TU Dortmund University
2+
* This file is part of LearnLib <https://learnlib.de>.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package de.learnlib.example.reactive;
17+
18+
import java.util.Objects;
19+
import java.util.Random;
20+
21+
import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy;
22+
import de.learnlib.driver.simulator.MealySimulatorSUL;
23+
import de.learnlib.oracle.equivalence.KWayStateCoverEQOracleBuilder;
24+
import de.learnlib.oracle.equivalence.RandomWMethodEQOracle;
25+
import de.learnlib.oracle.parallelism.ParallelOracleBuilders;
26+
import de.learnlib.query.DefaultQuery;
27+
import io.reactivex.rxjava3.core.Flowable;
28+
import io.reactivex.rxjava3.schedulers.Schedulers;
29+
import net.automatalib.alphabet.impl.Alphabets;
30+
import net.automatalib.automaton.transducer.MealyMachine;
31+
import net.automatalib.util.automaton.Automata;
32+
import net.automatalib.util.automaton.random.RandomAutomata;
33+
import net.automatalib.word.Word;
34+
35+
/**
36+
* An example of constructing a learn-loop using reactive streams (from RXJava) to compute counterexamples.
37+
*/
38+
@SuppressWarnings({"PMD.SystemPrintln", "PMD.UseExplicitTypes"}) // allow println and vars in examples
39+
public final class RXJavaExample {
40+
41+
private static final int SEED = 42;
42+
private static final int SIZE = 10;
43+
private static final int BATCH_SIZE = 10;
44+
private static final int NUM_INPUTS = 4;
45+
private static final int RND_LENGTH = 4;
46+
private static final int LIMIT = 1000;
47+
48+
private RXJavaExample() {
49+
// prevent instantiation
50+
}
51+
52+
public static void main(String[] args) {
53+
// setup symbols
54+
var inputs = Alphabets.integers(0, NUM_INPUTS);
55+
var outputs = Alphabets.characters('a', 'd');
56+
57+
// setup membership oracle
58+
var mealy = RandomAutomata.randomMealy(new Random(SEED), SIZE, inputs, outputs);
59+
var sul = new MealySimulatorSUL<>(mealy);
60+
// note that we can still use a parallel oracle to answer query batches in parallel
61+
var mqo = ParallelOracleBuilders.newStaticParallelOracle(sul)
62+
.withNumInstances(BATCH_SIZE)
63+
.withMinBatchSize(1)
64+
.create();
65+
66+
// setup equivalence oracles
67+
var eqo = new RandomWMethodEQOracle<>(mqo, SIZE / 2, RND_LENGTH);
68+
var eqo2 =
69+
new KWayStateCoverEQOracleBuilder<MealyMachine<?, Integer, ?, Character>, Integer, Word<Character>>().withOracle(
70+
mqo).withRandom(new Random(SEED)).create();
71+
72+
// setup learner
73+
var learner = new TTTLearnerMealy<>(inputs, mqo);
74+
75+
// learning loop
76+
learner.startLearning();
77+
var hyp = learner.getHypothesisModel();
78+
79+
while (true) {
80+
// since we only access the hypothesis in read-only fashion it is fine to share the reference across threads
81+
final var finalHyp = hyp;
82+
var ce = Flowable // create a (cold) publisher from first quivalence oracle
83+
.defer(() -> Flowable.fromStream(eqo.generateTestWords(finalHyp, inputs)))
84+
// sample on own thread
85+
.subscribeOn(Schedulers.computation())
86+
// limit to 1000 elements
87+
.take(LIMIT)
88+
// merge with elements from second equivalence oracle, also sampled in its own thread
89+
.mergeWith(Flowable.defer(() -> Flowable.fromStream(eqo2.generateTestWords(finalHyp,
90+
inputs)))
91+
.subscribeOn(Schedulers.computation()))
92+
// map to queries ...
93+
.map(DefaultQuery<Integer, Word<Character>>::new)
94+
// ... create batches ...
95+
.buffer(BATCH_SIZE)
96+
// ... and process in bulk
97+
// NOTE: this happens synchronously to allow the oracle/SUL to gracefully shutdown
98+
// There exist ways to create schedulers that do not interrupt running threads
99+
// (Schedulers#from) but I still experienced the occasional race condition ...
100+
// However, the oracle can still answer the batch in parallel itself
101+
.doOnNext(mqo::processQueries)
102+
// flat to individual queries
103+
.flatMapIterable(l -> l)
104+
// filter for counterexamples
105+
.filter(q -> !Objects.equals(q.getOutput(),
106+
finalHyp.computeSuffixOutput(q.getPrefix(), q.getSuffix())))
107+
// the first counterexample suffices for refinement
108+
.firstElement()
109+
// use a blocking operation to ensure that all pipelines are cleared for the next iteration
110+
.blockingGet();
111+
112+
if (ce != null) {
113+
learner.refineHypothesis(ce);
114+
hyp = learner.getHypothesisModel();
115+
} else {
116+
break;
117+
}
118+
}
119+
120+
// cleanup
121+
mqo.shutdown();
122+
123+
// process results
124+
hyp = learner.getHypothesisModel();
125+
126+
System.out.println("Final hypothesis size " + hyp.size());
127+
System.out.println("Is equivalent? " + Automata.testEquivalence(mealy, hyp, inputs));
128+
}
129+
}

0 commit comments

Comments
 (0)