feat: adds EventBus and introduces Mpsc as alternative of disruptor - #1238
feat: adds EventBus and introduces Mpsc as alternative of disruptor#1238killme2008 wants to merge 4 commits into
Conversation
Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
There was a problem hiding this comment.
Pull request overview
This PR introduces an EventBus abstraction layer to provide an alternative to the existing Disruptor-based event processing, addressing cross-generation reference issues in generational garbage collectors (Issue #1231). The implementation adds a new MPSC (Multi-Producer Single-Consumer) queue-based EventBus alongside the existing Disruptor implementation, wrapped behind a common interface.
Key Changes:
- Introduces EventBus abstraction with two implementations: DisruptorEventBus (object reuse) and MpscEventBus (no object reuse)
- Adds EventBusMode configuration option (DISRUPTOR/MPSC) to RaftOptions for selecting the implementation
- Refactors LogManagerImpl, NodeImpl, FSMCallerImpl, and ReadOnlyServiceImpl to use the new EventBus abstraction
- Provides comprehensive test coverage including parameterized tests across both implementations
Reviewed changes
Copilot reviewed 19 out of 20 changed files in this pull request and generated 8 comments.
Show a summary per file
| File | Description |
|---|---|
| EventBus.java | Core interface defining the event bus contract with publish, shutdown, and capacity management methods |
| EventBusHandler.java | Functional interface for event processing with endOfBatch signaling |
| EventBusOptions.java | Configuration class for EventBus instances with fluent API and validation |
| EventBusMode.java | Enum defining DISRUPTOR and MPSC modes with documentation on GC compatibility |
| EventBusFactory.java | Factory for creating EventBus instances based on configured mode |
| WaitStrategyType.java | Enum for consumer thread wait strategies (BLOCKING, TIMEOUT_BLOCKING) |
| DisruptorEventBus.java | Disruptor-based implementation wrapping existing RingBuffer functionality |
| MpscEventBus.java | New MPSC queue-based implementation using JCTools without object reuse |
| LogManagerImpl.java | Refactored to use EventBus instead of direct Disruptor usage for disk operations |
| NodeImpl.java | Refactored to use EventBus for log entry application queue |
| FSMCallerImpl.java | Refactored to use EventBus for FSM task queue |
| ReadOnlyServiceImpl.java | Refactored to use EventBus for read-index request queue |
| RaftOptions.java | Added eventBusMode configuration field with DISRUPTOR as default |
| FSMCallerOptions.java | Added eventBusMode field for FSMCaller configuration |
| LogManager.java | Minor whitespace fix |
| EventBusTest.java | Parameterized tests covering both implementations with concurrent scenarios |
| MpscEventBusTest.java | MPSC-specific tests for maxBatchSize behavior and shutdown draining |
| EventBusOptionsTest.java | Comprehensive tests for configuration options and validation |
| EventBusFactoryTest.java | Tests for factory creation logic and mode selection |
| NodeEventBusModeTest.java | Integration tests exercising Node operations with both EventBus modes |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| ThreadPoolsFactory.runInThread(this.node.getGroupId(), () -> { | ||
| final ReadIndexEvent event = new ReadIndexEvent(); | ||
| event.shutdownLatch = this.shutdownLatch; | ||
| this.readIndexEventBus.publish(event); |
There was a problem hiding this comment.
Similar to LogManagerImpl, the shutdown event publish call lacks error handling. If readIndexEventBus.publish() fails, the shutdownLatch will never be counted down, causing join() to hang indefinitely. Consider checking the return value or implementing a timeout mechanism.
| () -> this.applyQueue.publishEvent((event, sequence) -> event.shutdownLatch = latch)); | ||
| final LogEntryAndClosure shutdownEvent = new LogEntryAndClosure(); | ||
| shutdownEvent.shutdownLatch = latch; | ||
| ThreadPoolsFactory.runInThread(this.groupId, () -> this.applyEventBus.publish(shutdownEvent)); |
There was a problem hiding this comment.
The shutdown event publish call lacks error handling. If applyEventBus.publish() fails, the shutdownLatch will never be counted down, causing join() to hang indefinitely. Consider checking the return value or implementing a timeout mechanism.
| ThreadPoolsFactory.runInThread(this.groupId, () -> this.applyEventBus.publish(shutdownEvent)); | |
| ThreadPoolsFactory.runInThread(this.groupId, () -> { | |
| try { | |
| this.applyEventBus.publish(shutdownEvent); | |
| } catch (final Throwable t) { | |
| // Ensure that shutdown does not hang if the shutdown event cannot be published. | |
| LOG.error("Failed to publish shutdown event on applyEventBus for node {}.", this.nodeId, t); | |
| latch.countDown(); | |
| final int num = GLOBAL_NUM_NODES.decrementAndGet(); | |
| LOG.info("The number of active nodes decrement to {}.", num); | |
| } | |
| }); |
| task.type = TaskType.SHUTDOWN; | ||
| task.shutdownLatch = latch; | ||
| })); | ||
| this.taskEventBus.publish(task); |
There was a problem hiding this comment.
The shutdown event publish call lacks error handling. If taskEventBus.publish() fails, the shutdownLatch will never be counted down, causing join() to hang indefinitely. Consider checking the return value or implementing a timeout mechanism.
| this.taskEventBus.publish(task); | |
| try { | |
| final boolean published = this.taskEventBus.publish(task); | |
| if (!published) { | |
| LOG.error("Failed to publish SHUTDOWN task to taskEventBus, counting down latch directly."); | |
| latch.countDown(); | |
| } | |
| } catch (final Exception e) { | |
| LOG.error("Exception while publishing SHUTDOWN task to taskEventBus, counting down latch directly.", e); | |
| latch.countDown(); | |
| } |
| this.consumerThread = opts.getThreadFactory().newThread(() -> consumeLoop(handler)); | ||
| this.consumerThread.start(); |
There was a problem hiding this comment.
The MpscEventBus constructor starts the consumer thread before the constructor completes (line 86), which could lead to the consumer thread accessing fields before they are fully initialized. While the current implementation may work due to the specific field initialization order, this pattern is generally unsafe. Consider deferring thread start until after the constructor completes, or ensure all fields are initialized before starting the thread.
| final int initialCapacity = Math.min(INITIAL_CAPACITY, this.bufferSize / 2); | ||
| this.queue = new MpscGrowableAtomicArrayQueue<>(Math.max(2, initialCapacity), this.bufferSize); |
There was a problem hiding this comment.
The queue initialization at line 83 uses integer division for initialCapacity calculation. When bufferSize is small (e.g., 2 or 3), bufferSize / 2 will be 1 or 1 respectively, but the Math.max(2, initialCapacity) ensures at least 2. However, for bufferSize=1, this would result in initialCapacity=2 which is greater than maxCapacity=1, violating the MpscGrowableAtomicArrayQueue requirement that initialCapacity must be less than maxCapacity. Consider adding validation to ensure bufferSize is at least 2, or adjust the calculation to handle edge cases.
| final int initialCapacity = Math.min(INITIAL_CAPACITY, this.bufferSize / 2); | |
| this.queue = new MpscGrowableAtomicArrayQueue<>(Math.max(2, initialCapacity), this.bufferSize); | |
| final int maxCapacity = this.bufferSize; | |
| Requires.requireTrue(maxCapacity > 1, "bufferSize must be greater than 1"); | |
| int initialCapacity = Math.min(INITIAL_CAPACITY, maxCapacity / 2); | |
| // Ensure initialCapacity is strictly less than maxCapacity | |
| if (initialCapacity >= maxCapacity) { | |
| initialCapacity = maxCapacity - 1; | |
| } | |
| // Prefer a minimum initial capacity of 2 when possible (and still < maxCapacity) | |
| if (initialCapacity < 2 && maxCapacity >= 3) { | |
| initialCapacity = 2; | |
| } | |
| this.queue = new MpscGrowableAtomicArrayQueue<>(initialCapacity, maxCapacity); |
| // Queue is empty | ||
| if (this.shutdown) { | ||
| // Shutdown requested and queue is empty, exit | ||
| return; | ||
| } | ||
| // Reset batch counter and wait for new events |
There was a problem hiding this comment.
The shutdown logic has a potential race condition. The consumer thread checks this.shutdown at line 162 to exit when the queue is empty, but this check happens before the ShutdownEvent is processed. If the queue is empty when shutdown is called, the consumer might exit before seeing the ShutdownEvent, causing the shutdown latch to never be counted down. Consider checking for shutdown only after processing ShutdownEvent, or ensure the ShutdownEvent is added before setting the shutdown flag.
| // Queue is empty | |
| if (this.shutdown) { | |
| // Shutdown requested and queue is empty, exit | |
| return; | |
| } | |
| // Reset batch counter and wait for new events | |
| // Queue is empty, reset batch counter and wait for new events |
| public void shutdown(final CountDownLatch latch) { | ||
| this.shutdown = true; | ||
| this.ringBuffer.publishEvent((wrapper, seq) -> wrapper.shutdownLatch = latch); | ||
| } |
There was a problem hiding this comment.
The DisruptorEventBus shutdown method may block indefinitely if the RingBuffer is full. After setting this.shutdown = true, the call to publishEvent on line 127 will block if there's no capacity, but since shutdown is already set to true, any new publish attempts will fail (line 93-96), preventing the queue from draining. Consider using tryPublishEvent with a timeout or loop, or ensure there's always capacity for the shutdown event.
| })); | ||
| final StableClosureEvent shutdownEvent = new StableClosureEvent(); | ||
| shutdownEvent.type = EventType.SHUTDOWN; | ||
| ThreadPoolsFactory.runInThread(this.groupId, () -> this.diskEventBus.publish(shutdownEvent)); |
There was a problem hiding this comment.
The shutdown event publish call lacks proper error handling. If diskEventBus.publish() fails (e.g., if the event bus is already shut down or the queue is full), the shutDownLatch will never be counted down, causing join() to hang indefinitely. Consider adding a check for the return value or using a timeout mechanism.
| ThreadPoolsFactory.runInThread(this.groupId, () -> this.diskEventBus.publish(shutdownEvent)); | |
| ThreadPoolsFactory.runInThread(this.groupId, () -> { | |
| final EventBus eventBus = this.diskEventBus; | |
| if (eventBus == null) { | |
| // Event bus already shut down or not available; avoid hanging join(). | |
| if (this.shutDownLatch != null) { | |
| this.shutDownLatch.countDown(); | |
| } | |
| return; | |
| } | |
| try { | |
| final boolean published = eventBus.publish(shutdownEvent); | |
| if (!published) { | |
| // Failed to enqueue shutdown event; ensure join() does not block indefinitely. | |
| LOG.warn("Failed to publish shutdown event for group {} because the event bus rejected the event.", | |
| this.groupId); | |
| if (this.shutDownLatch != null) { | |
| this.shutDownLatch.countDown(); | |
| } | |
| } | |
| } catch (final Throwable t) { | |
| // Any unexpected error while publishing the shutdown event should not prevent shutdown completion. | |
| LOG.warn("Exception occurred while publishing shutdown event for group {}.", this.groupId, t); | |
| if (this.shutDownLatch != null) { | |
| this.shutDownLatch.countDown(); | |
| } | |
| } | |
| }); |
|
Important Review skippedDraft detected. Please check the settings in the CodeRabbit UI or the You can disable this status message by setting the Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. 📝 WalkthroughWalkthroughReplaces Disruptor-only task queues with a configurable EventBus abstraction (DISRUPTOR or MPSC). Adds EventBus core APIs, two implementations, options and factory, and migrates FSMCaller, Node, ReadOnlyService, and LogManager to publish/consume via EventBus instead of Disruptor. Changes
Sequence Diagram(s)mermaid Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (7)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/MpscEventBus.java (2)
82-83: Critical: Queue initialization violates MpscGrowableAtomicArrayQueue requirements for bufferSize < 3.The calculation can produce
initialCapacity >= maxCapacitywhenbufferSizeis 1 or 2. For example, withbufferSize=1:initialCapacity = Math.min(1024, 0) = 0, thenMath.max(2, 0) = 2, resulting ininitialCapacity=2 > maxCapacity=1. This violates the MpscGrowableAtomicArrayQueue requirement that initial capacity must be strictly less than max capacity.🔎 Apply this diff to add validation and fix the edge case:
+ Requires.requireTrue(this.bufferSize >= 2, "bufferSize must be at least 2"); // Use bounded MPSC queue // initialCapacity must be less than maxCapacity for MpscGrowableAtomicArrayQueue final int initialCapacity = Math.min(INITIAL_CAPACITY, this.bufferSize / 2); - this.queue = new MpscGrowableAtomicArrayQueue<>(Math.max(2, initialCapacity), this.bufferSize); + final int actualInitialCapacity = Math.min(Math.max(2, initialCapacity), this.bufferSize - 1); + this.queue = new MpscGrowableAtomicArrayQueue<>(actualInitialCapacity, this.bufferSize);
138-177: Critical: Shutdown race condition can prevent graceful termination.There's a race between setting
this.shutdown = true(line 140) and adding theShutdownEventto the queue (lines 142-144). If the consumer thread polls and sees an empty queue aftershutdownis set but before theShutdownEventis enqueued, it will exit at line 162-164 without processing theShutdownEvent, causing the shutdown latch to never be counted down.🔎 Apply this diff to fix the race condition:
@Override public void shutdown(final CountDownLatch latch) { - this.shutdown = true; // Spin until ShutdownEvent is successfully added while (!this.queue.offer(new ShutdownEvent(latch))) { LockSupport.parkNanos(SPIN_PARK_NANOS); } + this.shutdown = true; LockSupport.unpark(this.consumerThread); }Or alternatively, modify the consumer logic:
if (item == null) { // Queue is empty + // Reset batch counter and wait for new events + processedInBatch = 0; + parkWait(); + continue; + } + + if (item instanceof ShutdownEvent) { + LOG.info("EventBus '{}' received shutdown signal, draining remaining events", this.name); + drainAndProcess(handler); + ((ShutdownEvent) item).latch.countDown(); + return; + } + + // Check shutdown after processing ShutdownEvent + if (this.shutdown) { if (this.shutdown) { // Shutdown requested and queue is empty, exit return; } - // Reset batch counter and wait for new events - processedInBatch = 0; - parkWait(); - continue; }jraft-core/src/main/java/com/alipay/sofa/jraft/storage/impl/LogManagerImpl.java (1)
232-237: Major: Shutdown can hang indefinitely if event publish fails.If
diskEventBus.publish()returns false (e.g., bus already shut down or queue full) or throws an exception, theshutDownLatchwill never be counted down, causingjoin()to hang indefinitely. The async execution viaThreadPoolsFactory.runInThread()makes this harder to detect.🔎 Apply this diff to add error handling:
private void stopDiskThread() { this.shutDownLatch = new CountDownLatch(1); final StableClosureEvent shutdownEvent = new StableClosureEvent(); shutdownEvent.type = EventType.SHUTDOWN; - ThreadPoolsFactory.runInThread(this.groupId, () -> this.diskEventBus.publish(shutdownEvent)); + ThreadPoolsFactory.runInThread(this.groupId, () -> { + try { + final boolean published = this.diskEventBus.publish(shutdownEvent); + if (!published) { + LOG.warn("Failed to publish shutdown event for group {}, counting down latch anyway", this.groupId); + this.shutDownLatch.countDown(); + } + } catch (final Throwable t) { + LOG.error("Exception while publishing shutdown event for group {}, counting down latch", this.groupId, t); + this.shutDownLatch.countDown(); + } + }); }jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/DisruptorEventBus.java (1)
124-128: Shutdown may block indefinitely if RingBuffer is full.After setting
this.shutdown = true, thepublishEventcall on line 127 will block if there's no capacity. However, sincepublish()now returnsfalsewhen shutdown is true (lines 93-96), no new events will drain the queue, creating a potential deadlock.Consider using
tryPublishEventwith retry/timeout logic, or ensure capacity exists before the blocking publish.🔎 Suggested fix using tryPublishEvent with spin
@Override public void shutdown(final CountDownLatch latch) { this.shutdown = true; - this.ringBuffer.publishEvent((wrapper, seq) -> wrapper.shutdownLatch = latch); + // Use tryPublishEvent to avoid blocking indefinitely if buffer is full + while (!this.ringBuffer.tryPublishEvent((wrapper, seq) -> wrapper.shutdownLatch = latch)) { + Thread.yield(); + } }jraft-core/src/main/java/com/alipay/sofa/jraft/core/ReadOnlyServiceImpl.java (1)
284-288: Shutdown publish lacks error handling —shutdownLatchmay never count down.If
readIndexEventBus.publish(event)fails (returnsfalseor throws), theshutdownLatchwill never be counted down, causingjoin()to hang indefinitely.🔎 Apply this diff to handle publish failures:
ThreadPoolsFactory.runInThread(this.node.getGroupId(), () -> { final ReadIndexEvent event = new ReadIndexEvent(); event.shutdownLatch = this.shutdownLatch; - this.readIndexEventBus.publish(event); + if (!this.readIndexEventBus.publish(event)) { + LOG.warn("Failed to publish shutdown event, counting down latch directly."); + this.shutdownLatch.countDown(); + } });jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java (1)
197-202: Shutdown publish lacks error handling —shutdownLatchmay never count down.If
taskEventBus.publish(task)fails, theshutdownLatchwill never be counted down, causingjoin()to hang indefinitely.🔎 Apply this diff to handle publish failures:
ThreadPoolsFactory.runInThread(getNode().getGroupId(), () -> { final ApplyTask task = new ApplyTask(); task.type = TaskType.SHUTDOWN; task.shutdownLatch = latch; - this.taskEventBus.publish(task); + if (!this.taskEventBus.publish(task)) { + LOG.error("Failed to publish SHUTDOWN task, counting down latch directly."); + latch.countDown(); + } });jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java (1)
2813-2818: Critical: Unaddressed past review - shutdown event publish lacks error handling.This issue was previously flagged but remains unresolved. If
applyEventBus.publish(shutdownEvent)fails or throws an exception, theshutdownLatchwill never be counted down, causingjoin()at line 2905 to hang indefinitely.🔎 Apply this fix to handle publish failures:
if (this.applyEventBus != null) { final CountDownLatch latch = new CountDownLatch(1); this.shutdownLatch = latch; final LogEntryAndClosure shutdownEvent = new LogEntryAndClosure(); shutdownEvent.shutdownLatch = latch; - ThreadPoolsFactory.runInThread(this.groupId, () -> this.applyEventBus.publish(shutdownEvent)); + ThreadPoolsFactory.runInThread(this.groupId, () -> { + try { + this.applyEventBus.publish(shutdownEvent); + } catch (final Throwable t) { + LOG.error("Failed to publish shutdown event for node {}.", getNodeId(), t); + latch.countDown(); + final int num = GLOBAL_NUM_NODES.decrementAndGet(); + LOG.info("The number of active nodes decrement to {}.", num); + } + }); } else {
🧹 Nitpick comments (9)
jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusOptionsTest.java (1)
192-194: Optional: Simplify assertion helper.The private
assertTrue()wrapper is unnecessary since you can use a static import or callorg.junit.Assert.assertTrue()directly. This adds a minor indirection without clear benefit.🔎 Consider this simpler approach:
+import static org.junit.Assert.assertTrue; + @Test public void testToString() { final EventBusOptions opts = new EventBusOptions().setMode(EventBusMode.MPSC).setName("test-bus"); final String str = opts.toString(); assertNotNull(str); assertTrue(str.contains("MPSC")); assertTrue(str.contains("test-bus")); } - - private void assertTrue(final boolean condition) { - org.junit.Assert.assertTrue(condition); - }jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactoryTest.java (1)
79-96: Consider asserting the await() result in testFactoryRespectsMode.Lines 87 and 95 call
latch.await()but don't assert the result, unlike lines 48 and 65 which do assert. This could mask test failures if shutdown doesn't complete in time.🔎 Suggested fix
CountDownLatch latch = new CountDownLatch(1); bus.shutdown(latch); - latch.await(5, TimeUnit.SECONDS); + assertTrue(latch.await(5, TimeUnit.SECONDS)); // Test MPSC mode opts = new EventBusOptions().setMode(EventBusMode.MPSC); bus = EventBusFactory.create(opts, (e, b) -> {}); assertTrue("Expected MpscEventBus for MPSC mode", bus instanceof MpscEventBus); latch = new CountDownLatch(1); bus.shutdown(latch); - latch.await(5, TimeUnit.SECONDS); + assertTrue(latch.await(5, TimeUnit.SECONDS));jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/MpscEventBusTest.java (1)
159-169: Spin-wait without timeout could hang indefinitely.The
whileloop on lines 166-168 spins untileventCount.get() >= 3without any timeout. If the test fails for some reason, this will hang indefinitely rather than failing with a clear error.🔎 Suggested fix with timeout
// Wait for all events to be processed - while (eventCount.get() < 3) { - Thread.sleep(10); + long deadline = System.currentTimeMillis() + 5000; + while (eventCount.get() < 3) { + if (System.currentTimeMillis() > deadline) { + throw new AssertionError("Timeout waiting for events to be processed"); + } + Thread.sleep(10); }jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusTest.java (2)
165-166: Thread.sleep for synchronization can cause flaky tests.Using
Thread.sleep(100)to wait for "event processing to complete" is timing-dependent and may fail on slow CI environments. Consider using a more deterministic synchronization mechanism like aCountDownLatchthat the handler counts down after setting the flag.
251-274: Consider ensuringblockingBusis properly shut down in all test paths.If an assertion fails before the
blockingBus.shutdown()call (e.g., line 265), the bus and its consumer thread may leak. Thefinallyblock only unblocks the consumer but doesn't shut down the bus if shutdown wasn't reached.🔎 Suggested improvement:
try { // ... test logic ... } finally { blockLatch.countDown(); // Ensure unblocked even if test fails + if (!blockingBus.isShutdown()) { + final CountDownLatch cleanupLatch = new CountDownLatch(1); + blockingBus.shutdown(cleanupLatch); + cleanupLatch.await(5, TimeUnit.SECONDS); + } }jraft-core/src/test/java/com/alipay/sofa/jraft/core/NodeEventBusModeTest.java (1)
203-242: Consider using lambda or named Runnable instead of anonymous Thread subclass.The anonymous
Threadsubclass pattern (new Thread() { @Override public void run() {...} }) is verbose. Usingnew Thread(() -> {...})or extracting the logic would improve readability.jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusOptions.java (2)
43-45: Default ThreadFactory uses initial name, won't reflect subsequentsetName()calls.The constructor creates a
NamedThreadFactoryusingthis.name + "-". If a user callssetName()after construction but beforesetThreadFactory(), the thread factory prefix won't match the updated name. This is a minor inconsistency.🔎 Option 1: Lazy-initialize threadFactory in getter:
public EventBusOptions() { - this.threadFactory = new NamedThreadFactory(this.name + "-", true); + // threadFactory will be lazily initialized in getThreadFactory() if not set } public ThreadFactory getThreadFactory() { + if (this.threadFactory == null) { + this.threadFactory = new NamedThreadFactory(this.name + "-", true); + } return this.threadFactory; }🔎 Option 2: Document that setThreadFactory should be called after setName:
Add Javadoc noting the order dependency.
81-85: Consider enforcing power-of-2 validation for Disruptor mode's bufferSize.The Javadoc notes that for Disruptor mode,
bufferSize"must be power of 2", but the setter only validatesbufferSize > 0. While the Disruptor implementation likely enforces this, validating here would provide earlier, clearer error messages.jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java (1)
1629-1630: Consider reformatting for readability.The closure invocation is split across two lines in an inconsistent manner. Consider reformatting for better readability.
🔎 Suggested formatting improvement:
- ThreadPoolsFactory.runClosureInThread(this.groupId, task.getDone(), new Status(RaftError.ENODESHUTDOWN, - "Node is shutting down.")); + ThreadPoolsFactory.runClosureInThread(this.groupId, task.getDone(), + new Status(RaftError.ENODESHUTDOWN, "Node is shutting down."));
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (20)
jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java(7 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java(9 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/core/ReadOnlyServiceImpl.java(10 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/option/FSMCallerOptions.java(3 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/option/RaftOptions.java(5 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/storage/LogManager.java(1 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/storage/impl/LogManagerImpl.java(10 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/DisruptorEventBus.java(1 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBus.java(1 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactory.java(1 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusHandler.java(1 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusMode.java(1 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusOptions.java(1 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/MpscEventBus.java(1 hunks)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/WaitStrategyType.java(1 hunks)jraft-core/src/test/java/com/alipay/sofa/jraft/core/NodeEventBusModeTest.java(1 hunks)jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactoryTest.java(1 hunks)jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusOptionsTest.java(1 hunks)jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusTest.java(1 hunks)jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/MpscEventBusTest.java(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (6)
jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactoryTest.java (1)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactory.java (1)
EventBusFactory(26-52)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/DisruptorEventBus.java (1)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/DisruptorBuilder.java (1)
DisruptorBuilder(32-98)
jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java (2)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactory.java (1)
EventBusFactory(26-52)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusOptions.java (1)
EventBusOptions(29-192)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusOptions.java (1)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/NamedThreadFactory.java (1)
NamedThreadFactory(32-70)
jraft-core/src/main/java/com/alipay/sofa/jraft/storage/impl/LogManagerImpl.java (2)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactory.java (1)
EventBusFactory(26-52)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusOptions.java (1)
EventBusOptions(29-192)
jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusTest.java (2)
jraft-core/src/test/java/com/alipay/sofa/jraft/core/NodeEventBusModeTest.java (1)
RunWith(67-369)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactory.java (1)
EventBusFactory(26-52)
🔇 Additional comments (39)
jraft-core/src/main/java/com/alipay/sofa/jraft/storage/LogManager.java (1)
249-249: Formatting change approved.This is a cosmetic addition of a trailing newline at the end of the file. No functional or API changes are present.
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/WaitStrategyType.java (1)
24-41: LGTM! Clear and well-documented enum.The enum constants are well-defined with comprehensive Javadoc explaining their behavior in both Disruptor and MPSC contexts. This provides clear guidance for users selecting a wait strategy.
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/MpscEventBus.java (2)
92-120: LGTM! Publish methods correctly handle blocking and non-blocking semantics.The
publish()method properly spins with backpressure when the queue is full, whiletryPublish()provides a non-blocking alternative. Both correctly unpark the consumer thread to ensure timely processing.
153-223: LGTM! Event consumption logic correctly implements batching and error handling.The consumer loop properly tracks batch boundaries using queue emptiness and max batch size, and handles exceptions without crashing the consumer thread. The drain logic during shutdown correctly ignores additional
ShutdownEventinstances.jraft-core/src/main/java/com/alipay/sofa/jraft/storage/impl/LogManagerImpl.java (2)
197-206: LGTM! EventBus initialization properly configured.The EventBus is correctly initialized with options from RaftOptions, including the configurable mode, buffer size, wait strategy, and timeout. The factory pattern cleanly abstracts the implementation selection.
498-571: LGTM! EventBusHandler implementation correctly handles batching and shutdown.The handler properly implements the
EventBusHandler<StableClosureEvent>interface, respects batch boundaries for efficiency, and correctly handles shutdown by flushing pending work before counting down the latch.jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusMode.java (1)
25-52: LGTM! Well-documented enum with clear GC guidance.The enum provides excellent documentation explaining when to use each mode based on GC characteristics. The guidance clearly addresses the cross-generation reference issue mentioned in #1231, helping users make informed configuration choices.
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactory.java (1)
26-52: LGTM! Clean factory implementation with proper validation.The factory correctly validates inputs and uses a switch statement to instantiate the appropriate EventBus implementation based on the configured mode. The default case properly handles unknown modes with a clear exception.
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusHandler.java (1)
25-41: LGTM! Well-designed functional interface with clear batch semantics.The interface provides a clean contract for event processing with explicit batch boundary signaling. The Javadoc clearly explains when
endOfBatchis true, enabling handlers to optimize batch operations.jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusOptionsTest.java (1)
35-190: LGTM! Comprehensive test coverage for EventBusOptions.The tests thoroughly validate all configuration options including default values, setters, boundary conditions, null checks, and fluent API chaining. The use of expected exceptions for negative test cases is appropriate.
jraft-core/src/main/java/com/alipay/sofa/jraft/option/FSMCallerOptions.java (1)
45-64: LGTM! EventBusMode property properly integrated.The new
eventBusModeproperty with a default value ofDISRUPTORmaintains backward compatibility while enabling users to opt into the MPSC mode for generational GC environments. The getter/setter follow the existing pattern in the class.jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactoryTest.java (3)
34-49: LGTM!Good test coverage for DisruptorEventBus creation with proper type assertion and graceful shutdown handling.
51-66: LGTM!Good test coverage for MpscEventBus creation with proper type assertion and graceful shutdown handling.
68-77: LGTM!Null parameter validation tests are correctly implemented using JUnit's
expectedattribute.jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBus.java (1)
31-89: LGTM!Well-designed interface with clear API documentation. The separation between blocking (
publish) and non-blocking (tryPublish) operations is appropriate, and theCountDownLatch-based shutdown coordination provides a clean synchronization mechanism.jraft-core/src/main/java/com/alipay/sofa/jraft/option/RaftOptions.java (3)
116-128: LGTM!Excellent documentation explaining the trade-offs between DISRUPTOR and MPSC modes with respect to GC behavior. The default value of
DISRUPTORmaintains backward compatibility.
137-143: LGTM!Standard getter/setter pattern consistent with other fields in the class.
331-331: LGTM!The
copy()method correctly propagates the neweventBusModefield.jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/DisruptorEventBus.java (2)
57-65: LGTM!The
EventWrapperclass withreset()method properly supports object reuse and reference clearing for GC-friendliness.
156-172: LGTM!The
InternalHandlercorrectly adapts the Disruptor event model toEventBusHandler, handles shutdown gracefully by counting down the latch and stopping the disruptor, and ensures wrapper cleanup in thefinallyblock.jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/MpscEventBusTest.java (3)
40-47: LGTM!Proper cleanup in
@Afterwith null check and shutdown handling.
52-98: LGTM!Good test for batch size triggering behavior. The assertion on line 95-96 correctly accounts for potential additional
endOfBatch=trueoccurrences when the queue empties.
183-216: LGTM!Excellent test for drain-on-shutdown behavior. The test correctly verifies that all 50 events are processed even when shutdown is called while events are still queued, with appropriate timeout on the shutdown latch.
jraft-core/src/main/java/com/alipay/sofa/jraft/core/ReadOnlyServiceImpl.java (2)
195-196: Good addition: Clearing state references helps reduce cross-generation GC pressure.Explicitly clearing the
stateslist after processing ensures that completedReadIndexStateobjects can be collected sooner, addressing the GC pressure issue described in Issue #1231.
255-260: LGTM: EventBus initialization is properly configured.The initialization correctly sources the mode from
raftOptions.getEventBusMode(), sets an appropriate buffer size, and uses a descriptive thread factory name.jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusTest.java (1)
176-212: Good concurrency test coverage for multi-threaded publish scenarios.This test properly validates thread safety by using multiple producer threads with proper synchronization via
CountDownLatch. The assertions verify that all published events are received.jraft-core/src/test/java/com/alipay/sofa/jraft/core/NodeEventBusModeTest.java (2)
122-174: Good test coverage for basic cluster operations with both EventBus modes.This test validates that the fundamental apply and TaskClosure semantics work identically under both DISRUPTOR and MPSC modes, which is essential for ensuring the new abstraction maintains compatibility.
301-352: LGTM: Good stress test for concurrent apply scenarios.This test validates that both EventBus implementations can handle high-concurrency task submissions (10 threads × 100 tasks) without failures, which is critical for production workloads.
jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java (3)
118-135: LGTM: Handler correctly implements EventBusHandler with preserved batching semantics.The
ApplyTaskHandlerproperly implements the newEventBusHandlerinterface while maintaining themaxCommittedIndexbatching optimization for committed tasks. ThefsmThreadassignment is safe since there's only one consumer thread.
176-179: LGTM: EventBus initialization properly configured from FSMCallerOptions.
229-234: Consider the GC implications of allocating new ApplyTask per operation.Each call to
onCommitted,onSnapshotLoad, etc., allocates anew ApplyTask(). While this simplifies the code (no translator/factory pattern), it may still create cross-generation references if these tasks reference older objects.For MPSC mode, this is expected since there's no object reuse. For DISRUPTOR mode, the original Disruptor pattern pre-allocated events in the RingBuffer and used translators to populate them, avoiding allocations. This change trades that optimization for simpler code.
Is this allocation pattern intentional? The PR objective mentions eliminating "new" allocations to reduce GC pressure.
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusOptions.java (1)
29-191: Well-designed configuration class with fluent API and sensible defaults.The
EventBusOptionsclass provides a clean configuration interface with:
- Sensible defaults (16K buffer, 32 batch size, blocking wait strategy)
- Fluent setters for chaining
- Validation on all setters
- Informative
toString()for debuggingjraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java (7)
123-128: LGTM - EventBus imports added.The new imports for EventBus infrastructure are correctly added and all are used in the refactored code.
209-210: LGTM - Core abstraction change from Disruptor to EventBus.The field type and documentation have been correctly updated to reflect the new EventBus-based architecture.
274-297: LGTM - Handler correctly adapted to EventBus interface.The handler now implements
EventBusHandler<LogEntryAndClosure>with the updated signature. The use oftasks.clear()is appropriate for the EventBus abstraction, which may use different object lifecycle strategies depending on the mode (Disruptor vs MPSC).
750-750: LGTM - EventBus mode propagated to FSMCaller.The eventBusMode configuration is correctly propagated from RaftOptions to FSMCallerOptions, ensuring consistent EventBus behavior across components.
970-976: LGTM - EventBus initialization correctly configured.The EventBus is properly initialized with appropriate parameters:
- Mode from RaftOptions for flexibility
- Buffer size and batch size from existing Raft options for consistency
- Blocking wait strategy suitable for the apply path
- Daemon threads for proper lifecycle management
1638-1661: LGTM - Apply method correctly refactored for EventBus.The apply() method now:
- Creates LogEntryAndClosure events directly (appropriate for both MPSC and Disruptor modes)
- Handles blocking and non-blocking publish modes
- Properly handles overload conditions in non-blocking mode
The object allocation pattern is intentional: MPSC mode avoids object reuse to prevent cross-generation references (the main goal of this PR), while Disruptor mode can handle reuse internally if needed.
2905-2907: LGTM - EventBus cleanup in join().The EventBus reference is properly nulled out after the shutdown latch completes, ensuring clean resource release.
| final ReadIndexEvent event = new ReadIndexEvent(); | ||
| event.readOnlyOptions = readOnlyOptions; | ||
| event.done = closure; | ||
| event.requestContext = new Bytes(reqCtx); | ||
| event.startTime = Utils.monotonicMs(); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
find . -type f -name "ReadOnlyServiceImpl.java" | head -5Repository: sofastack/sofa-jraft
Length of output: 142
🏁 Script executed:
head -100 ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/ReadOnlyServiceImpl.javaRepository: sofastack/sofa-jraft
Length of output: 4426
🏁 Script executed:
sed -n '300,330p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/ReadOnlyServiceImpl.javaRepository: sofastack/sofa-jraft
Length of output: 1611
🏁 Script executed:
sed -n '50,120p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/ReadOnlyServiceImpl.javaRepository: sofastack/sofa-jraft
Length of output: 3077
NewReadIndexEvent allocations per request defeat the GC optimization goal when using EventBus instead of Disruptor.
Lines 314 and 317 allocate new objects on every addRequest call:
new ReadIndexEvent()creates a fresh event objectnew Bytes(reqCtx)creates a fresh wrapper object
Disruptor pre-allocates event objects in the ring buffer during construction and reuses them by allowing callers to modify fields on the stored objects. EventBus, by contrast, uses a thread pool to reuse threads but does not provide pre-allocated event object pooling.
If the PR migrated ReadOnlyService from Disruptor to EventBus to address Issue #1231's GC pressure concerns, this implementation does not achieve that goal without explicit event object pooling. The handler at lines 98-117 only batches event processing; it does not reuse event objects across requests.
Consider implementing an object pool for ReadIndexEvent or reverting to Disruptor's event reuse pattern for the ReadOnlyService queue.
🤖 Prompt for AI Agents
In jraft-core/src/main/java/com/alipay/sofa/jraft/core/ReadOnlyServiceImpl.java
around lines 314 to 318, new ReadIndexEvent() and new Bytes(reqCtx) are
allocated per addRequest which defeats the GC optimization when using EventBus;
change addRequest to borrow a reusable ReadIndexEvent (and reusable Bytes
wrapper) from a thread-safe object pool preallocated at service construction,
set its fields (readOnlyOptions, done, requestContext, startTime), and change
the event handler (lines ~98-117) to reset/clear the event fields and return the
event (and wrapper) to the pool after processing; ensure the pool is
thread-safe, prefilled to a reasonable size, defines a clear policy on
exhaustion (block, grow, or drop with fallback allocation), and that returned
events are safely reset to avoid leaking state between requests.
| final StableClosureEvent event = new StableClosureEvent(); | ||
| event.type = EventType.OTHER; | ||
| event.done = done; | ||
| this.diskEventBus.publish(event); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find the event bus implementations
find . -type f -name "*.java" | xargs grep -l "class DisruptorEventBus\|class MpscEventBus" | head -20Repository: sofastack/sofa-jraft
Length of output: 321
🏁 Script executed:
#!/bin/bash
# Read DisruptorEventBus to check publish() method
echo "=== DisruptorEventBus ==="
cat ./jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/DisruptorEventBus.java
echo -e "\n\n=== MpscEventBus ==="
cat ./jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/MpscEventBus.javaRepository: sofastack/sofa-jraft
Length of output: 14325
🏁 Script executed:
#!/bin/bash
# Read LogManagerImpl around lines 340-380 to see context and error handling
sed -n '340,380p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/storage/impl/LogManagerImpl.javaRepository: sofastack/sofa-jraft
Length of output: 1302
Check publish() return value to prevent silent closure failures.
Both DisruptorEventBus.publish() and MpscEventBus.publish() can return false if the event bus is shut down. At lines 350-353 in the appendEntries method and lines 374-377 in the offerEvent method, the return values are ignored. If publish() fails and returns false, the closure callback will never be invoked, causing a silent failure where the caller never receives a response. Even in offerEvent where this.stopped is checked beforehand, a race condition exists if shutdown occurs between the check and the publish call. Add return value checks or error handling to ensure closures are always invoked on failure.
🤖 Prompt for AI Agents
In
jraft-core/src/main/java/com/alipay/sofa/jraft/storage/impl/LogManagerImpl.java
around lines 350-353 and 374-377, the calls to this.diskEventBus.publish(...)
ignore the boolean return value and can silently drop events during shutdown;
change both sites to capture the publish result and, if publish returns false,
immediately invoke the provided closure (done) with a failure/aborted status and
log the condition so the caller is notified; do the same even when this.stopped
is true (to cover the race) — i.e., check the boolean return, call the closure
with an appropriate error Status and return, and add a small debug/error log
explaining that publish failed due to event bus shutdown.
| @Override | ||
| public boolean publish(final T event) { | ||
| if (this.shutdown) { | ||
| LOG.warn("EventBus '{}' is shutdown, cannot publish event", this.name); | ||
| return false; | ||
| } | ||
| this.ringBuffer.publishEvent((wrapper, seq) -> wrapper.event = event); | ||
| return true; | ||
| } |
There was a problem hiding this comment.
Minor race between shutdown check and publish.
There's a small window where shutdown could be set to true after the check on line 93 but before publishEvent on line 97. This could result in an event being published after the shutdown signal. In practice, this is likely benign since the event will still be processed before the shutdown event, but it's worth noting for correctness.
🤖 Prompt for AI Agents
In
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/DisruptorEventBus.java
around lines 91 to 99, there's a race where shutdown can flip true after the
initial check but before publishEvent, allowing publishes after shutdown; wrap
the shutdown-check and publishEvent call in a single atomic guarded section (for
example use a private final lock object and synchronize the check + publish, or
use an AtomicBoolean and compare-and-act) so you check shutdown inside the same
guard that performs ringBuffer.publishEvent; return false if shutdown is
observed inside the guard.
|
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactory.java I was reading the code and noticed that the implementation here hardcodes two concrete classes. |
Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (7)
jraft-core/src/main/java/com/alipay/sofa/jraft/storage/impl/LogManagerImpl.java (3)
232-237: Shutdown publish lacks error handling.The
publish()return value is ignored. If the EventBus is already shut down or the queue is full, the shutdown event won't be processed, andshutDownLatchwill never count down, causingjoin()to hang indefinitely.🔎 Proposed fix
private void stopDiskThread() { this.shutDownLatch = new CountDownLatch(1); final StableClosureEvent shutdownEvent = new StableClosureEvent(); shutdownEvent.type = EventType.SHUTDOWN; - ThreadPoolsFactory.runInThread(this.groupId, () -> this.diskEventBus.publish(shutdownEvent)); + ThreadPoolsFactory.runInThread(this.groupId, () -> { + if (!this.diskEventBus.publish(shutdownEvent)) { + LOG.warn("Failed to publish shutdown event, counting down latch directly."); + this.shutDownLatch.countDown(); + } + }); }
350-353: Check publish() return value to prevent silent failures.Both
appendEntriesandofferEventignore the boolean return fromdiskEventBus.publish(). If publish fails, the closure callback will never be invoked, causing silent failures where the caller never receives a response.🔎 Proposed fix
// publish event out of lock final StableClosureEvent event = new StableClosureEvent(); event.type = EventType.OTHER; event.done = done; - this.diskEventBus.publish(event); + if (!this.diskEventBus.publish(event)) { + ThreadPoolsFactory.runClosureInThread(this.groupId, done, + new Status(RaftError.EBUSY, "Failed to publish event to disk EventBus")); + }
374-377: Same issue: publish() return value ignored in offerEvent.If publish fails after the
stoppedcheck passes (race condition), the closure won't be invoked.🔎 Proposed fix
final StableClosureEvent event = new StableClosureEvent(); event.type = type; event.done = done; - this.diskEventBus.publish(event); + if (!this.diskEventBus.publish(event)) { + ThreadPoolsFactory.runClosureInThread(this.groupId, done, + new Status(RaftError.EBUSY, "Failed to publish event to disk EventBus")); + } }jraft-core/src/main/java/com/alipay/sofa/jraft/core/ReadOnlyServiceImpl.java (2)
283-287: Shutdown publish lacks error handling.Similar to other components, if
readIndexEventBus.publish()fails, theshutdownLatchwill never count down, causingjoin()to hang.
313-318: New object allocations per request may not fully address GC optimization goal.Each
addRequestcreates a newReadIndexEventandnew Bytes(reqCtx). The PR objective (Issue #1231) was to reduce GC pressure from cross-generation references in Disruptor by avoiding new allocations. Without object pooling, this EventBus migration doesn't achieve the GC optimization goal—it just moves allocations from Disruptor's translator to the caller side.jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java (1)
196-201: Shutdown publish lacks error handling.If
taskEventBus.publish(task)fails (returns false), the shutdown latch will never be counted down, causingjoin()to hang indefinitely. This is a common pattern issue across all EventBus usages in this PR.🔎 Proposed fix
ThreadPoolsFactory.runInThread(getNode().getGroupId(), () -> { final ApplyTask task = new ApplyTask(); task.type = TaskType.SHUTDOWN; task.shutdownLatch = latch; - this.taskEventBus.publish(task); + if (!this.taskEventBus.publish(task)) { + LOG.warn("Failed to publish SHUTDOWN task, counting down latch directly."); + latch.countDown(); + } });jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java (1)
2814-2819: Add error handling for shutdown event publication to prevent hang.If
applyEventBus.publish(shutdownEvent)fails or throws an exception, theshutdownLatchwill never be counted down, causingjoin()at line 2906 to hang indefinitely. This is critical for graceful shutdown.🔎 Recommended fix
if (this.applyEventBus != null) { final CountDownLatch latch = new CountDownLatch(1); this.shutdownLatch = latch; final LogEntryAndClosure shutdownEvent = new LogEntryAndClosure(); shutdownEvent.shutdownLatch = latch; - ThreadPoolsFactory.runInThread(this.groupId, () -> this.applyEventBus.publish(shutdownEvent)); + ThreadPoolsFactory.runInThread(this.groupId, () -> { + try { + this.applyEventBus.publish(shutdownEvent); + } catch (final Throwable t) { + LOG.error("Failed to publish shutdown event on applyEventBus for node {}.", this.nodeId, t); + latch.countDown(); + final int num = GLOBAL_NUM_NODES.decrementAndGet(); + LOG.info("The number of active nodes decrement to {}.", num); + } + }); }
🧹 Nitpick comments (3)
jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusTest.java (1)
166-174: Consider using awaitility instead of Thread.sleep for timing-sensitive assertions.
Thread.sleep(100)in tests can be flaky on slow CI machines. Consider using a polling mechanism or the Awaitility library for more robust timing verification.jraft-core/src/main/java/com/alipay/sofa/jraft/option/RaftOptions.java (2)
159-161: Consider adding null validation in setters.The setters for
eventBusModeandeventBusFactoryaccept null values without validation. If null is passed, it could lead toNullPointerExceptionat runtime when these values are used.🔎 Proposed validation
public void setEventBusMode(final EventBusMode eventBusMode) { + Requires.requireNonNull(eventBusMode, "eventBusMode"); this.eventBusMode = eventBusMode; }public void setEventBusFactory(final EventBusFactory eventBusFactory) { + Requires.requireNonNull(eventBusFactory, "eventBusFactory"); this.eventBusFactory = eventBusFactory; }
373-373: Consider including eventBusFactory in toString() for better debugging.The
toString()method includeseventBusModebut omitseventBusFactory. Including the factory (even just its class name) could help with debugging configuration issues.🔎 Proposed enhancement
- + ", startUpOldStorage=" + startupOldStorage + ", eventBusMode=" + eventBusMode + '}'; + + ", startUpOldStorage=" + startupOldStorage + ", eventBusMode=" + eventBusMode + + ", eventBusFactory=" + (eventBusFactory != null ? eventBusFactory.getClass().getSimpleName() : "null") + '}';
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (12)
jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.javajraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.javajraft-core/src/main/java/com/alipay/sofa/jraft/core/ReadOnlyServiceImpl.javajraft-core/src/main/java/com/alipay/sofa/jraft/option/FSMCallerOptions.javajraft-core/src/main/java/com/alipay/sofa/jraft/option/RaftOptions.javajraft-core/src/main/java/com/alipay/sofa/jraft/storage/impl/LogManagerImpl.javajraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/DefaultEventBusFactory.javajraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactory.javajraft-core/src/main/resources/META-INF/services/com.alipay.sofa.jraft.util.concurrent.EventBusFactoryjraft-core/src/test/java/com/alipay/sofa/jraft/core/NodeEventBusModeTest.javajraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactoryTest.javajraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusTest.java
✅ Files skipped from review due to trivial changes (1)
- jraft-core/src/main/resources/META-INF/services/com.alipay.sofa.jraft.util.concurrent.EventBusFactory
🚧 Files skipped from review as they are similar to previous changes (1)
- jraft-core/src/test/java/com/alipay/sofa/jraft/core/NodeEventBusModeTest.java
🧰 Additional context used
🧬 Code graph analysis (4)
jraft-core/src/main/java/com/alipay/sofa/jraft/option/RaftOptions.java (2)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/JRaftServiceLoader.java (1)
JRaftServiceLoader(42-357)jraft-core/src/main/java/com/alipay/sofa/jraft/util/RpcFactoryHelper.java (1)
RpcFactoryHelper(25-37)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactory.java (1)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/DefaultEventBusFactory.java (1)
SPI(30-47)
jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java (3)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusOptions.java (1)
EventBusOptions(29-192)jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/LongHeldDetectingReadWriteLock.java (1)
LongHeldDetectingReadWriteLock(31-153)jraft-core/src/main/java/com/alipay/sofa/jraft/util/ThreadPoolsFactory.java (1)
ThreadPoolsFactory(38-114)
jraft-core/src/main/java/com/alipay/sofa/jraft/storage/impl/LogManagerImpl.java (2)
jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusOptions.java (1)
EventBusOptions(29-192)jraft-core/src/main/java/com/alipay/sofa/jraft/util/ThreadPoolsFactory.java (1)
ThreadPoolsFactory(38-114)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (5)
- GitHub Check: test_rpc_grpc_impl
- GitHub Check: test_rheakv_core
- GitHub Check: test_rheakv_pd
- GitHub Check: test_jraft_core
- GitHub Check: sca
🔇 Additional comments (21)
jraft-core/src/main/java/com/alipay/sofa/jraft/storage/impl/LogManagerImpl.java (2)
195-206: EventBus initialization looks correct.The initialization properly configures the EventBus with mode from RaftOptions, buffer size, thread factory, and timeout-based wait strategy. The factory pattern allows for SPI-based customization as intended.
498-505: Handler correctly implements EventBusHandler interface.The signature change from
onEvent(event, sequence, endOfBatch)toonEvent(event, endOfBatch)aligns with the new EventBusHandler contract. The logic is preserved correctly.jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactory.java (1)
34-46: Clean SPI interface design.The interface is well-documented, supports both SPI-based discovery and direct injection via
RaftOptions.setEventBusFactory(), and addresses the extensibility concern raised in the PR comments about supporting custom implementations.jraft-core/src/main/java/com/alipay/sofa/jraft/util/concurrent/DefaultEventBusFactory.java (1)
30-47: Factory implementation is correct and follows best practices.Proper null validation with
Requires.requireNonNull, exhaustive switch handling with a defensive default case, and@SPI(priority = 0)ensures this is the default implementation while allowing higher-priority custom factories to override it.jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusFactoryTest.java (1)
39-208: Comprehensive test coverage for EventBusFactory.The test suite covers:
- Both bus modes (DISRUPTOR, MPSC)
- Null argument validation
- SPI loading mechanism
- Custom factory injection
- RaftOptions integration and copy semantics
All tests properly clean up by shutting down EventBus instances, preventing resource leaks.
jraft-core/src/main/java/com/alipay/sofa/jraft/option/FSMCallerOptions.java (1)
46-77: Configuration options correctly extended for EventBus support.The new
eventBusModewithDISRUPTORdefault ensures backward compatibility. TheeventBusFactoryfield allows callers to inject custom factories. Standard getter/setter pattern is consistent with existing options.jraft-core/src/main/java/com/alipay/sofa/jraft/core/ReadOnlyServiceImpl.java (2)
253-259: EventBus initialization is correct.Configuration properly uses RaftOptions for mode, buffer size, and creates a dedicated thread factory. The factory pattern correctly delegates to the configured EventBusFactory.
194-196: Good addition of explicit collection clearing for GC.The
states.clear()calls after processing help break reference chains and allow earlier garbage collection, directly addressing part of Issue #1231's concerns about long-lived objects holding references to newly allocated objects.Also applies to: 426-428, 443-445
jraft-core/src/test/java/com/alipay/sofa/jraft/util/concurrent/EventBusTest.java (1)
42-277: Comprehensive parameterized test suite for EventBus implementations.Excellent coverage testing both
DisruptorEventBusandMpscEventBusimplementations for:
- Basic publish/consume semantics
- Non-blocking
tryPublish- Shutdown lifecycle and post-shutdown behavior
- Capacity and buffer size reporting
- Concurrent publishing from multiple threads
- Pending event counting
The parameterized approach ensures both implementations conform to the same contract.
jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java (2)
174-178: EventBus initialization is properly configured.Uses mode and buffer size from FSMCallerOptions, with a dedicated named thread factory for the consumer thread.
117-126: Handler implementation correctly adapted for EventBusHandler.The
ApplyTaskHandlerproperly implementsEventBusHandler<ApplyTask>with the updated signature. The FSM thread initialization and batch processing logic are preserved correctly.jraft-core/src/main/java/com/alipay/sofa/jraft/option/RaftOptions.java (2)
357-358: LGTM! EventBus configuration properly propagated in copy().The new fields are correctly copied. The shallow copy of
eventBusFactoryis appropriate since factories are typically stateless and designed to be shared.
144-145: EventBusFactory SPI implementation is properly registered and null-safe.A default
DefaultEventBusFactoryimplementation is registered inMETA-INF/services/com.alipay.sofa.jraft.util.concurrent.EventBusFactory. Additionally,JRaftServiceLoader.first()throws aServiceConfigurationErrorwhen no provider is found rather than returning null, making the code safe fromNullPointerException.Likely an incorrect or invalid review comment.
jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java (8)
123-127: LGTM! EventBus imports properly added.The new imports support the migration from Disruptor to EventBus and are correctly organized.
208-209: LGTM! Field declaration correctly updated for EventBus.The field type change from Disruptor to
EventBus<LogEntryAndClosure>is appropriate and the comment accurately describes its purpose.
273-278: LGTM! Handler correctly migrated to EventBusHandler interface.The signature change from
onEvent(event, sequence, endOfBatch)toonEvent(event, endOfBatch)aligns with the EventBus API. The removal of thesequenceparameter is correct as EventBus doesn't expose sequence numbers.
280-295: LGTM! Event batch processing correctly updated.The change from per-event
reset()calls to batchclear()aligns with the PR objective of reducing cross-generational references. This approach is cleaner and better suited for generational GC modes like ZGC-gen and G1.
749-750: LGTM! EventBus configuration properly propagated to FSMCaller.The EventBus mode and factory are correctly passed from
RaftOptionstoFSMCallerOptions, ensuring consistent EventBus configuration across components.
970-976: LGTM! EventBus initialization is comprehensive and well-configured.The EventBus is properly initialized with:
- Mode from
RaftOptions(supporting DISRUPTOR/MPSC choice)- Appropriate buffer size and batch size from existing configurations
- Named thread factory for debuggability
- BLOCKING wait strategy matching previous Disruptor behavior
1629-1662: LGTM! Task application properly migrated to EventBus with good error handling.The changes correctly implement:
- Direct event creation instead of translator-based publishing
- Proper handling of both
BlockingandNonBlockingapply modes- Appropriate error handling when the EventBus is full
- Metrics recording for overload situations
The error message accurately references
applyEventBus.getBufferSize()for debugging.
2906-2907: LGTM! EventBus properly cleaned up in join().Setting
applyEventBustonullafter the shutdown latch completes is appropriate and prevents accidental use after shutdown.
| private boolean enqueueTask(final ApplyTask task) { | ||
| if (this.shutdownLatch != null) { | ||
| // Shutting down | ||
| LOG.warn("FSMCaller is stopped, can not apply new task."); | ||
| return false; | ||
| } | ||
| this.taskQueue.publishEvent(tpl); | ||
| return true; | ||
| return this.taskEventBus.publish(task); | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find and examine enqueueTask usage in FSMCallerImpl.java
find . -name "FSMCallerImpl.java" -type f | head -5Repository: sofastack/sofa-jraft
Length of output: 136
🏁 Script executed:
#!/bin/bash
# Read the FSMCallerImpl.java file and search for enqueueTask calls
wc -l ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.javaRepository: sofastack/sofa-jraft
Length of output: 140
🏁 Script executed:
#!/bin/bash
# Search for all enqueueTask calls in the file with context
rg -n "enqueueTask" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java -A 5 -B 2Repository: sofastack/sofa-jraft
Length of output: 2779
🏁 Script executed:
#!/bin/bash
# Search for callers of onCommitted, onSnapshotLoad, onSnapshotSave, onLeaderStart, etc.
# First, let's see the full context around the flush method
sed -n '235,250p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.javaRepository: sofastack/sofa-jraft
Length of output: 548
🏁 Script executed:
#!/bin/bash
# Check how external code calls these FSMCaller methods
# Search for callers of onCommitted, onSnapshotLoad, onSnapshotSave, etc.
rg -n "\.onCommitted\(|\.onSnapshotLoad\(|\.onSnapshotSave\(|\.onLeaderStart\(" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/ -A 2 | head -60Repository: sofastack/sofa-jraft
Length of output: 2696
🏁 Script executed:
#!/bin/bash
# Check callers in NodeImpl and BallotBox
# First check BallotBox.commitAt method
sed -n '135,145p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/BallotBox.javaRepository: sofastack/sofa-jraft
Length of output: 382
🏁 Script executed:
#!/bin/bash
# Search for actual calls to fsmCaller or waiter methods from FSMCaller interface
# in other files to see if return values are checked
rg -n "fsmCaller\.(onCommitted|onSnapshotLoad|onSnapshotSave|onLeaderStart|onLeaderStop|onStartFollowing|onStopFollowing)" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/ -B 2 -A 2Repository: sofastack/sofa-jraft
Length of output: 2549
🏁 Script executed:
#!/bin/bash
# Check if the callers in NodeImpl check return values from these methods
# Line 1199 - onStopFollowing
sed -n '1195,1210p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.javaRepository: sofastack/sofa-jraft
Length of output: 753
🏁 Script executed:
#!/bin/bash
# Check line 2935 - onLeaderStart call in NodeImpl
sed -n '2930,2945p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.javaRepository: sofastack/sofa-jraft
Length of output: 606
🏁 Script executed:
#!/bin/bash
# Check line 3270 - onLeaderStop call in NodeImpl
sed -n '3265,3275p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.javaRepository: sofastack/sofa-jraft
Length of output: 414
🏁 Script executed:
#!/bin/bash
# Now let's check how BallotBox calls onCommitted
# to see if it checks the return value
sed -n '125,150p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/BallotBox.javaRepository: sofastack/sofa-jraft
Length of output: 1384
🏁 Script executed:
#!/bin/bash
# Let's look at how the flush method (line 244) handles the return value
# and check what taskEventBus.publish does
sed -n '240,250p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.javaRepository: sofastack/sofa-jraft
Length of output: 430
🏁 Script executed:
#!/bin/bash
# Let's check the publish method and see what it does
# Also check if there's any exception handling
sed -n '210,220p' ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.javaRepository: sofastack/sofa-jraft
Length of output: 430
🏁 Script executed:
#!/bin/bash
# Let's verify the actual implementation by checking where taskEventBus is initialized
# and what publish() returns
rg -n "taskEventBus" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java | head -20Repository: sofastack/sofa-jraft
Length of output: 480
🏁 Script executed:
#!/bin/bash
# Let's check if there are any other callers that DO check return values
# to compare patterns
rg -n "if.*onCommitted|if.*onSnapshotLoad|if.*onSnapshotSave|if.*onLeaderStart|if.*onLeaderStop|if.*onStartFollowing|if.*onStopFollowing" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/ -B 1 -A 1Repository: sofastack/sofa-jraft
Length of output: 428
🏁 Script executed:
#!/bin/bash
# Let's check the exact lines in NodeImpl where return values are not checked
grep -n "fsmCaller\.\(onStopFollowing\|onStartFollowing\|onLeaderStart\|onLeaderStop\)" ./jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.javaRepository: sofastack/sofa-jraft
Length of output: 412
Verify return value handling in FSMCaller method callers to prevent silent task drops.
The concern is valid. Several callers in NodeImpl and BallotBox invoke FSMCaller methods without checking their boolean return values:
NodeImpl.resetLeaderId()(lines 1199, 1204): ignores returns fromonStopFollowing()andonStartFollowing()NodeImpl(line 2935): ignores return fromonLeaderStart()NodeImpl.onLeaderStop()(line 3270): ignores return fromonLeaderStop()BallotBox.commitAt()(line 141): ignores return fromwaiter.onCommitted()FSMCallerImpl.flush()(line 244): ignores return fromenqueueTask(), could deadlock if publish fails
When taskEventBus.publish() returns false (e.g., due to overload), tasks are silently dropped. The flush method is particularly problematic—if publish fails, the latch never signals, potentially causing indefinite blocking. All these callers should handle false returns to avoid silent task drops and state machine inconsistencies.
Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
Motivation:
Explain the context, and why you're making that change.
To make others understand what is the problem you're trying to solve.
Modification:
Describe the idea and modifications you've done.
Result:
Fixes #1231
If there is no issue then describe the changes introduced by this PR.
Summary by CodeRabbit
New Features
Tests
✏️ Tip: You can customize this high-level summary in your review settings.