Concurrency fixes

Author: scaliby

market-aggregator-core/src/main/java/net/scaliby/marketaggregator/core/common/ConcurrentExecutionsCounter.java

@@ -3,8 +3,8 @@
 public class ConcurrentExecutionsCounter {
 
     private final Object lock = new Object();
-    private volatile int count = 0;
-    private volatile int maxCount = 0;
+    private int count = 0;
+    private int maxCount = 0;
 
     public int getMaxCount() {
         synchronized (lock) {

market-aggregator-core/src/main/java/net/scaliby/marketaggregator/core/concurrency/ChannelScheduler.java

@@ -5,41 +5,28 @@
 import lombok.Setter;
 import lombok.SneakyThrows;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 @RequiredArgsConstructor
 public class ChannelScheduler<K> {
 
     private final Executor executor;
 
-    private final Map<K, ChannelRunner> runners = new HashMap<>();
+    private final Map<K, ChannelRunner> runners = new ConcurrentHashMap<>();
 
     @Setter
     private int queueCapacity = 100_000;
 
-    @SneakyThrows
     public void push(ChannelRunnable<K> job) {
-        synchronized (runners) {
-            boolean alreadyRunning = runners.containsKey(job.getChannel());
-            ChannelRunner channelRunner = runners.computeIfAbsent(job.getChannel(), ChannelRunner::new);
-            channelRunner.push(job);
-            if (!alreadyRunning) {
-                executor.execute(channelRunner);
-            }
-        }
-    }
-
-    private void finalizeQueueRunner(ChannelRunner runner) {
-        synchronized (runners) {
-            if (runner.hasData()) {
-                executor.execute(runner);
-            } else {
-                runners.remove(runner.getKey());
-            }
+        ChannelRunner channelRunner = runners.computeIfAbsent(job.getChannel(), ChannelRunner::new);
+        channelRunner.push(job);
+        if (!channelRunner.isRunning()) {
+            executor.execute(channelRunner);
         }
     }
 
@@ -48,18 +35,30 @@ private final class ChannelRunner implements Runnable {
 
         @Getter
         private final K key;
+        private final AtomicBoolean running = new AtomicBoolean(false);
         private final BlockingQueue<Runnable> workload = new ArrayBlockingQueue<>(queueCapacity);
 
         @Override
         public void run() {
+            boolean oldValue = running.getAndSet(true);
+            if (oldValue) {
+                return;
+            }
             while (hasData()) {
                 Runnable elem = workload.poll();
                 if (elem == null) {
                     break;
                 }
                 elem.run();
             }
-            finalizeQueueRunner(this);
+            running.set(false);
+            if (hasData()) {
+                run();
+            }
+        }
+
+        boolean isRunning() {
+            return running.get();
         }
 
         @SneakyThrows

market-aggregator-core/src/test/java/net/scaliby/marketaggregator/core/concurrency/ChannelSchedulerTest.java

@@ -55,13 +55,34 @@ public void runningTwoJobsForDifferentKeys_executesThemInParallel() throws Inter
         assertEquals(2, counter.getMaxCount());
     }
 
+    @Test(timeout = 2000L)
+    public void runningJobForTheSameKeyTwice_executesItSuccessfullyWithoutDeadlock_forQueueCapacityEqualToOne() throws InterruptedException {
+        // given
+        ExecutorService executorService = Executors.newFixedThreadPool(1);
+        ChannelScheduler<String> channelScheduler = new ChannelScheduler<>(executorService);
+        channelScheduler.setQueueCapacity(1);
+        ConcurrentExecutionsCounter counter = new ConcurrentExecutionsCounter();
+        TestingChannelRunnable firstJob = new TestingChannelRunnable("BTC_USD", counter);
+        TestingChannelRunnable secondJob = new TestingChannelRunnable("ETH_USD", counter);
+
+        // when
+        channelScheduler.push(firstJob);
+        channelScheduler.push(secondJob);
+        channelScheduler.push(secondJob);
+        executorService.awaitTermination(1, TimeUnit.SECONDS);
+
+        // then
+        assertTrue(firstJob.isExecuted());
+        assertTrue(secondJob.isExecuted());
+    }
+
     @RequiredArgsConstructor
     private static class TestingChannelRunnable implements ChannelRunnable<String> {
         @Getter
         private final String channel;
         private final ConcurrentExecutionsCounter concurrentExecutionsCounter;
         @Getter
-        private boolean executed = false;
+        private volatile boolean executed = false;
 
         @Override
         @SneakyThrows