Skip to content

Commit 53d92e6

Browse files
sterlpsterlp
committed
adjusted synchronization
1 parent 9a6f554 commit 53d92e6

File tree

7 files changed

+90
-61
lines changed

7 files changed

+90
-61
lines changed

‎core/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,7 @@
8686
<dependency>
8787
<groupId>org.sterl.test</groupId>
8888
<artifactId>hibernate-asserts</artifactId>
89-
<version>1.0.0</version>
89+
<version>1.0.1</version>
9090
<scope>test</scope>
9191
</dependency>
9292
<dependency>

‎core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/component/RunOrQueueComponent.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ private TriggerEntity offerToRun(TriggerEntity trigger) {
4747
if (taskExecutor.getFreeThreads() > 0) {
4848
trigger = triggerService.markTriggersAsRunning(trigger, schedulerName);
4949
shouldRun.put(trigger.getId(), trigger);
50-
log.debug("{} added for immediate execution, waiting for commit on={}", trigger.getKey(), schedulerName);
50+
log.debug("Immediate execution for={}, waiting for commit on={}", trigger.getKey(), schedulerName);
5151
} else {
5252
log.debug("Currently not enough free thread available {} of {} in use. PersistentTask {} queued.",
5353
taskExecutor.getFreeThreads(), taskExecutor.getMaxThreads(), trigger.getKey());

‎core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/component/TaskExecutorComponent.java‎

Lines changed: 47 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -8,12 +8,14 @@
88
import java.util.List;
99
import java.util.concurrent.CompletableFuture;
1010
import java.util.concurrent.ConcurrentHashMap;
11-
import java.util.concurrent.ExecutorService;
12-
import java.util.concurrent.Executors;
1311
import java.util.concurrent.Future;
12+
import java.util.concurrent.LinkedBlockingQueue;
13+
import java.util.concurrent.ThreadPoolExecutor;
1414
import java.util.concurrent.TimeUnit;
1515
import java.util.concurrent.atomic.AtomicBoolean;
1616
import java.util.concurrent.atomic.AtomicInteger;
17+
import java.util.concurrent.locks.Lock;
18+
import java.util.concurrent.locks.ReentrantLock;
1719

1820
import org.springframework.lang.NonNull;
1921
import org.springframework.lang.Nullable;
@@ -41,10 +43,10 @@ public class TaskExecutorComponent implements Closeable {
4143
@Setter
4244
private Duration maxShutdownWaitTime = Duration.ofSeconds(10);
4345
@Nullable
44-
private ExecutorService executor;
45-
// also the LOCK object ...
46+
private ThreadPoolExecutor executor;
4647
private final ConcurrentHashMap<TriggerEntity, Future<TriggerKey>> runningTasks = new ConcurrentHashMap<>();
4748
private final AtomicBoolean stopped = new AtomicBoolean(true);
49+
private final Lock lock = new ReentrantLock(true);
4850

4951
public TaskExecutorComponent(String schedulerName, TriggerService triggerService, int maxThreads) {
5052
super();
@@ -72,15 +74,16 @@ public Future<TriggerKey> submit(@Nullable TriggerEntity trigger) {
7274
}
7375
assertStarted();
7476
try {
75-
synchronized (runningTasks) {
76-
assertStarted();
77-
var result = executor.submit(() -> runTrigger(trigger));
78-
runningTasks.put(trigger, result);
79-
return result;
80-
}
77+
assertStarted();
78+
lock.lock();
79+
var result = executor.submit(() -> runTrigger(trigger));
80+
runningTasks.put(trigger, result);
81+
return result;
8182
} catch (Exception e) {
8283
runningTasks.remove(trigger);
8384
throw new RuntimeException("Failed to run " + trigger.getKey(), e);
85+
} finally {
86+
lock.unlock();
8487
}
8588

8689
}
@@ -96,61 +99,59 @@ private TriggerKey runTrigger(TriggerEntity trigger) {
9699
triggerService.run(trigger);
97100
return trigger.getKey();
98101
} finally {
99-
synchronized (runningTasks) {
102+
try {
103+
lock.lock();
100104
if (runningTasks.remove(trigger) == null && runningTasks.size() > 0) {
101105
var runningKeys = runningTasks.keySet().stream().map(TriggerEntity::key);
102106
log.error("Failed to remove trigger with {} - {}", trigger.key(), runningKeys);
103107
}
108+
} finally {
109+
lock.unlock();
104110
}
105111
}
106112
}
107113

108114
public void start() {
109-
synchronized (runningTasks) {
110-
if (stopped.compareAndExchange(true, false)) {
111-
runningTasks.clear();
112-
executor = Executors.newFixedThreadPool(maxThreads.get());
113-
log.info("Started {} with {} threads.", schedulerName, maxThreads.get());
114-
stopped.set(false);
115-
}
115+
if (stopped.compareAndExchange(true, false)) {
116+
runningTasks.clear();
117+
executor = new ThreadPoolExecutor(1, this.maxThreads.get(),
118+
0L, TimeUnit.MILLISECONDS,
119+
new LinkedBlockingQueue<Runnable>());
120+
log.info("Started {} with {} threads.", schedulerName, maxThreads.get());
121+
stopped.set(false);
116122
}
117123
}
118124

119125
@Override
120126
public void close() {
121127
stopped.set(true);
122-
synchronized (runningTasks) {
123-
if (executor != null) {
124-
var execRef = executor;
125-
execRef.shutdown();
126-
log.info("Shutdown {} with {} running tasks, waiting for {}.", schedulerName, runningTasks.size(),
127-
maxShutdownWaitTime);
128-
129-
if (runningTasks.size() > 0) {
130-
try {
131-
execRef.awaitTermination(maxShutdownWaitTime.getSeconds(), TimeUnit.SECONDS);
132-
} catch (InterruptedException e) {
133-
Thread.currentThread().interrupt();
134-
log.warn("Failed to complete runnings tasks.", e.getCause() == null ? e : e.getCause());
135-
shutdownNow();
136-
} finally {
137-
executor = null;
138-
runningTasks.clear();
139-
}
128+
if (executor != null) {
129+
executor.shutdown();
130+
log.info("Shutdown {} with {} running tasks, waiting for {}.", schedulerName, runningTasks.size(),
131+
maxShutdownWaitTime);
132+
133+
if (runningTasks.size() > 0) {
134+
try {
135+
executor.awaitTermination(maxShutdownWaitTime.getSeconds(), TimeUnit.SECONDS);
136+
} catch (InterruptedException e) {
137+
Thread.currentThread().interrupt();
138+
log.warn("Failed to complete runnings tasks.", e.getCause() == null ? e : e.getCause());
139+
shutdownNow();
140+
} finally {
141+
executor = null;
142+
runningTasks.clear();
140143
}
141144
}
142145
}
143146
}
144147

145148
public void shutdownNow() {
146-
synchronized (runningTasks) {
147-
if (executor != null) {
148-
executor.shutdownNow();
149-
log.info("Force stop {} with {} running tasks", schedulerName, runningTasks.size());
150-
runningTasks.clear();
151-
executor = null;
152-
stopped.set(true);
153-
}
149+
stopped.set(true);
150+
if (executor != null) {
151+
executor.shutdownNow();
152+
log.info("Force stop {} with {} running tasks", schedulerName, runningTasks.size());
153+
runningTasks.clear();
154+
executor = null;
154155
}
155156
}
156157

@@ -159,7 +160,7 @@ public int getFreeThreads() {
159160
return 0;
160161
}
161162
if (maxThreads.get() - runningTasks.size() < 0) {
162-
log.warn("Already {}" + runningTasks.size() + " running more than threads={}",
163+
log.warn("Already {} running tasks, more than threads {} in pool.",
163164
runningTasks.size(), maxThreads.get());
164165
}
165166
return Math.max(maxThreads.get() - runningTasks.size(), 0);
@@ -178,7 +179,7 @@ public List<TriggerEntity> getRunningTriggers() {
178179
.toList();
179180

180181
if (doneAndNotRemovedFutures.size() > 0) {
181-
log.error("Found still pending futures, maybe an issue, report a bug if so {}",
182+
log.warn("Found still pending futures, maybe an issue, report a bug if so {}",
182183
doneAndNotRemovedFutures.stream().map(e -> e.getKey().getKey()));
183184
for (var entry : doneAndNotRemovedFutures) {
184185
runningTasks.remove(entry.getKey());

‎core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerServiceTransactionTest.java‎

Lines changed: 27 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import java.util.concurrent.TimeoutException;
88
import java.util.concurrent.atomic.AtomicBoolean;
99

10+
import org.awaitility.Awaitility;
1011
import org.junit.jupiter.api.BeforeEach;
1112
import org.junit.jupiter.api.Test;
1213
import org.springframework.beans.factory.annotation.Autowired;
@@ -178,6 +179,31 @@ void test_fail_in_transaction() throws Exception {
178179
assertThat(history.get(2).getData().getStatus()).isEqualTo(TriggerStatus.WAITING);
179180
}
180181

182+
@Test
183+
void testRunOrQueueTransactions() throws Exception {
184+
// GIVEN & WHEN
185+
var k1 = subject.runOrQueue(TriggerBuilder.newTrigger("savePersonInTrx").state("Paul").build());
186+
187+
// THEN 1 to save and 1 to start it and 1 for the history
188+
Awaitility.await().until(() -> hibernateAsserts.getStatistics().getTransactionCount() > 2);
189+
Thread.sleep(250); // wait for the history async events
190+
hibernateAsserts.assertTrxCount(3);
191+
assertThat(persistentTaskService.getLastTriggerData(k1).get().getStatus())
192+
.isEqualTo(TriggerStatus.RUNNING);
193+
194+
// WHEN
195+
hibernateAsserts.reset();
196+
COUNTDOWN.countDown();
197+
Awaitility.await().until(() -> hibernateAsserts.getStatistics().getTransactionCount() >= 1);
198+
hibernateAsserts.assertTrxCount(1);
199+
200+
// THEN
201+
assertThat(personRepository.count()).isEqualTo(1);
202+
// AND
203+
assertThat(persistentTaskService.getLastTriggerData(k1).get().getStatus())
204+
.isEqualTo(TriggerStatus.SUCCESS);
205+
}
206+
181207
@Test
182208
void testRunOrQueueShowsRunning() throws Exception {
183209
// GIVEN
@@ -190,13 +216,9 @@ void testRunOrQueueShowsRunning() throws Exception {
190216
assertThat(persistentTaskService.getLastTriggerData(k2).get().getStatus())
191217
.isEqualTo(TriggerStatus.RUNNING);
192218

193-
// THEN
194-
Thread.sleep(150); // wait for the history async events
195-
hibernateAsserts.assertTrxCount(7);
196-
197-
// WHEN
198219
COUNTDOWN.countDown();
199220
awaitRunningTasks();
221+
200222
// THEN
201223
assertThat(personRepository.count()).isEqualTo(2);
202224
// AND

‎core/src/test/java/org/sterl/spring/persistent_tasks/test/Countdown.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ public class Countdown {
1111

1212
public void await() {
1313
Awaitility
14-
.await("Countdown")
14+
.await("Countdown " + count.get())
1515
.atMost(Duration.ofSeconds(3))
1616
.until(() -> count.get() <= 0);
1717
}
@@ -23,4 +23,8 @@ public void countDown() {
2323
public void reset() {
2424
count.set(1);
2525
}
26+
27+
public void reset(int newCount) {
28+
count.set(newCount);
29+
}
2630
}

‎example/README.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
# UI login
22

3+
`docker run --cap-add SYS_PTRACE -e 'ACCEPT_EULA=Y' -e 'MSSQL_SA_PASSWORD=veryStrong123' -p 1433:1433 --name azuresqledge -d mcr.microsoft.com/azure-sql-edge`
4+
35
- url: http://localhost:8080/task-ui
46
- user: admin
57
- password: admin

‎pom.xml‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -52,12 +52,12 @@
5252

5353
<properties>
5454
<java.version>21</java.version>
55-
<spring-boot.version>3.3.8</spring-boot.version>
55+
<spring-boot.version>3.3.11</spring-boot.version>
5656
<maven.compiler.source>${java.version}</maven.compiler.source>
5757
<maven.compiler.target>${java.version}</maven.compiler.target>
5858
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
5959
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
60-
<pmd.version>7.2.0</pmd.version>
60+
<pmd.version>7.13.0</pmd.version>
6161
</properties>
6262

6363
<modules>
@@ -92,7 +92,7 @@
9292
<plugin>
9393
<groupId>org.apache.maven.plugins</groupId>
9494
<artifactId>maven-javadoc-plugin</artifactId>
95-
<version>3.7.0</version>
95+
<version>3.11.2</version>
9696
</plugin>
9797

9898
<!-- Site Plugin Configuration -->
@@ -110,7 +110,7 @@
110110
<plugin>
111111
<groupId>org.apache.maven.plugins</groupId>
112112
<artifactId>maven-javadoc-plugin</artifactId>
113-
<version>3.7.0</version>
113+
<version>3.11.2</version>
114114
<executions>
115115
<execution>
116116
<id>attach-javadoc</id>
@@ -138,7 +138,7 @@
138138
<plugin>
139139
<groupId>org.apache.maven.plugins</groupId>
140140
<artifactId>maven-compiler-plugin</artifactId>
141-
<version>3.13.0</version>
141+
<version>3.14.0</version>
142142
<configuration>
143143
<source>${java.version}</source>
144144
<target>${java.version}</target>
@@ -154,7 +154,7 @@
154154
<plugin>
155155
<groupId>org.apache.maven.plugins</groupId>
156156
<artifactId>maven-surefire-plugin</artifactId>
157-
<version>3.3.0</version>
157+
<version>3.5.3</version>
158158
</plugin>
159159
<plugin>
160160
<groupId>org.sonatype.plugins</groupId>
@@ -179,7 +179,7 @@
179179
<plugin>
180180
<groupId>org.apache.maven.plugins</groupId>
181181
<artifactId>maven-gpg-plugin</artifactId>
182-
<version>3.2.4</version>
182+
<version>3.2.7</version>
183183
<executions>
184184
<execution>
185185
<id>sign-artifacts</id>

0 commit comments

Comments
 (0)