modify code

Author: luoluoyuyu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java

@@ -71,7 +71,6 @@ public void run() {
     T event = null;
     long nextSequence = sequence.get() + 1L;
 
-    // CORE: Batch processing loop (MUST keep identical logic)
     while (running) {
       try {
         // Wait for available sequence

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java

@@ -19,20 +19,51 @@
 
 package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor;
 
+import java.util.concurrent.atomic.AtomicIntegerArray;
 import java.util.concurrent.locks.LockSupport;
 
 public final class MultiProducerSequencer {
 
+  /** Ring buffer size (must be power of 2) - immutable after construction */
   private final int bufferSize;
+
+  /**
+   * Producer cursor tracking highest claimed sequence Updated via CAS in next() method Volatile
+   * reads/writes handled by Sequence class
+   */
   private final Sequence cursor = new Sequence();
+
+  /**
+   * Array of consumer sequences for backpressure control MUST be volatile for safe publication when
+   * modified by SequenceGroups Array reference is replaced atomically via
+   * AtomicReferenceFieldUpdater
+   */
   volatile Sequence[] gatingSequences;
 
-  // CRITICAL: Cache to avoid repeated getMinimumSequence calls
+  /**
+   * Cached minimum gating sequence to reduce contention Updated opportunistically in next() to
+   * avoid expensive array scan Does not need to be perfectly accurate (conservative is safe)
+   */
   private final Sequence gatingSequenceCache = new Sequence();
 
-  // CRITICAL: Available buffer tracks published sequences
-  private final int[] availableBuffer;
+  /**
+   * CRITICAL: Availability flags for tracking published sequences
+   *
+   * <p>Handles out-of-order publishing in multi-producer scenario: - Thread A claims seq 10, still
+   * writing - Thread B claims seq 11, finishes and publishes - Consumer MUST wait for seq 10 before
+   * reading seq 11
+   *
+   * <p>Memory visibility guarantees: - Writers use lazySet() for store-store barrier (cheaper than
+   * volatile write) - Readers use get() for volatile read (ensures visibility across threads)
+   *
+   * <p>AtomicIntegerArray provides same semantics as Unsafe without reflection
+   */
+  private final AtomicIntegerArray availableBuffer;
+
+  /** Mask for fast modulo: sequence & indexMask == sequence % bufferSize */
   private final int indexMask;
+
+  /** Shift for calculating wrap count: sequence >>> indexShift */
   private final int indexShift;
 
   public MultiProducerSequencer(int bufferSize, Sequence[] gatingSequences) {
@@ -45,7 +76,7 @@ public MultiProducerSequencer(int bufferSize, Sequence[] gatingSequences) {
 
     this.bufferSize = bufferSize;
     this.gatingSequences = gatingSequences != null ? gatingSequences : new Sequence[0];
-    this.availableBuffer = new int[bufferSize];
+    this.availableBuffer = new AtomicIntegerArray(bufferSize);
     this.indexMask = bufferSize - 1;
     this.indexShift = log2(bufferSize);
 
@@ -73,8 +104,8 @@ public long next(int n) {
       current = cursor.get();
       next = current + n;
 
-      long wrapPoint = next - bufferSize;
-      long cachedGatingSequence = gatingSequenceCache.get();
+      final long wrapPoint = next - bufferSize;
+      final long cachedGatingSequence = gatingSequenceCache.get();
 
       if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current) {
         long gatingSequence = Sequence.getMinimumSequence(gatingSequences, current);
@@ -105,11 +136,14 @@ public void publish(long lo, long hi) {
     }
   }
 
-  /** CORE: Check if available - MUST use Unsafe.getIntVolatile */
+  /**
+   * CORE: Check if sequence is available for consumption Uses volatile read to ensure visibility of
+   * published sequences
+   */
   public boolean isAvailable(long sequence) {
     int index = calculateIndex(sequence);
     int flag = calculateAvailabilityFlag(sequence);
-    return availableBuffer[index] == flag;
+    return availableBuffer.get(index) == flag;
   }
 
   /** CORE: Get highest published - exact same algorithm */
@@ -178,25 +212,29 @@ public SequenceBarrier newBarrier(Sequence... sequencesToTrack) {
 
   /** Initialize available buffer */
   private void initialiseAvailableBuffer() {
-    for (int i = availableBuffer.length - 1; i != 0; i--) {
+    for (int i = availableBuffer.length() - 1; i != 0; i--) {
       setAvailableBufferValue(i, -1);
     }
     setAvailableBufferValue(0, -1);
   }
 
   /**
-   * CORE: Set available - MUST use Unsafe.putOrderedInt
+   * CORE: Mark sequence as available for consumption
    *
-   * <p>putOrderedInt provides: - Store-store barrier (not full fence) - Cheaper than volatile write
-   * - Sufficient for this use case
+   * <p>Uses lazySet() which provides: - Store-store barrier (ensures all prior writes are visible)
+   * - Cheaper than full volatile write (no store-load barrier) - Sufficient for this use case
+   * (readers use volatile get)
    */
   private void setAvailable(final long sequence) {
     setAvailableBufferValue(calculateIndex(sequence), calculateAvailabilityFlag(sequence));
   }
 
-  /** CRITICAL: Use Unsafe.putOrderedInt for correct memory semantics */
+  /**
+   * Set availability flag with release semantics lazySet() ensures previous event writes are
+   * visible before flag update
+   */
   private void setAvailableBufferValue(int index, int flag) {
-    availableBuffer[index] = flag;
+    availableBuffer.lazySet(index, flag);
   }
 
   /** Calculate availability flag */