Skip to content

Commit 01ce388

Browse files
CoderMChuCoderMChu
authored
Merge pull request #6 from aehrc/perf/rule-executor-rolling-workers
Keep validation workers supplied with concepts instead of every 10
2 parents 7625eec + 0ca9b80 commit 01ce388

4 files changed

Lines changed: 110 additions & 28 deletions

File tree

‎snomed-drools-engine/src/main/java/org/ihtsdo/drools/RuleExecutor.java‎

Lines changed: 36 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,11 @@
2626
import java.io.IOException;
2727
import java.util.*;
2828
import java.util.concurrent.Callable;
29+
import java.util.concurrent.ExecutionException;
2930
import java.util.concurrent.ExecutorService;
3031
import java.util.concurrent.Executors;
32+
import java.util.concurrent.Future;
33+
import java.util.concurrent.atomic.AtomicInteger;
3134
import java.util.stream.Collectors;
3235

3336
public class RuleExecutor {
@@ -114,7 +117,7 @@ public List<InvalidContent> execute(
114117
checkComponentsIntegrity(concepts, conceptService);
115118

116119
final List<List<InvalidContent>> sessionInvalidContent = new ArrayList<>();
117-
final List<InvalidContent> exceptionContents = new ArrayList<>();
120+
final List<InvalidContent> exceptionContents = Collections.synchronizedList(new ArrayList<>());
118121
int threads = concepts.size() == 1 ? 1 : 10;
119122

120123
Map<String, List<StatelessKieSession>> sessionMap = createKieSessionMap(ruleSetNames, conceptService, descriptionService, relationshipService, threads, sessionInvalidContent);
@@ -149,38 +152,38 @@ private void doValidateComponents(Collection<? extends Concept> concepts, boolea
149152
Date start = new Date();
150153
try (ExecutorService executorService = Executors.newFixedThreadPool(threads)) {
151154
List<Concept> conceptList = new ArrayList<>(concepts);
152-
List<Callable<String>> tasks = new ArrayList<>();
155+
List<Callable<String>> workers = new ArrayList<>(threads);
156+
AtomicInteger nextConceptIndex = new AtomicInteger();
157+
AtomicInteger completedConceptCount = new AtomicInteger();
153158
String total = String.format("%,d", concepts.size());
154-
int i = 0;
155-
while (i < concepts.size()) {
156-
Set<Component> components = new HashSet<>();
157-
Concept concept = conceptList.get(i++);
158-
addConcept(components, concept, includeInferredRelationships);
159-
int sessionIndex = tasks.size();
160-
tasks.add(() -> {
161-
try {
162-
List<StatelessKieSession> statelessKieSessions = sessionMap.get(String.valueOf(sessionIndex));
163-
statelessKieSessions.forEach(statelessKieSession -> statelessKieSession.execute(components));
164-
components.clear();
165-
} catch (Exception e) {
166-
exceptionContents.add(new InvalidContent(concept.getId(), concept, "An error occurred while running concept validation. Technical detail: " + e.getMessage(), Severity.ERROR));
159+
160+
for (int workerIndex = 0; workerIndex < threads; workerIndex++) {
161+
List<StatelessKieSession> statelessKieSessions = sessionMap.get(String.valueOf(workerIndex));
162+
workers.add(() -> {
163+
int conceptIndex;
164+
while ((conceptIndex = nextConceptIndex.getAndIncrement()) < conceptList.size()) {
165+
Concept concept = conceptList.get(conceptIndex);
166+
Set<Component> components = new HashSet<>();
167+
addConcept(components, concept, includeInferredRelationships);
168+
try {
169+
statelessKieSessions.forEach(statelessKieSession -> statelessKieSession.execute(components));
170+
} catch (Exception e) {
171+
exceptionContents.add(new InvalidContent(concept.getId(), concept, "An error occurred while running concept validation. Technical detail: " + e.getMessage(), Severity.ERROR));
172+
} finally {
173+
components.clear();
174+
}
175+
176+
int completed = completedConceptCount.incrementAndGet();
177+
if (completed % 10_000 == 0) {
178+
logger.info("Validated {} of {}", String.format("%,d", completed), total);
179+
}
167180
}
168181
return null;
169182
});
170-
171-
if (tasks.size() == threads) {
172-
runTasks(executorService, tasks);
173-
tasks.clear();
174-
}
175-
if (i % 10_000 == 0) {
176-
logger.info("Validated {} of {}", String.format("%,d", i), total);
177-
}
178-
}
179-
if (!tasks.isEmpty()) {
180-
runTasks(executorService, tasks);
181183
}
184+
runTasks(executorService, workers);
182185

183-
logger.info("Validated {} of {}", String.format("%,d", i), total);
186+
logger.info("Validated {} of {}", String.format("%,d", completedConceptCount.get()), total);
184187
logger.info("Rule execution took {} seconds", (new Date().getTime() - start.getTime()) / 1000);
185188
}
186189
}
@@ -317,9 +320,14 @@ private StatelessKieSession newStatelessKieSession(KieContainer kieContainer, Co
317320

318321
private void runTasks(ExecutorService executorService, List<Callable<String>> tasks) {
319322
try {
320-
executorService.invokeAll(tasks);
323+
for (Future<String> task : executorService.invokeAll(tasks)) {
324+
task.get();
325+
}
321326
} catch (InterruptedException e) {
327+
Thread.currentThread().interrupt();
322328
throw new RuleExecutorException("Validation tasks were interrupted.", e);
329+
} catch (ExecutionException e) {
330+
throw new RuleExecutorException("A validation worker failed.", e.getCause());
323331
}
324332
}
325333

‎snomed-drools-engine/src/test/java/org/ihtsdo/drools/unittest/RuleExecutorTest.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
public class RuleExecutorTest {
2626

2727
private static final Set<String> RULE_SET_NAMES = Collections.singleton("Common");
28+
private static final Set<String> SCHEDULING_RULE_SET_NAMES = Collections.singleton("Scheduling");
2829
private RuleExecutor ruleExecutor;
2930
private TestConceptService conceptService;
3031
private TestDescriptionService descriptionService;
@@ -138,4 +139,20 @@ public void testExecuteWithExcludedRules() {
138139
invalidContents = ruleExecutor.execute(RULE_SET_NAMES, excludedRules, Collections.singleton(concept), conceptService, descriptionService, relationshipService, true, false);
139140
Assert.assertEquals(0, invalidContents.size());
140141
}
142+
143+
@Test
144+
public void testWorkersClaimNextConceptWithoutBatchBarrier() {
145+
List<Concept> concepts = new ArrayList<>();
146+
for (int i = 1; i <= 11; i++) {
147+
concepts.add(new ConceptImpl(String.valueOf(i)));
148+
}
149+
RuleSchedulingProbe.reset();
150+
151+
ruleExecutor.execute(SCHEDULING_RULE_SET_NAMES, null, concepts, conceptService,
152+
descriptionService, relationshipService, true, false);
153+
154+
Assert.assertEquals(11, RuleSchedulingProbe.getVisitedConceptCount());
155+
Assert.assertTrue("A free worker should claim concept 11 while concept 1 is still running.",
156+
RuleSchedulingProbe.wasBlockedConceptReleasedByLaterConcept());
157+
}
141158
}

‎snomed-drools-engine/src/test/java/org/ihtsdo/drools/unittest/RuleSchedulingProbe.java‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
package org.ihtsdo.drools.unittest;
2+
3+
import java.util.Set;
4+
import java.util.concurrent.ConcurrentHashMap;
5+
import java.util.concurrent.CountDownLatch;
6+
import java.util.concurrent.TimeUnit;
7+
import java.util.concurrent.atomic.AtomicBoolean;
8+
9+
public final class RuleSchedulingProbe {
10+
11+
private static final String BLOCKED_CONCEPT_ID = "1";
12+
private static final String RELEASING_CONCEPT_ID = "11";
13+
private static final Set<String> visitedConceptIds = ConcurrentHashMap.newKeySet();
14+
private static final AtomicBoolean blockedConceptReleasedByLaterConcept = new AtomicBoolean();
15+
private static volatile CountDownLatch laterConceptVisited = new CountDownLatch(1);
16+
17+
private RuleSchedulingProbe() {
18+
}
19+
20+
public static void reset() {
21+
visitedConceptIds.clear();
22+
blockedConceptReleasedByLaterConcept.set(false);
23+
laterConceptVisited = new CountDownLatch(1);
24+
}
25+
26+
public static boolean visit(String conceptId) {
27+
visitedConceptIds.add(conceptId);
28+
if (RELEASING_CONCEPT_ID.equals(conceptId)) {
29+
laterConceptVisited.countDown();
30+
} else if (BLOCKED_CONCEPT_ID.equals(conceptId)) {
31+
try {
32+
blockedConceptReleasedByLaterConcept.set(laterConceptVisited.await(2, TimeUnit.SECONDS));
33+
} catch (InterruptedException e) {
34+
Thread.currentThread().interrupt();
35+
}
36+
}
37+
return false;
38+
}
39+
40+
public static boolean wasBlockedConceptReleasedByLaterConcept() {
41+
return blockedConceptReleasedByLaterConcept.get();
42+
}
43+
44+
public static int getVisitedConceptCount() {
45+
return visitedConceptIds.size();
46+
}
47+
}

‎snomed-drools-engine/src/test/resources/rules/Scheduling/rolling-workers/SchedulingProbe.drl‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
import org.ihtsdo.drools.domain.Concept
2+
import org.ihtsdo.drools.unittest.RuleSchedulingProbe
3+
4+
rule "Scheduling probe"
5+
dialect "mvel"
6+
when
7+
c : Concept()
8+
eval(RuleSchedulingProbe.visit(c.id))
9+
then
10+
end

0 commit comments

Comments
 (0)