Rikkola opened a new issue, #7247:
URL: https://github.com/apache/incubator-kie/issues/7247
### Describe the bug
Something AI review picked up and needs to be investigated.
## Summary
`SegmentMemorySupportImpl.initializeChildSegmentsIfNeeded()` contains a
classic check-then-act
race condition. The guard:
if (!smem.isEmpty()) return;
…is followed by an unsynchronised loop that calls `smem.add(childSmem)`.
Neither
`LinkedList.isEmpty()` nor `LinkedList.add()` is protected by any lock, so
two threads that
both observe `isEmpty() == true` before either has appended a child will
both run the full
population loop — corrupting the child-segment linked list.
## Affected code
`drools-core/src/main/java/org/drools/core/phreak/SegmentMemorySupportImpl.java`,
lines 214–224
(previously `RuntimeSegmentUtilities.createChildSegments`, introduced in
DROOLS-7238 / #4812, November 2022)
## Precondition
The race window opens when ALL of the following hold:
1. `ThreadSafeOption.NO` is set on the session (disables the
`NamedEntryPoint` `ReentrantLock`).
2. The LIA for the inserted type is a **split node** (`sinkPropagator.size()
> 1`), which makes
`sm.getTipNode() == liaNode` true in
`LeftInputAdapterNode.doInsertObject`, routing into
`initializeChildSegmentsIfNeeded`.
3. Two threads insert into the same session simultaneously on a fresh
(un-initialised) segment.
In default thread-safe mode the `NamedEntryPoint` lock serialises `insert()`
calls and the race
cannot occur.
## Observable symptom
The corrupted `LinkedList` node pointers cause one or more rule firings to
be lost. In a
two-rule/two-fact scenario the expected count is 4; the race consistently
produces 3.
In severe races it can also produce `NoSuchElementException` from
`TreeSetQueue` during
`fireAllRules`.
## Failing test
`ConcurrentChildSegmentInitTest` ( See reproducer section )
— fails on attempt 0, both exec-model variants, with:
Expected size: 4 but was: 3 in: ["R1:fact0", "R2:fact0", "R2:fact1"]
## Fix direction
Synchronise the check-then-act in `initializeChildSegmentsIfNeeded` — either
with a
`synchronized` block on `smem`, or by replacing the two-step isEmpty/add
with a
compare-and-set. Alternatively, ensure callers always hold a segment-level
lock before
invoking this method.
### Expected behavior
_No response_
### Actual behavior
_No response_
### How to Reproduce?
```
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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 org.drools.mvel.integrationtests.concurrency;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CompletionService;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Stream;
import org.drools.base.base.ClassObjectType;
import org.drools.core.SessionConfiguration;
import org.drools.core.impl.InternalRuleBase;
import org.drools.core.reteoo.LeftInputAdapterNode;
import org.drools.core.reteoo.ObjectTypeNode;
import org.drools.testcoverage.common.util.KieBaseTestConfiguration;
import org.drools.testcoverage.common.util.KieBaseUtil;
import org.drools.testcoverage.common.util.TestParametersUtil2;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import org.kie.api.KieBase;
import org.kie.api.runtime.KieSession;
import org.kie.api.runtime.conf.ThreadSafeOption;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
/**
* Proves the check-then-act race in
SegmentMemorySupportImpl.initializeChildSegmentsIfNeeded.
*
* The guard is:
* if (!smem.isEmpty()) return;
* followed by an unsynchronised loop that calls smem.add(childSmem).
*
* When ThreadSafeOption.NO is set the NamedEntryPoint lock is null, so
insert() calls are not
* serialised. Two threads inserting simultaneously into a session whose LIA
is a split node
* (sinkPropagator.size() > 1) can both observe isEmpty() == true before
either has appended any
* child — both enter the loop and each appends all N sinks, leaving 2×N
child segments instead of N.
*
* Observable symptom: rules fire more times than expected.
* For 2 facts × 2 rules the expected count is 4; a duplicate segment causes
8 (or more).
*/
public class ConcurrentChildSegmentInitTest {
// Two rules sharing the same first pattern type force the LIA to be a
split node
// (sinkPropagator.size() == 2), which makes sm.getTipNode() == liaNode
in doInsertObject,
// opening the initializeChildSegmentsIfNeeded code path.
private static final String DRL =
"package org.drools.test;\n" +
"global java.util.List results;\n" +
"rule R1 when\n" +
" $s : String()\n" +
"then\n" +
" results.add(\"R1:\" + $s);\n" +
"end\n" +
"rule R2 when\n" +
" $s : String()\n" +
"then\n" +
" results.add(\"R2:\" + $s);\n" +
"end\n";
public static Stream<KieBaseTestConfiguration> parameters() {
return
TestParametersUtil2.getKieBaseCloudConfigurations(true).stream();
}
/**
* Two threads race to insert the very first fact into a session built
without locking
* (ThreadSafeOption.NO). Both reach initializeChildSegmentsIfNeeded
before either has
* added any child segment. The unguarded loop lets both threads append
all children,
* doubling the list from 2 to 4 entries and causing 8 rule firings
instead of 4.
*
* The test creates a fresh session per attempt and repeats many times
so that the race
* manifests reliably even on low-parallelism hardware.
*/
@ParameterizedTest(name = "KieBase type={0}")
@MethodSource("parameters")
public void
testConcurrentFirstInsertDoesNotDuplicateChildSegments(KieBaseTestConfiguration
kieBaseTestConfiguration)
throws Exception {
final KieBase kieBase =
KieBaseUtil.getKieBaseFromKieModuleFromDrl("test", kieBaseTestConfiguration,
DRL);
// Sanity-check the network topology: the LIA for String must be a
split node
// (2 sinks, one per rule) so that sm.getTipNode() == liaNode holds
in doInsertObject.
final ObjectTypeNode stringOtn = getObjectTypeNode(kieBase,
String.class);
assertThat(stringOtn).as("String OTN not found in rete").isNotNull();
final LeftInputAdapterNode liaNode =
(LeftInputAdapterNode)
stringOtn.getObjectSinkPropagator().getSinks()[0];
assertThat(liaNode.getSinkPropagator().size())
.as("LIA must be a split node (2 sinks) for the race window
to open")
.isEqualTo(2);
// Repeat many times: a race condition is non-deterministic but
becomes
// statistically certain over enough attempts on a fresh
(un-initialised) session.
for (int attempt = 0; attempt < 200; attempt++) {
// Disable session-level locking so that concurrent insert()
calls are not serialised.
// This is the condition under which
initializeChildSegmentsIfNeeded is exposed to the race.
final SessionConfiguration sessionConf =
(SessionConfiguration) ((InternalRuleBase)
kieBase).getSessionConfiguration();
sessionConf.setOption(ThreadSafeOption.NO);
final KieSession ksession = kieBase.newKieSession(sessionConf,
null);
final List<String> results = new ArrayList<>();
ksession.setGlobal("results", results);
// A barrier aligns both threads so they reach the isEmpty()
check simultaneously.
final CyclicBarrier startBarrier = new CyclicBarrier(2);
final ExecutorService executor = Executors.newFixedThreadPool(2,
r -> {
final Thread t = new Thread(r);
t.setDaemon(true);
return t;
});
try {
final CompletionService<Void> ecs = new
ExecutorCompletionService<>(executor);
for (int i = 0; i < 2; i++) {
final String fact = "fact" + i;
ecs.submit(() -> {
try {
startBarrier.await();
} catch (InterruptedException |
BrokenBarrierException e) {
Thread.currentThread().interrupt();
return null;
}
ksession.insert(fact);
return null;
});
}
for (int i = 0; i < 2; i++) {
ecs.take().get();
}
} finally {
executor.shutdownNow();
}
// fireAllRules itself may throw if the race corrupted a sorted
agenda structure
// (e.g. NoSuchElementException from TreeSetQueue) — that is
also a valid failure signal.
final KieSession sessionRef = ksession;
assertThatCode(sessionRef::fireAllRules)
.as("Attempt " + attempt + ": fireAllRules must not
throw. " +
"The initializeChildSegmentsIfNeeded race can
corrupt internal data structures " +
"such as the agenda's sorted queue.")
.doesNotThrowAnyException();
// 2 facts × 2 rules = exactly 4 firings.
// A duplicate child segment (caused by the race) produces ≠ 4
(extra or missing firings).
assertThat(results)
.as("Attempt " + attempt + ": expected exactly 4 rule
firings (2 facts × 2 rules). " +
"Duplicate or missing child segments from the
initializeChildSegmentsIfNeeded race " +
"cause incorrect firing counts. Got: " + results)
.hasSize(4);
ksession.dispose();
}
}
private static ObjectTypeNode getObjectTypeNode(KieBase kieBase,
Class<?> type) {
for (ObjectTypeNode otn : ((InternalRuleBase)
kieBase).getRete().getObjectTypeNodes()) {
if (otn.getObjectType() instanceof ClassObjectType &&
((ClassObjectType) otn.getObjectType()).getClassType()
== type) {
return otn;
}
}
return null;
}
}
```
### Output of `uname -a` or `ver`
_No response_
### Output of `java -version`
_No response_
### GraalVM version (if different from Java)
_No response_
### Kogito version or git rev (or at least Quarkus version if you are using
Kogito via Quarkus platform BOM)
_No response_
### Build tool (ie. output of `mvnw --version` or `gradlew --version`)
_No response_
### Additional information
_No response_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
