diff --git a/tarantool-core/src/main/java/io/tarantool/core/state/AbstractDrainLoopMachine.java b/tarantool-core/src/main/java/io/tarantool/core/state/AbstractDrainLoopMachine.java index ed3a423c..dc2ce7e9 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/state/AbstractDrainLoopMachine.java +++ b/tarantool-core/src/main/java/io/tarantool/core/state/AbstractDrainLoopMachine.java @@ -16,12 +16,21 @@ public abstract class AbstractDrainLoopMachine implements State private static final Logger LOGGER = LoggerFactory.getLogger(AbstractDrainLoopMachine.class); + private final Queue controlEvents = new ConcurrentLinkedQueue<>(); private final Queue 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); diff --git a/tarantool-core/src/main/java/io/tarantool/core/state/Event.java b/tarantool-core/src/main/java/io/tarantool/core/state/Event.java index eb53520f..9daa0fa2 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/state/Event.java +++ b/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; + } +} diff --git a/tarantool-core/src/test/java/io/tarantool/core/unit/state/AbstractDrainLoopMachineTest.java b/tarantool-core/src/test/java/io/tarantool/core/unit/state/AbstractDrainLoopMachineTest.java index eb70a7e2..816ad4f7 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/unit/state/AbstractDrainLoopMachineTest.java +++ b/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 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 { private final List processedEvents = Collections.synchronizedList(new ArrayList<>());