From b4aa066ab563ce7bc5d17e3c98eff52bc5e3de0e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=9D=D0=B8=D0=BA=D0=BE=D0=BB=D0=B0=D0=B9=20=D0=91=D0=B5?= =?UTF-8?q?=D0=BB=D0=BE=D0=BD=D0=BE=D0=B3=D0=BE=D0=B2?= Date: Mon, 2 Nov 2026 18:15:24 +0300 Subject: [PATCH 1/2] feat(core): add abstract drain-loop StateMachine Add realization of drain-loop state machine --- tarantool-core/pom.xml | 5 + .../core/state/AbstractDrainLoopMachine.java | 57 ++++++ .../java/io/tarantool/core/state/Event.java | 8 + .../java/io/tarantool/core/state/State.java | 11 ++ .../io/tarantool/core/state/StateMachine.java | 11 ++ .../state/AbstractDrainLoopMachineTest.java | 176 ++++++++++++++++++ tarantool-java-sdk-bom/pom.xml | 9 + 7 files changed, 277 insertions(+) create mode 100644 tarantool-core/src/main/java/io/tarantool/core/state/AbstractDrainLoopMachine.java create mode 100644 tarantool-core/src/main/java/io/tarantool/core/state/Event.java create mode 100644 tarantool-core/src/main/java/io/tarantool/core/state/State.java create mode 100644 tarantool-core/src/main/java/io/tarantool/core/state/StateMachine.java create mode 100644 tarantool-core/src/test/java/io/tarantool/core/unit/state/AbstractDrainLoopMachineTest.java diff --git a/tarantool-core/pom.xml b/tarantool-core/pom.xml index 7096902a..71c2e09f 100644 --- a/tarantool-core/pom.xml +++ b/tarantool-core/pom.xml @@ -72,6 +72,11 @@ io.tarantool testcontainers + + org.mockito + mockito-junit-jupiter + test + 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 new file mode 100644 index 00000000..ed3a423c --- /dev/null +++ b/tarantool-core/src/main/java/io/tarantool/core/state/AbstractDrainLoopMachine.java @@ -0,0 +1,57 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core.state; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicInteger; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public abstract class AbstractDrainLoopMachine implements StateMachine { + + private static final Logger LOGGER = LoggerFactory.getLogger(AbstractDrainLoopMachine.class); + + private final Queue events = new ConcurrentLinkedQueue<>(); + private final AtomicInteger wip = new AtomicInteger(0); + + @Override + public void handle(E event) { + this.events.offer(event); + + if (this.wip.getAndIncrement() != 0) { + return; + } + + int missed = 1; + 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); + } + } + } + missed = this.wip.addAndGet(-missed); + } while (missed != 0); + } catch (Throwable criticalJvmError) { + this.wip.set(0); + throw criticalJvmError; + } + } + + 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 new file mode 100644 index 00000000..eb53520f --- /dev/null +++ b/tarantool-core/src/main/java/io/tarantool/core/state/Event.java @@ -0,0 +1,8 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core.state; + +public interface Event {} diff --git a/tarantool-core/src/main/java/io/tarantool/core/state/State.java b/tarantool-core/src/main/java/io/tarantool/core/state/State.java new file mode 100644 index 00000000..dea426eb --- /dev/null +++ b/tarantool-core/src/main/java/io/tarantool/core/state/State.java @@ -0,0 +1,11 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core.state; + +public interface State { + + void handle(E event); +} diff --git a/tarantool-core/src/main/java/io/tarantool/core/state/StateMachine.java b/tarantool-core/src/main/java/io/tarantool/core/state/StateMachine.java new file mode 100644 index 00000000..181bf99f --- /dev/null +++ b/tarantool-core/src/main/java/io/tarantool/core/state/StateMachine.java @@ -0,0 +1,11 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core.state; + +public interface StateMachine { + + void handle(E event); +} 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 new file mode 100644 index 00000000..eb70a7e2 --- /dev/null +++ b/tarantool-core/src/test/java/io/tarantool/core/unit/state/AbstractDrainLoopMachineTest.java @@ -0,0 +1,176 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core.unit.state; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import io.tarantool.core.state.AbstractDrainLoopMachine; +import io.tarantool.core.state.Event; + +class AbstractDrainLoopMachineTest { + + private TestStateMachine stateMachine; + + @BeforeEach + void setUp() { + stateMachine = new TestStateMachine(); + } + + @Test + void testSequentialProcessing() { + final Event event1 = Mockito.mock(Event.class); + final Event event2 = Mockito.mock(Event.class); + + this.stateMachine.handle(event1); + this.stateMachine.handle(event2); + + final List processed = this.stateMachine.processedEvents(); + Assertions.assertEquals(2, processed.size()); + Assertions.assertSame(event1, processed.get(0)); + Assertions.assertSame(event2, processed.get(1)); + Assertions.assertEquals(0, this.stateMachine.errorCount()); + } + + @Test + void testRecoveryAfterException() { + final Event event1 = Mockito.mock(Event.class); + final Event event2 = Mockito.mock(Event.class); + final Event event3 = Mockito.mock(Event.class); + + this.stateMachine.handle(event1); + this.stateMachine.throwOnEvent = new RuntimeException("Simulation error"); + + this.stateMachine.handle(event2); + this.stateMachine.handle(event3); + + Assertions.assertEquals(1, this.stateMachine.errorCount()); + final List processed = this.stateMachine.processedEvents(); + Assertions.assertEquals(2, processed.size()); + Assertions.assertSame(event1, processed.get(0)); + Assertions.assertSame(event3, processed.get(1)); + } + + @Test + void testConcurrentDrainLoop() throws InterruptedException { + int threadCount = 10; + int eventsPerThread = 10000; + int totalEvents = threadCount * eventsPerThread; + + final ExecutorService executor = Executors.newFixedThreadPool(threadCount); + final CountDownLatch startLatch = new CountDownLatch(1); + final CountDownLatch finishLatch = new CountDownLatch(threadCount); + + // Генерируем пачку потоков, которые будут одновременно бомбардировать машину событиями + for (int i = 0; i < threadCount; i++) { + executor.submit( + () -> { + try { + startLatch.await(); + for (int j = 0; j < eventsPerThread; j++) { + this.stateMachine.handle(Mockito.mock(Event.class)); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + finishLatch.countDown(); + } + }); + } + + startLatch.countDown(); // Даем старт всем потокам одновременно + boolean completed = finishLatch.await(5, TimeUnit.SECONDS); + executor.shutdown(); + + Assertions.assertTrue(completed); + Assertions.assertEquals(totalEvents, stateMachine.processedEvents().size()); + Assertions.assertEquals(0, stateMachine.errorCount()); + } + + @Test + void testEventSerialization() throws InterruptedException { + final Event event1 = Mockito.mock(Event.class); + final Event event2 = Mockito.mock(Event.class); + + final CountDownLatch inProcessEvent1 = new CountDownLatch(1); + final CountDownLatch proceedEvent1 = new CountDownLatch(1); + final AtomicInteger activeThreadsInHandler = new AtomicInteger(0); + + this.stateMachine.onProcessHook = + () -> { + activeThreadsInHandler.incrementAndGet(); + inProcessEvent1.countDown(); + try { + proceedEvent1.await(2, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + activeThreadsInHandler.decrementAndGet(); + } + }; + + final Thread t1 = new Thread(() -> stateMachine.handle(event1)); + t1.start(); + + Assertions.assertTrue(inProcessEvent1.await(1, TimeUnit.SECONDS)); + + final long startTime = System.currentTimeMillis(); + this.stateMachine.handle(event2); + final long duration = System.currentTimeMillis() - startTime; + + Assertions.assertTrue(duration < 500); + + proceedEvent1.countDown(); + t1.join(2000); + + Assertions.assertEquals(2, stateMachine.processedEvents().size()); + } + + private static class TestStateMachine extends AbstractDrainLoopMachine { + + private final List processedEvents = Collections.synchronizedList(new ArrayList<>()); + private final AtomicInteger errorCount = new AtomicInteger(0); + private volatile Runnable onProcessHook = null; + private volatile RuntimeException throwOnEvent = null; + + @Override + protected void processEvent(Event event) { + if (throwOnEvent != null) { + final RuntimeException ex = throwOnEvent; + throwOnEvent = null; + throw ex; + } + if (onProcessHook != null) { + onProcessHook.run(); + } + processedEvents.add(event); + } + + @Override + protected void onHandleError(Throwable error) { + this.errorCount.incrementAndGet(); + } + + public List processedEvents() { + return this.processedEvents; + } + + public int errorCount() { + return this.errorCount.get(); + } + } +} diff --git a/tarantool-java-sdk-bom/pom.xml b/tarantool-java-sdk-bom/pom.xml index c0c2be06..171f1827 100644 --- a/tarantool-java-sdk-bom/pom.xml +++ b/tarantool-java-sdk-bom/pom.xml @@ -34,6 +34,7 @@ 5.6.0 1.3.3 1.67.0 + 4.8.1 @@ -242,6 +243,14 @@ ${instancio.version} test + + + org.mockito + mockito-bom + ${mockito.version} + pom + import + From f0ff5dec9fb9a88d3622ed872ed0d733044bf967 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=9D=D0=B8=D0=BA=D0=BE=D0=BB=D0=B0=D0=B9=20=D0=91=D0=B5?= =?UTF-8?q?=D0=BB=D0=BE=D0=BD=D0=BE=D0=B3=D0=BE=D0=B2?= Date: Mon, 2 Nov 2026 22:34:45 +0300 Subject: [PATCH 2/2] feat(core): add abstract Netty based StateMachine Add realization of Netty based state machine --- .../core/state/AbstractNettyMachine.java | 61 +++++++ .../core/state/AbstractNettyMachineTest.java | 151 ++++++++++++++++++ 2 files changed, 212 insertions(+) create mode 100644 tarantool-core/src/main/java/io/tarantool/core/state/AbstractNettyMachine.java create mode 100644 tarantool-core/src/test/java/io/tarantool/core/state/AbstractNettyMachineTest.java diff --git a/tarantool-core/src/main/java/io/tarantool/core/state/AbstractNettyMachine.java b/tarantool-core/src/main/java/io/tarantool/core/state/AbstractNettyMachine.java new file mode 100644 index 00000000..59942756 --- /dev/null +++ b/tarantool-core/src/main/java/io/tarantool/core/state/AbstractNettyMachine.java @@ -0,0 +1,61 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core.state; + +import java.util.concurrent.RejectedExecutionException; + +import io.netty.channel.EventLoop; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public abstract class AbstractNettyMachine implements StateMachine { + + private static final Logger LOGGER = LoggerFactory.getLogger(AbstractNettyMachine.class); + + protected final EventLoop loop; + + protected AbstractNettyMachine(EventLoop loop) { + this.loop = loop; + } + + @Override + public void handle(E event) { + if (this.loop.inEventLoop()) { + internalHandle(event); + } else { + try { + this.loop.execute(() -> internalHandle(event)); + } catch (RejectedExecutionException ex) { + LOGGER.error( + "[{}]: catch an unexpected machine event execution. Try handle this...", this, ex); + try { + onHandleRejectedException(event, ex); + } catch (Throwable failSafeEx) { + LOGGER.error("[{}]: fail-safe handler failed", this, failSafeEx); + } + } + } + } + + private void internalHandle(E event) { + try { + processEvent(event); + } catch (Throwable error) { + try { + LOGGER.error("[{}]: unexpected error in event handler. Try handle error...", this, error); + onHandleError(error); + } catch (Throwable failSafeEx) { + LOGGER.error("[{}]: fail-safe handler failed", this, failSafeEx); + } + } + } + + protected abstract void processEvent(E event); + + protected abstract void onHandleError(Throwable error); + + protected abstract void onHandleRejectedException(E event, RejectedExecutionException e); +} diff --git a/tarantool-core/src/test/java/io/tarantool/core/state/AbstractNettyMachineTest.java b/tarantool-core/src/test/java/io/tarantool/core/state/AbstractNettyMachineTest.java new file mode 100644 index 00000000..d3cb2f22 --- /dev/null +++ b/tarantool-core/src/test/java/io/tarantool/core/state/AbstractNettyMachineTest.java @@ -0,0 +1,151 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core.state; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.RejectedExecutionException; + +import io.netty.channel.EventLoop; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +class AbstractNettyMachineTest { + + private EventLoop mockLoop; + private TestNettyMachine machine; + + @BeforeEach + void setUp() { + mockLoop = Mockito.mock(EventLoop.class); + machine = new TestNettyMachine(mockLoop); + } + + @Test + void testHandleInEventLoop() { + Mockito.when(mockLoop.inEventLoop()).thenReturn(true); + final Event event = Mockito.mock(Event.class); + + machine.handle(event); + + Assertions.assertEquals(1, machine.processedEvents.size()); + Assertions.assertSame(event, machine.processedEvents.get(0)); + Mockito.verify(mockLoop, Mockito.never()).execute(Mockito.any(Runnable.class)); + } + + @Test + void testHandleOutsideEventLoop() { + Mockito.when(mockLoop.inEventLoop()).thenReturn(false); + final Event event = Mockito.mock(Event.class); + + machine.handle(event); + + Assertions.assertEquals(0, machine.processedEvents.size()); + + final ArgumentCaptor runnableCaptor = ArgumentCaptor.forClass(Runnable.class); + Mockito.verify(mockLoop, Mockito.times(1)).execute(runnableCaptor.capture()); + + runnableCaptor.getValue().run(); + + Assertions.assertEquals(1, machine.processedEvents.size()); + Assertions.assertSame(event, machine.processedEvents.get(0)); + } + + @Test + void testRecoveryAfterThrowable() { + Mockito.when(mockLoop.inEventLoop()).thenReturn(true); + final Event event = Mockito.mock(Event.class); + final RuntimeException expectedException = new RuntimeException("Сбой бизнес-логики"); + + machine.onProcessHook = + () -> { + throw expectedException; + }; + + Assertions.assertDoesNotThrow(() -> machine.handle(event)); + + Assertions.assertEquals(1, machine.caughtErrors.size()); + Assertions.assertSame(expectedException, machine.caughtErrors.get(0)); + Assertions.assertEquals(0, machine.processedEvents.size()); + } + + @Test + void testHandleRejectedExecutionException() { + Mockito.when(mockLoop.inEventLoop()).thenReturn(false); + Mockito.doThrow(new RejectedExecutionException("Loop is shutdown")) + .when(mockLoop) + .execute(Mockito.any(Runnable.class)); + + final Event event = Mockito.mock(Event.class); + + Assertions.assertDoesNotThrow(() -> machine.handle(event)); + + // Проверяем, что сработал наш новый безопасный обработчик отказа и контекст события НЕ потерян + Assertions.assertEquals(1, machine.rejectedEvents.size()); + Assertions.assertSame(event, machine.rejectedEvents.get(0)); + Assertions.assertEquals(0, machine.processedEvents.size()); + } + + @Test + void testFailSafeHandlerProtectsAgainstDoubleFault() { + Mockito.when(mockLoop.inEventLoop()).thenReturn(true); + final Event event = Mockito.mock(Event.class); + + machine.onProcessHook = + () -> { + throw new RuntimeException("First error"); + }; + TestNettyMachine brokenMachine = + new TestNettyMachine(mockLoop) { + @Override + protected void processEvent(Event event) { + throw new RuntimeException("First error"); + } + + @Override + protected void onHandleError(Throwable error) { + throw new RuntimeException("Second error onError inside"); + } + }; + + Assertions.assertDoesNotThrow(() -> brokenMachine.handle(event)); + } + + private static class TestNettyMachine extends AbstractNettyMachine { + + final List processedEvents = Collections.synchronizedList(new ArrayList<>()); + final List caughtErrors = Collections.synchronizedList(new ArrayList<>()); + final List rejectedEvents = Collections.synchronizedList(new ArrayList<>()); + + volatile Runnable onProcessHook = null; + + protected TestNettyMachine(EventLoop loop) { + super(loop); + } + + @Override + protected void processEvent(Event event) { + if (onProcessHook != null) { + onProcessHook.run(); + } + processedEvents.add(event); + } + + @Override + protected void onHandleError(Throwable error) { + caughtErrors.add(error); + } + + @Override + protected void onHandleRejectedException(Event event, RejectedExecutionException e) { + rejectedEvents.add(event); + } + } +}