refactor(pool): improve drain-loop machine

tarantool-core/src/main/java/io/tarantool/core/state/AbstractDrainLoopMachine.java

@@ -16,12 +16,21 @@ public abstract class AbstractDrainLoopMachine<E extends Event> implements State
 
   private static final Logger LOGGER = LoggerFactory.getLogger(AbstractDrainLoopMachine.class);
 
+  private final Queue<E> controlEvents = new ConcurrentLinkedQueue<>();
   private final Queue<E> events = new ConcurrentLinkedQueue<>();
   private final AtomicInteger wip = new AtomicInteger(0);
 
   @Override
   public void handle(E event) {
-    this.events.offer(event);
+    if (event == null) {
+      return;
+    }
+
+    if (event.isControl()) {
+      this.controlEvents.offer(event);
+    } else {
+      this.events.offer(event);
+    }
 
     if (this.wip.getAndIncrement() != 0) {
       return;
@@ -31,26 +40,40 @@ public void handle(E event) {
     try {
       do {
         E e;
-        while ((e = this.events.poll()) != null) {
-          try {
-            processEvent(e);
-          } catch (Throwable ex) {
-            LOGGER.error("[{}]: unexpected error in event handler. Try handle error...", this, ex);
-            try {
-              onHandleError(ex);
-            } catch (Throwable failSafeEx) {
-              LOGGER.error("[{}]: fail-safe handler failed", this, failSafeEx);
-            }
-          }
+
+        while ((e = this.controlEvents.poll()) != null) {
+          safelyProcess(e);
+        }
+
+        if ((e = this.events.poll()) != null) {
+          safelyProcess(e);
+          // go to check control events again
+          continue;
         }
+
         missed = this.wip.addAndGet(-missed);
       } while (missed != 0);
     } catch (Throwable criticalJvmError) {
+      this.controlEvents.clear();
+      this.events.clear();
       this.wip.set(0);
       throw criticalJvmError;
     }
   }
 
+  private void safelyProcess(E e) {
+    try {
+      processEvent(e);
+    } catch (Exception ex) {
+      LOGGER.error("[{}]: unexpected error in event handler. Try handle error...", this, ex);
+      try {
+        onHandleError(ex);
+      } catch (Exception failSafeEx) {
+        LOGGER.error("[{}]: fail-safe handler failed", this, failSafeEx);
+      }
+    }
+  }
+
   protected abstract void processEvent(E event);
 
   protected abstract void onHandleError(Throwable error);

tarantool-core/src/main/java/io/tarantool/core/state/Event.java

@@ -5,4 +5,9 @@
 
 package io.tarantool.core.state;
 
-public interface Event {}
+public interface Event {
+
+  default boolean isControl() {
+    return false;
+  }
+}

tarantool-core/src/test/java/io/tarantool/core/unit/state/AbstractDrainLoopMachineTest.java

@@ -17,11 +17,17 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.TestInstance.Lifecycle;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
 import org.mockito.Mockito;
 
 import io.tarantool.core.state.AbstractDrainLoopMachine;
 import io.tarantool.core.state.Event;
 
+@Execution(ExecutionMode.CONCURRENT)
+@TestInstance(Lifecycle.PER_METHOD)
 class AbstractDrainLoopMachineTest {
 
   private TestStateMachine stateMachine;
@@ -140,6 +146,55 @@ void testEventSerialization() throws InterruptedException {
     Assertions.assertEquals(2, stateMachine.processedEvents().size());
   }
 
+  @Test
+  void testControlEventsPriority() throws InterruptedException {
+    final Event blockerEvent = Mockito.mock(Event.class);
+    final Event regularEvent = Mockito.mock(Event.class);
+    final Event controlEvent = Mockito.mock(Event.class);
+
+    Mockito.when(blockerEvent.isControl()).thenReturn(false);
+    Mockito.when(regularEvent.isControl()).thenReturn(false);
+    Mockito.when(controlEvent.isControl()).thenReturn(true);
+
+    final CountDownLatch inProcessBlocker = new CountDownLatch(1);
+    final CountDownLatch resumeBlocker = new CountDownLatch(1);
+
+    this.stateMachine.onProcessHook =
+        () -> {
+          if (inProcessBlocker.getCount() > 0) {
+            inProcessBlocker.countDown();
+            try {
+              resumeBlocker.await(2, TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+              Thread.currentThread().interrupt();
+            }
+          }
+        };
+
+    final Thread workerThread = new Thread(() -> this.stateMachine.handle(blockerEvent));
+    workerThread.start();
+
+    Assertions.assertTrue(inProcessBlocker.await(1, TimeUnit.SECONDS));
+
+    this.stateMachine.onProcessHook = null;
+
+    this.stateMachine.handle(regularEvent);
+    this.stateMachine.handle(controlEvent);
+
+    resumeBlocker.countDown();
+    workerThread.join(2000);
+
+    final List<Event> processed = this.stateMachine.processedEvents();
+
+    Assertions.assertEquals(3, processed.size());
+    Assertions.assertSame(blockerEvent, processed.get(0));
+    Assertions.assertSame(
+        controlEvent, processed.get(1), "Control event must bypass regular event due to priority");
+    Assertions.assertSame(regularEvent, processed.get(2));
+
+    Assertions.assertEquals(0, this.stateMachine.errorCount());
+  }
+
   private static class TestStateMachine extends AbstractDrainLoopMachine<Event> {
 
     private final List<Event> processedEvents = Collections.synchronizedList(new ArrayList<>());