From 22b31893579006b1eb18fa817ee89b71dc170e92 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 22 Oct 2025 12:06:11 +0800 Subject: [PATCH 01/11] Pipe: Implementing DisruptorQueue --- .../common/heartbeat/PipeHeartbeatEvent.java | 4 +- .../realtime/assigner/DisruptorQueue.java | 14 +- .../DisruptorQueueExceptionHandler.java | 3 +- .../disruptor/BatchEventProcessor.java | 120 ++++++ .../realtime/disruptor/Disruptor.java | 135 +++++++ .../realtime/disruptor/EventFactory.java | 35 ++ .../realtime/disruptor/EventHandler.java | 38 ++ .../realtime/disruptor/ExceptionHandler.java | 42 ++ .../disruptor/MultiProducerSequencer.java | 259 +++++++++++++ .../realtime/disruptor/RingBuffer.java | 363 ++++++++++++++++++ .../realtime/disruptor/Sequence.java | 153 ++++++++ .../realtime/disruptor/SequenceBarrier.java | 77 ++++ .../realtime/disruptor/SequenceGroups.java | 118 ++++++ 13 files changed, 1349 insertions(+), 12 deletions(-) create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceBarrier.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java index 3cacb5a9245ba..468292b8ecc4e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java @@ -28,10 +28,10 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; import org.apache.iotdb.db.pipe.metric.overview.PipeHeartbeatEventMetrics; +import org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor.RingBuffer; import org.apache.iotdb.db.utils.DateTimeUtils; import org.apache.iotdb.pipe.api.event.Event; -import com.lmax.disruptor.RingBuffer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -200,7 +200,7 @@ public void onTransferred() { /////////////////////////////// Queue size Reporting /////////////////////////////// - public void recordDisruptorSize(final RingBuffer ringBuffer) { + public void recordDisruptorSize(final RingBuffer ringBuffer) { if (shouldPrintMessage) { disruptorSize = ringBuffer.getBufferSize() - (int) ringBuffer.remainingCapacity(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java index 4c3daa4879a44..636646e95c1bb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java @@ -26,12 +26,9 @@ import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; - -import com.lmax.disruptor.BlockingWaitStrategy; -import com.lmax.disruptor.EventHandler; -import com.lmax.disruptor.RingBuffer; -import com.lmax.disruptor.dsl.Disruptor; -import com.lmax.disruptor.dsl.ProducerType; +import org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor.Disruptor; +import org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor.EventHandler; +import org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor.RingBuffer; import java.util.function.Consumer; @@ -68,9 +65,8 @@ public DisruptorQueue( 32, Math.toIntExact( allocatedMemoryBlock.getMemoryUsageInBytes() / ringBufferEntrySizeInBytes)), - THREAD_FACTORY, - ProducerType.MULTI, - new BlockingWaitStrategy()); + THREAD_FACTORY); + disruptor.handleEventsWith( (container, sequence, endOfBatch) -> { final PipeRealtimeEvent realtimeEvent = container.getEvent(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueueExceptionHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueueExceptionHandler.java index 91ad0224fc538..5330f3486f50e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueueExceptionHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueueExceptionHandler.java @@ -19,7 +19,8 @@ package org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner; -import com.lmax.disruptor.ExceptionHandler; +import org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor.ExceptionHandler; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java new file mode 100644 index 0000000000000..b9709f65da4ac --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java @@ -0,0 +1,120 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Batch event processor for consuming events + * + *

Simplified from Disruptor (removed complex lifecycle management) + * + *

CORE algorithm (MUST preserve): + * + *

+ * + * @param event type + */ +public final class BatchEventProcessor implements Runnable { + private static final Logger LOGGER = LoggerFactory.getLogger(BatchEventProcessor.class); + + private final RingBuffer ringBuffer; + private final SequenceBarrier sequenceBarrier; + private final EventHandler eventHandler; + private final Sequence sequence = new Sequence(-1L); + private ExceptionHandler exceptionHandler = new DefaultExceptionHandler<>(); + private volatile boolean running = true; + + public BatchEventProcessor( + RingBuffer ringBuffer, SequenceBarrier barrier, EventHandler eventHandler) { + this.ringBuffer = ringBuffer; + this.sequenceBarrier = barrier; + this.eventHandler = eventHandler; + } + + public Sequence getSequence() { + return sequence; + } + + public void setExceptionHandler(ExceptionHandler exceptionHandler) { + this.exceptionHandler = exceptionHandler; + } + + public void halt() { + running = false; + } + + @Override + 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 + final long availableSequence = sequenceBarrier.waitFor(nextSequence); + + // Batch process all available events + while (nextSequence <= availableSequence) { + event = ringBuffer.get(nextSequence); + eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence); + nextSequence++; + } + + // Update sequence + sequence.set(availableSequence); + + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + LOGGER.info("Processor interrupted"); + break; + } catch (final Throwable ex) { + exceptionHandler.handleEventException(ex, nextSequence, event); + sequence.set(nextSequence); + nextSequence++; + } + } + + LOGGER.info("Processor stopped"); + } + + private static class DefaultExceptionHandler implements ExceptionHandler { + @Override + public void handleEventException(Throwable ex, long sequence, T event) { + LoggerFactory.getLogger(getClass()).error("Exception processing: {} {}", sequence, event, ex); + } + + @Override + public void handleOnStartException(Throwable ex) { + LoggerFactory.getLogger(getClass()).error("Exception during onStart()", ex); + } + + @Override + public void handleOnShutdownException(Throwable ex) { + LoggerFactory.getLogger(getClass()).error("Exception during onShutdown()", ex); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java new file mode 100644 index 0000000000000..00a8da3aa13e8 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.ThreadFactory; + +/** + * High-level API for setting up a lock-free event processing pipeline + * + *

Provides a fluent interface for configuring producers and consumers. Internally manages a + * RingBuffer and event processor thread. + * + *

Configuration is simplified to multi-producer mode with blocking wait strategy. + * + * @param event type + */ +public class Disruptor { + private static final Logger LOGGER = LoggerFactory.getLogger(Disruptor.class); + + private final RingBuffer ringBuffer; + private final ThreadFactory threadFactory; + private BatchEventProcessor processor; + private Thread processorThread; + private ExceptionHandler exceptionHandler; + private volatile boolean started = false; + + /** + * Create a Disruptor instance + * + * @param eventFactory factory for creating pre-allocated events + * @param ringBufferSize buffer size (must be power of 2) + * @param threadFactory factory for creating consumer thread + */ + public Disruptor(EventFactory eventFactory, int ringBufferSize, ThreadFactory threadFactory) { + this.ringBuffer = RingBuffer.createMultiProducer(eventFactory, ringBufferSize); + this.threadFactory = threadFactory; + } + + /** + * Configure event handler for processing events + * + *

Creates a batch event processor that will run in its own thread + * + * @param handler event handler implementation + * @return this instance for method chaining + */ + public Disruptor handleEventsWith(final EventHandler handler) { + SequenceBarrier barrier = ringBuffer.newBarrier(); + processor = new BatchEventProcessor<>(ringBuffer, barrier, handler); + + if (exceptionHandler != null) { + processor.setExceptionHandler(exceptionHandler); + } + + ringBuffer.addGatingSequences(processor.getSequence()); + return this; + } + + /** + * Set exception handler for error handling + * + * @param exceptionHandler handler for processing exceptions + */ + public void setDefaultExceptionHandler(ExceptionHandler exceptionHandler) { + this.exceptionHandler = exceptionHandler; + if (processor != null) { + processor.setExceptionHandler(exceptionHandler); + } + } + + /** Start - MUST keep for IoTDB */ + public RingBuffer start() { + if (started) { + throw new IllegalStateException("Disruptor already started"); + } + + if (processor == null) { + throw new IllegalStateException("No event handler configured"); + } + + processorThread = threadFactory.newThread(processor); + processorThread.start(); + started = true; + + LOGGER.info("Disruptor started with buffer size: {}", ringBuffer.getBufferSize()); + return ringBuffer; + } + + /** Shutdown - MUST keep for IoTDB */ + public void shutdown() { + if (!started) { + return; + } + + if (processor != null) { + processor.halt(); + } + + if (processorThread != null) { + try { + processorThread.join(5000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.warn("Interrupted waiting for processor to stop"); + } + } + + started = false; + LOGGER.info("Disruptor shutdown completed"); + } + + public RingBuffer getRingBuffer() { + return ringBuffer; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java new file mode 100644 index 0000000000000..8214988a28499 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java @@ -0,0 +1,35 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; + +/** + * Event factory for pre-allocating events in RingBuffer + * + * @param event type + */ +@FunctionalInterface +public interface EventFactory { + /** + * Create new event instance + * + * @return new event + */ + T newInstance(); +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java new file mode 100644 index 0000000000000..b84d5895014c7 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java @@ -0,0 +1,38 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; + +/** + * Event handler for processing events from RingBuffer + * + * @param event type + */ +@FunctionalInterface +public interface EventHandler { + /** + * Handle event + * + * @param event the event + * @param sequence sequence number + * @param endOfBatch whether this is the last event in current batch + * @throws Exception if processing fails + */ + void onEvent(T event, long sequence, boolean endOfBatch) throws Exception; +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java new file mode 100644 index 0000000000000..e98138ca0f2f1 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java @@ -0,0 +1,42 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; + +/** + * Exception handler for event processing errors + * + * @param event type + */ +public interface ExceptionHandler { + /** + * Handle exception during event processing + * + * @param ex exception + * @param sequence sequence number + * @param event the event + */ + void handleEventException(Throwable ex, long sequence, T event); + + /** Handle exception during processor start */ + void handleOnStartException(Throwable ex); + + /** Handle exception during processor shutdown */ + void handleOnShutdownException(Throwable ex); +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java new file mode 100644 index 0000000000000..e716eddefde7e --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java @@ -0,0 +1,259 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; + +import sun.misc.Unsafe; + +import java.lang.reflect.Field; +import java.util.concurrent.locks.LockSupport; + +/** + * Multi-producer sequencer for coordinating concurrent event publishing + * + *

Manages sequence allocation and tracking for multiple producer threads: + * + *

    + *
  • Lock-free sequence claiming using CAS operations + *
  • Available buffer tracks out-of-order publishing + *
  • Gating sequence cache optimizes consumer progress checks + *
  • Backpressure mechanism prevents buffer overwrites + *
+ */ +public final class MultiProducerSequencer { + private static final Unsafe UNSAFE; + private static final long BASE; + private static final long SCALE; + + static { + try { + Field field = Unsafe.class.getDeclaredField("theUnsafe"); + field.setAccessible(true); + UNSAFE = (Unsafe) field.get(null); + + // Initialize array access offsets for available buffer + BASE = UNSAFE.arrayBaseOffset(int[].class); + SCALE = UNSAFE.arrayIndexScale(int[].class); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + private final int bufferSize; + protected final Sequence cursor = new Sequence(Sequence.INITIAL_VALUE); + protected volatile Sequence[] gatingSequences; + + // CRITICAL: Cache to avoid repeated getMinimumSequence calls + private final Sequence gatingSequenceCache = new Sequence(Sequence.INITIAL_VALUE); + + // CRITICAL: Available buffer tracks published sequences + private final int[] availableBuffer; + private final int indexMask; + private final int indexShift; + + public MultiProducerSequencer(int bufferSize, Sequence[] gatingSequences) { + if (bufferSize < 1) { + throw new IllegalArgumentException("bufferSize must not be less than 1"); + } + if (Integer.bitCount(bufferSize) != 1) { + throw new IllegalArgumentException("bufferSize must be a power of 2"); + } + + this.bufferSize = bufferSize; + this.gatingSequences = gatingSequences != null ? gatingSequences : new Sequence[0]; + this.availableBuffer = new int[bufferSize]; + this.indexMask = bufferSize - 1; + this.indexShift = log2(bufferSize); + + initialiseAvailableBuffer(); + } + + /** + * Claim next n sequences for publishing + * + *

Uses CAS loop to atomically claim sequence numbers. Implements backpressure by parking when + * buffer is full. + * + * @param n number of sequences to claim + * @return highest claimed sequence number + */ + public long next(int n) { + if (n < 1) { + throw new IllegalArgumentException("n must be > 0"); + } + + long current; + long next; + + do { + current = cursor.get(); + next = current + n; + + long wrapPoint = next - bufferSize; + long cachedGatingSequence = gatingSequenceCache.get(); + + if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current) { + long gatingSequence = Sequence.getMinimumSequence(gatingSequences, current); + + if (wrapPoint > gatingSequence) { + LockSupport.parkNanos(1); + continue; + } + + gatingSequenceCache.set(gatingSequence); + } else if (cursor.compareAndSet(current, next)) { + break; + } + } while (true); + + return next; + } + + /** Publish sequence */ + public void publish(final long sequence) { + setAvailable(sequence); + } + + /** Publish batch */ + public void publish(long lo, long hi) { + for (long l = lo; l <= hi; l++) { + setAvailable(l); + } + } + + /** CORE: Check if available - MUST use Unsafe.getIntVolatile */ + public boolean isAvailable(long sequence) { + int index = calculateIndex(sequence); + int flag = calculateAvailabilityFlag(sequence); + long bufferAddress = (index * SCALE) + BASE; + return UNSAFE.getIntVolatile(availableBuffer, bufferAddress) == flag; + } + + /** CORE: Get highest published - exact same algorithm */ + public long getHighestPublishedSequence(long lowerBound, long availableSequence) { + for (long sequence = lowerBound; sequence <= availableSequence; sequence++) { + if (!isAvailable(sequence)) { + return sequence - 1; + } + } + return availableSequence; + } + + public Sequence getCursor() { + return cursor; + } + + public int getBufferSize() { + return bufferSize; + } + + public long remainingCapacity() { + long consumed = Sequence.getMinimumSequence(gatingSequences, cursor.get()); + long produced = cursor.get(); + return bufferSize - (produced - consumed); + } + + /** + * Add gating sequences for consumer tracking + * + *

Atomically adds sequences to track consumer progress + * + * @param gatingSequences consumer sequences to add + */ + public final void addGatingSequences(Sequence... gatingSequences) { + SequenceGroups.addSequences(this, this.cursor, gatingSequences); + } + + /** + * Remove a gating sequence + * + * @param sequence sequence to remove + * @return true if sequence was found and removed + */ + public boolean removeGatingSequence(Sequence sequence) { + return SequenceGroups.removeSequence(this, sequence); + } + + /** + * Get the minimum sequence from all consumers + * + * @return minimum gating sequence + */ + public long getMinimumSequence() { + return Sequence.getMinimumSequence(gatingSequences, cursor.get()); + } + + /** + * Create a sequence barrier for consumers + * + * @param sequencesToTrack upstream sequences to wait for + * @return new sequence barrier + */ + public SequenceBarrier newBarrier(Sequence... sequencesToTrack) { + return new SequenceBarrier(this, sequencesToTrack); + } + + /** Initialize available buffer */ + private void initialiseAvailableBuffer() { + for (int i = availableBuffer.length - 1; i != 0; i--) { + setAvailableBufferValue(i, -1); + } + setAvailableBufferValue(0, -1); + } + + /** + * CORE: Set available - MUST use Unsafe.putOrderedInt + * + *

putOrderedInt provides: - Store-store barrier (not full fence) - Cheaper than volatile write + * - Sufficient for this use case + */ + private void setAvailable(final long sequence) { + setAvailableBufferValue(calculateIndex(sequence), calculateAvailabilityFlag(sequence)); + } + + /** CRITICAL: Use Unsafe.putOrderedInt for correct memory semantics */ + private void setAvailableBufferValue(int index, int flag) { + long bufferAddress = (index * SCALE) + BASE; + UNSAFE.putOrderedInt(availableBuffer, bufferAddress, flag); + } + + /** Calculate availability flag */ + private int calculateAvailabilityFlag(final long sequence) { + return (int) (sequence >>> indexShift); + } + + /** Calculate index */ + private int calculateIndex(final long sequence) { + return ((int) sequence) & indexMask; + } + + /** + * Calculate log2 for index shift calculation + * + * @param i input value (must be power of 2) + * @return log2 of input + */ + private static int log2(int i) { + int r = 0; + while ((i >>= 1) != 0) { + ++r; + } + return r; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java new file mode 100644 index 0000000000000..59746b756ce82 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java @@ -0,0 +1,363 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; + +import sun.misc.Unsafe; + +import java.lang.reflect.Field; + +/** + * Left-hand side padding for cache line alignment + * + *

Prevents false sharing by ensuring RingBuffer fields don't share cache lines with preceding + * objects + */ +abstract class RingBufferPad { + protected long p1, p2, p3, p4, p5, p6, p7; +} + +/** + * Core fields for RingBuffer implementation + * + *

Contains the actual event storage array and sequencing state + */ +abstract class RingBufferFields extends RingBufferPad { + /** Unsafe instance for direct memory access */ + private static final Unsafe UNSAFE; + + /** Base offset of Object array in memory */ + private static final long ARRAY_BASE; + + /** Number of padding elements at array boundaries (128 bytes / element size) */ + private static final int BUFFER_PAD; + + /** Actual base offset for accessing array elements (includes front padding) */ + private static final long REF_ARRAY_BASE; + + /** Bit shift for calculating element offset (2 for 32-bit, 3 for 64-bit) */ + private static final int REF_ELEMENT_SHIFT; + + static { + try { + // Get Unsafe instance through reflection + Field field = Unsafe.class.getDeclaredField("theUnsafe"); + field.setAccessible(true); + UNSAFE = (Unsafe) field.get(null); + + // Determine pointer size and calculate shift + final int scale = UNSAFE.arrayIndexScale(Object[].class); + if (4 == scale) { + REF_ELEMENT_SHIFT = 2; // 32-bit pointers: index << 2 = index * 4 + } else if (8 == scale) { + REF_ELEMENT_SHIFT = 3; // 64-bit pointers: index << 3 = index * 8 + } else { + throw new IllegalStateException("Unknown pointer size"); + } + + // Calculate padding size (128 bytes / element size) + BUFFER_PAD = 128 / scale; + ARRAY_BASE = UNSAFE.arrayBaseOffset(Object[].class); + + // Skip front padding to start from actual data + REF_ARRAY_BASE = ARRAY_BASE + (BUFFER_PAD << REF_ELEMENT_SHIFT); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + /** Pre-allocated event storage with padding to prevent false sharing */ + private final Object[] entries; + + /** Total number of events in the buffer (must be power of 2) */ + protected final int bufferSize; + + /** Mask for fast modulo operation (bufferSize - 1) */ + protected final int indexMask; + + /** Sequencer for managing producer/consumer coordination */ + protected final MultiProducerSequencer sequencer; + + /** + * Initialize ring buffer fields + * + * @param eventFactory factory for pre-allocating events + * @param sequencer multi-producer sequencer + */ + RingBufferFields(EventFactory eventFactory, MultiProducerSequencer sequencer) { + this.sequencer = sequencer; + this.bufferSize = sequencer.getBufferSize(); + + if (bufferSize < 1) { + throw new IllegalArgumentException("bufferSize must not be less than 1"); + } + if (Integer.bitCount(bufferSize) != 1) { + throw new IllegalArgumentException("bufferSize must be a power of 2"); + } + + this.indexMask = bufferSize - 1; + // Allocate array with padding on both sides to prevent false sharing + this.entries = new Object[bufferSize + 2 * BUFFER_PAD]; + fill(eventFactory); + } + + /** + * Pre-allocate all events in the buffer + * + * @param eventFactory factory for creating event instances + */ + private void fill(EventFactory eventFactory) { + for (int i = 0; i < bufferSize; i++) { + // Store events starting after front padding + entries[BUFFER_PAD + i] = eventFactory.newInstance(); + } + } + + /** + * Get event at sequence using direct memory access + * + * @param sequence sequence number + * @return event at the sequence position + */ + @SuppressWarnings("unchecked") + protected final E elementAt(long sequence) { + // Use Unsafe for lock-free array access with proper memory barriers + return (E) + UNSAFE.getObject(entries, REF_ARRAY_BASE + ((sequence & indexMask) << REF_ELEMENT_SHIFT)); + } +} + +/** + * Lock-free ring buffer for storing pre-allocated event objects + * + *

Supports multi-producer concurrent access with zero-garbage design. Events are pre-allocated + * and reused, avoiding GC pressure. Uses cache line padding to prevent false sharing. + * + * @param event type + */ +public final class RingBuffer extends RingBufferFields { + /** Initial cursor value for the ring buffer */ + public static final long INITIAL_CURSOR_VALUE = Sequence.INITIAL_VALUE; + + /** + * Right-hand side padding for cache line alignment + * + *

Prevents false sharing by ensuring RingBuffer fields don't share cache lines with following + * objects + */ + protected long p1, p2, p3, p4, p5, p6, p7; + + /** + * Construct a RingBuffer with given factory and sequencer + * + * @param eventFactory factory to create and pre-allocate events + * @param sequencer multi-producer sequencer for sequence management + */ + private RingBuffer(EventFactory eventFactory, MultiProducerSequencer sequencer) { + super(eventFactory, sequencer); + } + + /** + * Create a multi-producer RingBuffer + * + *

Supports concurrent publishing from multiple threads using lock-free CAS operations + * + * @param factory event factory for creating event instances + * @param bufferSize buffer size (must be power of 2) + * @param event type + * @return newly created ring buffer + */ + public static RingBuffer createMultiProducer(EventFactory factory, int bufferSize) { + MultiProducerSequencer sequencer = new MultiProducerSequencer(bufferSize, new Sequence[0]); + return new RingBuffer<>(factory, sequencer); + } + + /** + * Get the event at a specific sequence + * + * @param sequence sequence number to retrieve + * @return event at the given sequence + */ + public E get(long sequence) { + return elementAt(sequence); + } + + /** + * Claim the next sequence for publishing + * + *

Blocks if buffer is full until space becomes available + * + * @return claimed sequence number + */ + public long next() { + return sequencer.next(1); + } + + /** + * Claim next n sequences for batch publishing + * + * @param n number of sequences to claim + * @return highest claimed sequence number + */ + public long next(int n) { + return sequencer.next(n); + } + + /** + * Publish a single sequence + * + *

Makes the event at this sequence visible to consumers + * + * @param sequence sequence to publish + */ + public void publish(long sequence) { + sequencer.publish(sequence); + } + + /** + * Publish a batch of sequences + * + * @param lo lowest sequence in the batch (inclusive) + * @param hi highest sequence in the batch (inclusive) + */ + public void publish(long lo, long hi) { + sequencer.publish(lo, hi); + } + + /** + * Publish event using a translator function + * + *

Provides a higher-level API for publishing events with custom translation logic + * + * @param translator function to populate the event + * @param arg0 argument passed to translator + * @param argument type + */ + public void publishEvent(EventTranslator translator, A arg0) { + final long sequence = sequencer.next(1); + translateAndPublish(translator, sequence, arg0); + } + + /** + * Translate event and publish atomically + * + * @param translator event translator function + * @param sequence claimed sequence number + * @param arg0 argument for translation + * @param argument type + */ + private void translateAndPublish(EventTranslator translator, long sequence, A arg0) { + try { + translator.translateTo(get(sequence), sequence, arg0); + } finally { + sequencer.publish(sequence); + } + } + + /** + * Add gating sequences for consumer tracking + * + *

Gating sequences represent consumer progress and prevent overwriting unprocessed events + * + * @param gatingSequences consumer sequences to track + */ + public void addGatingSequences(Sequence... gatingSequences) { + sequencer.addGatingSequences(gatingSequences); + } + + /** + * Remove a gating sequence + * + *

Should be called when a consumer is shut down + * + * @param sequence sequence to remove + * @return true if sequence was found and removed + */ + public boolean removeGatingSequence(Sequence sequence) { + return sequencer.removeGatingSequence(sequence); + } + + /** + * Get the minimum sequence from all consumers + * + *

Represents the slowest consumer's progress + * + * @return minimum gating sequence + */ + public long getMinimumGatingSequence() { + return sequencer.getMinimumSequence(); + } + + /** + * Create a sequence barrier for consumers + * + *

Barrier coordinates when events become available for processing + * + * @param sequencesToTrack upstream sequences to wait for + * @return new sequence barrier + */ + public SequenceBarrier newBarrier(Sequence... sequencesToTrack) { + return sequencer.newBarrier(sequencesToTrack); + } + + /** + * Get current producer cursor position + * + * @return current cursor value + */ + public long getCursor() { + return sequencer.getCursor().get(); + } + + /** + * Get the buffer size + * + * @return configured buffer size + */ + public int getBufferSize() { + return bufferSize; + } + + /** + * Get remaining capacity in the buffer + * + * @return number of available slots + */ + public long remainingCapacity() { + return sequencer.remainingCapacity(); + } + + /** + * Function interface for translating data into events + * + * @param event type + * @param argument type + */ + @FunctionalInterface + public interface EventTranslator { + /** + * Translate argument into event + * + * @param event pre-allocated event to populate + * @param sequence sequence number for this event + * @param arg source data + */ + void translateTo(E event, long sequence, A arg); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java new file mode 100644 index 0000000000000..db8a7b3aa7bc1 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java @@ -0,0 +1,153 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; + +import sun.misc.Unsafe; + +import java.lang.reflect.Field; + +/** Left-hand side padding for cache line alignment */ +class LhsPadding { + protected long p1, p2, p3, p4, p5, p6, p7; +} + +/** Value class holding the actual sequence */ +class Value extends LhsPadding { + protected volatile long value; +} + +/** Right-hand side padding for cache line alignment */ +class RhsPadding extends Value { + protected long p9, p10, p11, p12, p13, p14, p15; +} + +/** + * Lock-free sequence counter with cache line padding + * + *

Key design features: + * + *

+ */ +public class Sequence extends RhsPadding { + public static final long INITIAL_VALUE = -1L; + private static final Unsafe UNSAFE; + private static final long VALUE_OFFSET; + + static { + try { + Field field = Unsafe.class.getDeclaredField("theUnsafe"); + field.setAccessible(true); + UNSAFE = (Unsafe) field.get(null); + // CRITICAL: Get offset of 'value' field from Value class + VALUE_OFFSET = UNSAFE.objectFieldOffset(Value.class.getDeclaredField("value")); + } catch (final Exception e) { + throw new RuntimeException(e); + } + } + + /** Create sequence with initial value -1 */ + public Sequence() { + this(INITIAL_VALUE); + } + + /** + * Create sequence with specified initial value + * + * @param initialValue initial value + */ + public Sequence(final long initialValue) { + UNSAFE.putOrderedLong(this, VALUE_OFFSET, initialValue); + } + + /** Volatile read */ + public long get() { + return value; + } + + /** + * Ordered write (store-store barrier only) + * + *

CRITICAL: Cheaper than volatile write, sufficient for most cases + */ + public void set(final long value) { + UNSAFE.putOrderedLong(this, VALUE_OFFSET, value); + } + + /** + * Volatile write (full memory barrier) + * + *

Use when need full visibility guarantees + */ + public void setVolatile(final long value) { + UNSAFE.putLongVolatile(this, VALUE_OFFSET, value); + } + + /** + * CAS operation - CORE for lock-free design + * + * @param expectedValue expected current value + * @param newValue new value + * @return true if successful + */ + public boolean compareAndSet(final long expectedValue, final long newValue) { + return UNSAFE.compareAndSwapLong(this, VALUE_OFFSET, expectedValue, newValue); + } + + /** Atomically increment */ + public long incrementAndGet() { + return addAndGet(1L); + } + + /** Atomically add */ + public long addAndGet(final long increment) { + long currentValue; + long newValue; + + do { + currentValue = get(); + newValue = currentValue + increment; + } while (!compareAndSet(currentValue, newValue)); + + return newValue; + } + + @Override + public String toString() { + return Long.toString(get()); + } + + /** Get minimum sequence from array - CORE utility method */ + public static long getMinimumSequence(final Sequence[] sequences, long minimum) { + for (int i = 0, n = sequences.length; i < n; i++) { + long value = sequences[i].get(); + minimum = Math.min(minimum, value); + } + return minimum; + } + + public static long getMinimumSequence(final Sequence[] sequences) { + return getMinimumSequence(sequences, Long.MAX_VALUE); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceBarrier.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceBarrier.java new file mode 100644 index 0000000000000..ef380a4f80d14 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceBarrier.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; + +/** + * Sequence barrier for consumer coordination + * + *

Simplified from Disruptor (removed Alert mechanism - IoTDB doesn't need it) + * + *

MUST preserve: + * + *

    + *
  • waitFor() logic for waiting sequences + *
  • Scan available buffer for out-of-order publishing + *
+ */ +public class SequenceBarrier { + private final MultiProducerSequencer sequencer; + private final Sequence[] dependentSequences; + + public SequenceBarrier(MultiProducerSequencer sequencer, Sequence[] dependentSequences) { + this.sequencer = sequencer; + this.dependentSequences = dependentSequences != null ? dependentSequences : new Sequence[0]; + } + + /** + * CORE: Wait for sequence to become available (MUST keep logic) + * + * @param sequence sequence to wait for + * @return highest available sequence + * @throws InterruptedException if interrupted + */ + public long waitFor(long sequence) throws InterruptedException { + // Wait for cursor + long availableSequence; + while ((availableSequence = sequencer.getCursor().get()) < sequence) { + Thread.sleep(1); + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException(); + } + } + + // Wait for dependent sequences + if (dependentSequences.length > 0) { + while (Sequence.getMinimumSequence(dependentSequences) < sequence) { + Thread.sleep(1); + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException(); + } + } + } + + // CORE: Scan available buffer for highest continuously published sequence + return sequencer.getHighestPublishedSequence(sequence, availableSequence); + } + + public long getCursor() { + return sequencer.getCursor().get(); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java new file mode 100644 index 0000000000000..2c0f8a7a6a7a7 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java @@ -0,0 +1,118 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; + +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; + +/** + * Utility for atomic management of sequence arrays + * + *

Provides thread-safe operations for adding and removing sequences from gating sequence arrays + * used to track consumer progress + */ +final class SequenceGroups { + + /** Field updater for atomic array replacement */ + private static final AtomicReferenceFieldUpdater + SEQUENCE_UPDATER = + AtomicReferenceFieldUpdater.newUpdater( + MultiProducerSequencer.class, Sequence[].class, "gatingSequences"); + + /** + * Atomically add sequences to the gating sequence array + * + *

Uses CAS loop to ensure thread-safe addition even under concurrent modification + * + * @param sequencer the multi-producer sequencer + * @param cursor the current cursor sequence + * @param sequencesToAdd sequences to add + */ + static void addSequences( + final MultiProducerSequencer sequencer, + final Sequence cursor, + final Sequence... sequencesToAdd) { + long cursorSequence; + Sequence[] updatedSequences; + Sequence[] currentSequences; + + do { + currentSequences = sequencer.gatingSequences; + updatedSequences = new Sequence[currentSequences.length + sequencesToAdd.length]; + System.arraycopy(currentSequences, 0, updatedSequences, 0, currentSequences.length); + + cursorSequence = cursor.get(); + + int index = currentSequences.length; + for (Sequence sequence : sequencesToAdd) { + sequence.set(cursorSequence); + updatedSequences[index++] = sequence; + } + } while (!SEQUENCE_UPDATER.compareAndSet(sequencer, currentSequences, updatedSequences)); + + cursorSequence = cursor.get(); + for (Sequence sequence : sequencesToAdd) { + sequence.set(cursorSequence); + } + } + + /** + * Remove sequence from the group + * + * @param sequencer the sequencer + * @param sequence sequence to remove + * @return true if removed + */ + static boolean removeSequence(final MultiProducerSequencer sequencer, final Sequence sequence) { + int numToRemove; + Sequence[] oldSequences; + Sequence[] newSequences; + + do { + oldSequences = sequencer.gatingSequences; + numToRemove = countMatching(oldSequences, sequence); + + if (0 == numToRemove) { + break; + } + + final int oldSize = oldSequences.length; + newSequences = new Sequence[oldSize - numToRemove]; + + for (int i = 0, pos = 0; i < oldSize; i++) { + final Sequence testSequence = oldSequences[i]; + if (sequence != testSequence) { + newSequences[pos++] = testSequence; + } + } + } while (!SEQUENCE_UPDATER.compareAndSet(sequencer, oldSequences, newSequences)); + + return numToRemove != 0; + } + + private static int countMatching(Sequence[] values, final Sequence toMatch) { + int numToRemove = 0; + for (Sequence value : values) { + if (value == toMatch) { + numToRemove++; + } + } + return numToRemove; + } +} From 3dd3262836179b05f8c57a68a9ebd86e2dc396de Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 22 Oct 2025 16:29:34 +0800 Subject: [PATCH 02/11] fix --- .../dataregion/realtime/disruptor/RingBuffer.java | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java index 59746b756ce82..d84ea89365b9b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java @@ -22,6 +22,8 @@ import sun.misc.Unsafe; import java.lang.reflect.Field; +import java.security.AccessController; +import java.security.PrivilegedExceptionAction; /** * Left-hand side padding for cache line alignment @@ -56,10 +58,13 @@ abstract class RingBufferFields extends RingBufferPad { static { try { - // Get Unsafe instance through reflection - Field field = Unsafe.class.getDeclaredField("theUnsafe"); - field.setAccessible(true); - UNSAFE = (Unsafe) field.get(null); + final PrivilegedExceptionAction action = () -> { + Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe"); + theUnsafe.setAccessible(true); + return (Unsafe) theUnsafe.get(null); + }; + + UNSAFE = AccessController.doPrivileged(action); // Determine pointer size and calculate shift final int scale = UNSAFE.arrayIndexScale(Object[].class); From f17f3fa412918b599077ad836ead64287656a11b Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 22 Oct 2025 17:30:04 +0800 Subject: [PATCH 03/11] Update RingBuffer.java --- .../dataregion/realtime/disruptor/RingBuffer.java | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java index d84ea89365b9b..64cd8e26e1889 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java @@ -58,11 +58,12 @@ abstract class RingBufferFields extends RingBufferPad { static { try { - final PrivilegedExceptionAction action = () -> { - Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe"); - theUnsafe.setAccessible(true); - return (Unsafe) theUnsafe.get(null); - }; + final PrivilegedExceptionAction action = + () -> { + Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe"); + theUnsafe.setAccessible(true); + return (Unsafe) theUnsafe.get(null); + }; UNSAFE = AccessController.doPrivileged(action); From 0017abbf6db27121c33a6fa5bdbf08f7f481f5c7 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 22 Oct 2025 17:48:06 +0800 Subject: [PATCH 04/11] may-fix --- .../realtime/disruptor/MultiProducerSequencer.java | 13 ++++++++++--- .../dataregion/realtime/disruptor/Sequence.java | 13 ++++++++++--- 2 files changed, 20 insertions(+), 6 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java index e716eddefde7e..8c2719e7513ee 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java @@ -22,6 +22,8 @@ import sun.misc.Unsafe; import java.lang.reflect.Field; +import java.security.AccessController; +import java.security.PrivilegedExceptionAction; import java.util.concurrent.locks.LockSupport; /** @@ -43,9 +45,14 @@ public final class MultiProducerSequencer { static { try { - Field field = Unsafe.class.getDeclaredField("theUnsafe"); - field.setAccessible(true); - UNSAFE = (Unsafe) field.get(null); + final PrivilegedExceptionAction action = + () -> { + Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe"); + theUnsafe.setAccessible(true); + return (Unsafe) theUnsafe.get(null); + }; + + UNSAFE = AccessController.doPrivileged(action); // Initialize array access offsets for available buffer BASE = UNSAFE.arrayBaseOffset(int[].class); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java index db8a7b3aa7bc1..4082c7706e11a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java @@ -22,6 +22,8 @@ import sun.misc.Unsafe; import java.lang.reflect.Field; +import java.security.AccessController; +import java.security.PrivilegedExceptionAction; /** Left-hand side padding for cache line alignment */ class LhsPadding { @@ -57,9 +59,14 @@ public class Sequence extends RhsPadding { static { try { - Field field = Unsafe.class.getDeclaredField("theUnsafe"); - field.setAccessible(true); - UNSAFE = (Unsafe) field.get(null); + final PrivilegedExceptionAction action = + () -> { + Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe"); + theUnsafe.setAccessible(true); + return (Unsafe) theUnsafe.get(null); + }; + + UNSAFE = AccessController.doPrivileged(action); // CRITICAL: Get offset of 'value' field from Value class VALUE_OFFSET = UNSAFE.objectFieldOffset(Value.class.getDeclaredField("value")); } catch (final Exception e) { From 13496b555b63dd9dd48248a6e96b5a77335a91f0 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 23 Oct 2025 10:59:50 +0800 Subject: [PATCH 05/11] modify code --- dependencies.json | 1 - iotdb-core/datanode/pom.xml | 4 -- .../realtime/assigner/DisruptorQueue.java | 2 +- .../disruptor/BatchEventProcessor.java | 2 +- .../realtime/disruptor/Disruptor.java | 1 - .../disruptor/MultiProducerSequencer.java | 51 ++-------------- .../realtime/disruptor/RingBuffer.java | 60 +------------------ .../realtime/disruptor/Sequence.java | 53 ++-------------- pom.xml | 6 -- 9 files changed, 16 insertions(+), 164 deletions(-) diff --git a/dependencies.json b/dependencies.json index c2a255d7ba4dc..2241b90734364 100644 --- a/dependencies.json +++ b/dependencies.json @@ -26,7 +26,6 @@ "com.google.guava:listenablefuture", "com.google.j2objc:j2objc-annotations", "com.h2database:h2-mvstore", - "com.lmax:disruptor", "com.nimbusds:content-type", "com.nimbusds:lang-tag", "com.nimbusds:nimbus-jose-jwt", diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml index 490780a87a738..e597b91aa0720 100644 --- a/iotdb-core/datanode/pom.xml +++ b/iotdb-core/datanode/pom.xml @@ -308,10 +308,6 @@ com.google.guava guava - - com.lmax - disruptor - org.java-websocket Java-WebSocket diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java index 636646e95c1bb..cb4aba1e15adb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java @@ -99,7 +99,7 @@ public boolean isClosed() { private static class EventContainer { - private PipeRealtimeEvent event; + private volatile PipeRealtimeEvent event; private EventContainer() {} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java index b9709f65da4ac..5891ebbdd81e3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java @@ -43,7 +43,7 @@ public final class BatchEventProcessor implements Runnable { private final RingBuffer ringBuffer; private final SequenceBarrier sequenceBarrier; private final EventHandler eventHandler; - private final Sequence sequence = new Sequence(-1L); + private final Sequence sequence = new Sequence(); private ExceptionHandler exceptionHandler = new DefaultExceptionHandler<>(); private volatile boolean running = true; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java index 00a8da3aa13e8..5973544cdc948 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java @@ -106,7 +106,6 @@ public RingBuffer start() { return ringBuffer; } - /** Shutdown - MUST keep for IoTDB */ public void shutdown() { if (!started) { return; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java index 8c2719e7513ee..5675f252569b5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java @@ -19,55 +19,16 @@ package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; -import sun.misc.Unsafe; - -import java.lang.reflect.Field; -import java.security.AccessController; -import java.security.PrivilegedExceptionAction; import java.util.concurrent.locks.LockSupport; -/** - * Multi-producer sequencer for coordinating concurrent event publishing - * - *

Manages sequence allocation and tracking for multiple producer threads: - * - *

    - *
  • Lock-free sequence claiming using CAS operations - *
  • Available buffer tracks out-of-order publishing - *
  • Gating sequence cache optimizes consumer progress checks - *
  • Backpressure mechanism prevents buffer overwrites - *
- */ public final class MultiProducerSequencer { - private static final Unsafe UNSAFE; - private static final long BASE; - private static final long SCALE; - - static { - try { - final PrivilegedExceptionAction action = - () -> { - Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe"); - theUnsafe.setAccessible(true); - return (Unsafe) theUnsafe.get(null); - }; - - UNSAFE = AccessController.doPrivileged(action); - - // Initialize array access offsets for available buffer - BASE = UNSAFE.arrayBaseOffset(int[].class); - SCALE = UNSAFE.arrayIndexScale(int[].class); - } catch (Exception e) { - throw new RuntimeException(e); - } - } private final int bufferSize; - protected final Sequence cursor = new Sequence(Sequence.INITIAL_VALUE); - protected volatile Sequence[] gatingSequences; + private final Sequence cursor = new Sequence(); + volatile Sequence[] gatingSequences; // CRITICAL: Cache to avoid repeated getMinimumSequence calls - private final Sequence gatingSequenceCache = new Sequence(Sequence.INITIAL_VALUE); + private final Sequence gatingSequenceCache = new Sequence(); // CRITICAL: Available buffer tracks published sequences private final int[] availableBuffer; @@ -148,8 +109,7 @@ public void publish(long lo, long hi) { public boolean isAvailable(long sequence) { int index = calculateIndex(sequence); int flag = calculateAvailabilityFlag(sequence); - long bufferAddress = (index * SCALE) + BASE; - return UNSAFE.getIntVolatile(availableBuffer, bufferAddress) == flag; + return availableBuffer[index] == flag; } /** CORE: Get highest published - exact same algorithm */ @@ -236,8 +196,7 @@ private void setAvailable(final long sequence) { /** CRITICAL: Use Unsafe.putOrderedInt for correct memory semantics */ private void setAvailableBufferValue(int index, int flag) { - long bufferAddress = (index * SCALE) + BASE; - UNSAFE.putOrderedInt(availableBuffer, bufferAddress, flag); + availableBuffer[index] = flag; } /** Calculate availability flag */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java index 64cd8e26e1889..32f1fd12dd620 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java @@ -19,12 +19,6 @@ package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; -import sun.misc.Unsafe; - -import java.lang.reflect.Field; -import java.security.AccessController; -import java.security.PrivilegedExceptionAction; - /** * Left-hand side padding for cache line alignment * @@ -41,53 +35,6 @@ abstract class RingBufferPad { *

Contains the actual event storage array and sequencing state */ abstract class RingBufferFields extends RingBufferPad { - /** Unsafe instance for direct memory access */ - private static final Unsafe UNSAFE; - - /** Base offset of Object array in memory */ - private static final long ARRAY_BASE; - - /** Number of padding elements at array boundaries (128 bytes / element size) */ - private static final int BUFFER_PAD; - - /** Actual base offset for accessing array elements (includes front padding) */ - private static final long REF_ARRAY_BASE; - - /** Bit shift for calculating element offset (2 for 32-bit, 3 for 64-bit) */ - private static final int REF_ELEMENT_SHIFT; - - static { - try { - final PrivilegedExceptionAction action = - () -> { - Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe"); - theUnsafe.setAccessible(true); - return (Unsafe) theUnsafe.get(null); - }; - - UNSAFE = AccessController.doPrivileged(action); - - // Determine pointer size and calculate shift - final int scale = UNSAFE.arrayIndexScale(Object[].class); - if (4 == scale) { - REF_ELEMENT_SHIFT = 2; // 32-bit pointers: index << 2 = index * 4 - } else if (8 == scale) { - REF_ELEMENT_SHIFT = 3; // 64-bit pointers: index << 3 = index * 8 - } else { - throw new IllegalStateException("Unknown pointer size"); - } - - // Calculate padding size (128 bytes / element size) - BUFFER_PAD = 128 / scale; - ARRAY_BASE = UNSAFE.arrayBaseOffset(Object[].class); - - // Skip front padding to start from actual data - REF_ARRAY_BASE = ARRAY_BASE + (BUFFER_PAD << REF_ELEMENT_SHIFT); - } catch (Exception e) { - throw new RuntimeException(e); - } - } - /** Pre-allocated event storage with padding to prevent false sharing */ private final Object[] entries; @@ -119,7 +66,7 @@ abstract class RingBufferFields extends RingBufferPad { this.indexMask = bufferSize - 1; // Allocate array with padding on both sides to prevent false sharing - this.entries = new Object[bufferSize + 2 * BUFFER_PAD]; + this.entries = new Object[bufferSize]; fill(eventFactory); } @@ -131,7 +78,7 @@ abstract class RingBufferFields extends RingBufferPad { private void fill(EventFactory eventFactory) { for (int i = 0; i < bufferSize; i++) { // Store events starting after front padding - entries[BUFFER_PAD + i] = eventFactory.newInstance(); + entries[i] = eventFactory.newInstance(); } } @@ -144,8 +91,7 @@ private void fill(EventFactory eventFactory) { @SuppressWarnings("unchecked") protected final E elementAt(long sequence) { // Use Unsafe for lock-free array access with proper memory barriers - return (E) - UNSAFE.getObject(entries, REF_ARRAY_BASE + ((sequence & indexMask) << REF_ELEMENT_SHIFT)); + return (E) entries[(int) (sequence & indexMask)]; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java index 4082c7706e11a..a0132c841f440 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java @@ -19,11 +19,7 @@ package org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor; -import sun.misc.Unsafe; - -import java.lang.reflect.Field; -import java.security.AccessController; -import java.security.PrivilegedExceptionAction; +import java.util.concurrent.atomic.AtomicLong; /** Left-hand side padding for cache line alignment */ class LhsPadding { @@ -32,7 +28,7 @@ class LhsPadding { /** Value class holding the actual sequence */ class Value extends LhsPadding { - protected volatile long value; + protected AtomicLong value = new AtomicLong(); } /** Right-hand side padding for cache line alignment */ @@ -54,43 +50,15 @@ class RhsPadding extends Value { */ public class Sequence extends RhsPadding { public static final long INITIAL_VALUE = -1L; - private static final Unsafe UNSAFE; - private static final long VALUE_OFFSET; - - static { - try { - final PrivilegedExceptionAction action = - () -> { - Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe"); - theUnsafe.setAccessible(true); - return (Unsafe) theUnsafe.get(null); - }; - - UNSAFE = AccessController.doPrivileged(action); - // CRITICAL: Get offset of 'value' field from Value class - VALUE_OFFSET = UNSAFE.objectFieldOffset(Value.class.getDeclaredField("value")); - } catch (final Exception e) { - throw new RuntimeException(e); - } - } /** Create sequence with initial value -1 */ public Sequence() { - this(INITIAL_VALUE); - } - - /** - * Create sequence with specified initial value - * - * @param initialValue initial value - */ - public Sequence(final long initialValue) { - UNSAFE.putOrderedLong(this, VALUE_OFFSET, initialValue); + value.set(INITIAL_VALUE); } /** Volatile read */ public long get() { - return value; + return value.get(); } /** @@ -99,16 +67,7 @@ public long get() { *

CRITICAL: Cheaper than volatile write, sufficient for most cases */ public void set(final long value) { - UNSAFE.putOrderedLong(this, VALUE_OFFSET, value); - } - - /** - * Volatile write (full memory barrier) - * - *

Use when need full visibility guarantees - */ - public void setVolatile(final long value) { - UNSAFE.putLongVolatile(this, VALUE_OFFSET, value); + this.value.set(value); } /** @@ -119,7 +78,7 @@ public void setVolatile(final long value) { * @return true if successful */ public boolean compareAndSet(final long expectedValue, final long newValue) { - return UNSAFE.compareAndSwapLong(this, VALUE_OFFSET, expectedValue, newValue); + return value.compareAndSet(expectedValue, newValue); } /** Atomically increment */ diff --git a/pom.xml b/pom.xml index c86df5f482d93..dbefb680aebe8 100644 --- a/pom.xml +++ b/pom.xml @@ -76,7 +76,6 @@ 2.11.1 4.4 false - 3.4.4 1.21.1 4.2.19 11.1.0 @@ -456,11 +455,6 @@ h2-mvstore ${h2.version} - - com.lmax - disruptor - ${disruptor.version} - io.jsonwebtoken jjwt-impl From f10dc651276b792f64b6ff89eff239c0d18df057 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 23 Oct 2025 12:06:32 +0800 Subject: [PATCH 06/11] remove-dependency --- iotdb-core/datanode/pom.xml | 4 ---- pom.xml | 6 ------ 2 files changed, 10 deletions(-) diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml index 490780a87a738..e597b91aa0720 100644 --- a/iotdb-core/datanode/pom.xml +++ b/iotdb-core/datanode/pom.xml @@ -308,10 +308,6 @@ com.google.guava guava - - com.lmax - disruptor - org.java-websocket Java-WebSocket diff --git a/pom.xml b/pom.xml index c86df5f482d93..dbefb680aebe8 100644 --- a/pom.xml +++ b/pom.xml @@ -76,7 +76,6 @@ 2.11.1 4.4 false - 3.4.4 1.21.1 4.2.19 11.1.0 @@ -456,11 +455,6 @@ h2-mvstore ${h2.version} - - com.lmax - disruptor - ${disruptor.version} - io.jsonwebtoken jjwt-impl From ae5f570efe4d4a7d94731095335b3277064e2f35 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 23 Oct 2025 12:13:53 +0800 Subject: [PATCH 07/11] modify code --- .../disruptor/BatchEventProcessor.java | 1 - .../disruptor/MultiProducerSequencer.java | 66 +++++++++++++++---- 2 files changed, 52 insertions(+), 15 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java index 5891ebbdd81e3..07a2e2c5d9f89 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java +++ b/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 diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java index 5675f252569b5..7caf6b60c8f33 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java +++ b/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 + * + *

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 + * + *

Memory visibility guarantees: - Writers use lazySet() for store-store barrier (cheaper than + * volatile write) - Readers use get() for volatile read (ensures visibility across threads) + * + *

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 * - *

putOrderedInt provides: - Store-store barrier (not full fence) - Cheaper than volatile write - * - Sufficient for this use case + *

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 */ From d41db9480ad2b18b3f6fa85aad78a32edeffc679 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 23 Oct 2025 14:20:37 +0800 Subject: [PATCH 08/11] docs: Add documentation for Disruptor implementation based on LMAX Disruptor Added JavaDoc comments to all Disruptor-related classes to clarify that this implementation is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) and adapted for IoTDB's Pipe module. Modified files: - BatchEventProcessor.java - Disruptor.java - EventFactory.java - EventHandler.java - ExceptionHandler.java - MultiProducerSequencer.java - RingBuffer.java - Sequence.java - SequenceBarrier.java - SequenceGroups.java Each class now includes clear documentation about: 1. Origin from LMAX Disruptor 2. Key features preserved from the original 3. Simplifications made for IoTDB's use case --- .../realtime/disruptor/BatchEventProcessor.java | 5 +++-- .../dataregion/realtime/disruptor/Disruptor.java | 14 +++++++++----- .../realtime/disruptor/EventFactory.java | 3 +++ .../realtime/disruptor/EventHandler.java | 3 +++ .../realtime/disruptor/ExceptionHandler.java | 3 +++ .../realtime/disruptor/MultiProducerSequencer.java | 14 ++++++++++++++ .../dataregion/realtime/disruptor/RingBuffer.java | 3 +++ .../dataregion/realtime/disruptor/Sequence.java | 5 ++++- .../realtime/disruptor/SequenceBarrier.java | 5 +++-- .../realtime/disruptor/SequenceGroups.java | 5 ++++- 10 files changed, 49 insertions(+), 11 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java index 07a2e2c5d9f89..34930be977e16 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/BatchEventProcessor.java @@ -25,9 +25,10 @@ /** * Batch event processor for consuming events * - *

Simplified from Disruptor (removed complex lifecycle management) + *

This implementation is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) + * and simplified for IoTDB's Pipe module (removed complex lifecycle management). * - *

CORE algorithm (MUST preserve): + *

Core algorithm preserved from LMAX Disruptor: * *

    *
  • Batch processing loop diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java index 5973544cdc948..4cdbfe8fa7b92 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java @@ -25,12 +25,17 @@ import java.util.concurrent.ThreadFactory; /** - * High-level API for setting up a lock-free event processing pipeline + * Simplified Disruptor implementation for IoTDB Pipe * - *

    Provides a fluent interface for configuring producers and consumers. Internally manages a - * RingBuffer and event processor thread. + *

    This implementation is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) + * and simplified for IoTDB's specific use case in the Pipe module. * - *

    Configuration is simplified to multi-producer mode with blocking wait strategy. + *

    Key simplifications: + *

      + *
    • Single event handler support (no complex dependency graphs) + *
    • Simplified lifecycle management + *
    • Removed wait strategies (using simple sleep-based waiting) + *
    * * @param event type */ @@ -88,7 +93,6 @@ public void setDefaultExceptionHandler(ExceptionHandler exceptionHand } } - /** Start - MUST keep for IoTDB */ public RingBuffer start() { if (started) { throw new IllegalStateException("Disruptor already started"); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java index 8214988a28499..407025a6d2d50 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java @@ -22,6 +22,9 @@ /** * Event factory for pre-allocating events in RingBuffer * + *

    This interface is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) + * and adapted for IoTDB's Pipe module. + * * @param event type */ @FunctionalInterface diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java index b84d5895014c7..93cf7f3619feb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java @@ -22,6 +22,9 @@ /** * Event handler for processing events from RingBuffer * + *

    This interface is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) + * and adapted for IoTDB's Pipe module. + * * @param event type */ @FunctionalInterface diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java index e98138ca0f2f1..480f0395a60e3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java @@ -22,6 +22,9 @@ /** * Exception handler for event processing errors * + *

    This interface is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) + * and adapted for IoTDB's Pipe module. + * * @param event type */ public interface ExceptionHandler { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java index 7caf6b60c8f33..43780221a0125 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java @@ -22,6 +22,20 @@ import java.util.concurrent.atomic.AtomicIntegerArray; import java.util.concurrent.locks.LockSupport; +/** + * Multi-producer sequencer for coordinating concurrent publishers + * + *

    This implementation is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) + * and preserves the core lock-free multi-producer algorithm for IoTDB's Pipe module. + * + *

    Key features preserved from LMAX Disruptor: + *

      + *
    • Lock-free CAS-based sequence claiming + *
    • Availability buffer for out-of-order publishing detection + *
    • Backpressure via gating sequences + *
    • Cache line padding to prevent false sharing + *
    + */ public final class MultiProducerSequencer { /** Ring buffer size (must be power of 2) - immutable after construction */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java index 32f1fd12dd620..e43200adaffce 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java @@ -98,6 +98,9 @@ protected final E elementAt(long sequence) { /** * Lock-free ring buffer for storing pre-allocated event objects * + *

    This implementation is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) + * and preserves the core ring buffer algorithm for IoTDB's Pipe module. + * *

    Supports multi-producer concurrent access with zero-garbage design. Events are pre-allocated * and reused, avoiding GC pressure. Uses cache line padding to prevent false sharing. * diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java index a0132c841f440..1f1d3445969f1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Sequence.java @@ -39,11 +39,14 @@ class RhsPadding extends Value { /** * Lock-free sequence counter with cache line padding * + *

    This implementation is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) + * and preserves the core sequence tracking mechanism for IoTDB's Pipe module. + * *

    Key design features: * *

      *
    • Three-level inheritance ensures proper field ordering for padding - *
    • Uses volatile long with Unsafe for direct memory access + *
    • Uses AtomicLong for thread-safe atomic operations *
    • Cache line padding prevents false sharing between CPU cores *
    • Supports both ordered writes (cheaper) and volatile writes (stronger) *
    diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceBarrier.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceBarrier.java index ef380a4f80d14..4c8011eb1c225 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceBarrier.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceBarrier.java @@ -22,9 +22,10 @@ /** * Sequence barrier for consumer coordination * - *

    Simplified from Disruptor (removed Alert mechanism - IoTDB doesn't need it) + *

    This implementation is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) + * and simplified for IoTDB's Pipe module (removed Alert mechanism - IoTDB doesn't need it). * - *

    MUST preserve: + *

    Core features preserved from LMAX Disruptor: * *

      *
    • waitFor() logic for waiting sequences diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java index 2c0f8a7a6a7a7..61571f5757d59 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java @@ -24,8 +24,11 @@ /** * Utility for atomic management of sequence arrays * + *

      This implementation is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) + * and adapted for IoTDB's Pipe module. + * *

      Provides thread-safe operations for adding and removing sequences from gating sequence arrays - * used to track consumer progress + * used to track consumer progress. */ final class SequenceGroups { From deff732e1c5c087086bee4940c65efa7109d7a1e Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 23 Oct 2025 14:26:06 +0800 Subject: [PATCH 09/11] spotless --- .../pipe/source/dataregion/realtime/disruptor/Disruptor.java | 1 + .../source/dataregion/realtime/disruptor/EventFactory.java | 4 ++-- .../source/dataregion/realtime/disruptor/EventHandler.java | 4 ++-- .../dataregion/realtime/disruptor/ExceptionHandler.java | 4 ++-- .../dataregion/realtime/disruptor/MultiProducerSequencer.java | 1 + 5 files changed, 8 insertions(+), 6 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java index 4cdbfe8fa7b92..9070faa917347 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java @@ -31,6 +31,7 @@ * and simplified for IoTDB's specific use case in the Pipe module. * *

      Key simplifications: + * *

        *
      • Single event handler support (no complex dependency graphs) *
      • Simplified lifecycle management diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java index 407025a6d2d50..785033c68247b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventFactory.java @@ -22,8 +22,8 @@ /** * Event factory for pre-allocating events in RingBuffer * - *

        This interface is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) - * and adapted for IoTDB's Pipe module. + *

        This interface is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) and + * adapted for IoTDB's Pipe module. * * @param event type */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java index 93cf7f3619feb..1fc81a3762373 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/EventHandler.java @@ -22,8 +22,8 @@ /** * Event handler for processing events from RingBuffer * - *

        This interface is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) - * and adapted for IoTDB's Pipe module. + *

        This interface is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) and + * adapted for IoTDB's Pipe module. * * @param event type */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java index 480f0395a60e3..28396b51ffed1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/ExceptionHandler.java @@ -22,8 +22,8 @@ /** * Exception handler for event processing errors * - *

        This interface is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) - * and adapted for IoTDB's Pipe module. + *

        This interface is based on LMAX Disruptor (https://github.com/LMAX-Exchange/disruptor) and + * adapted for IoTDB's Pipe module. * * @param event type */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java index 43780221a0125..debfa1ab50b77 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java @@ -29,6 +29,7 @@ * and preserves the core lock-free multi-producer algorithm for IoTDB's Pipe module. * *

        Key features preserved from LMAX Disruptor: + * *

          *
        • Lock-free CAS-based sequence claiming *
        • Availability buffer for out-of-order publishing detection From 57beaf1816bb9f56433b683248578909c6f09e8d Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 23 Oct 2025 15:42:49 +0800 Subject: [PATCH 10/11] license --- LICENSE | 10 ++++++++++ .../dataregion/realtime/disruptor/Disruptor.java | 4 ---- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/LICENSE b/LICENSE index 409860cfcd15b..9b176e8eea60b 100644 --- a/LICENSE +++ b/LICENSE @@ -329,4 +329,14 @@ Apache Commons Collections is open source software licensed under the Apache Lic Project page: https://github.com/apache/commons-collections License: https://github.com/apache/commons-collections/blob/master/LICENSE.txt +-------------------------------------------------------------------------------- + +The following files include code modified from LMax Disruptor project. + +./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/* + +LMax Disruptor is open source software licensed under the Apache License 2.0 and supported by the Apache Software Foundation. +Project page: https://github.com/LMAX-Exchange/disruptor +License: https://github.com/LMAX-Exchange/disruptor/blob/master/LICENCE.txt + -------------------------------------------------------------------------------- \ No newline at end of file diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java index 5973544cdc948..90421be3e60be 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/Disruptor.java @@ -127,8 +127,4 @@ public void shutdown() { started = false; LOGGER.info("Disruptor shutdown completed"); } - - public RingBuffer getRingBuffer() { - return ringBuffer; - } } From d23b86adb4a4698ebf68b9816859ce059c0c627c Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 23 Oct 2025 15:46:54 +0800 Subject: [PATCH 11/11] fix --- .../disruptor/MultiProducerSequencer.java | 21 +-------- .../realtime/disruptor/RingBuffer.java | 23 ---------- .../realtime/disruptor/SequenceGroups.java | 44 ------------------- 3 files changed, 1 insertion(+), 87 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java index debfa1ab50b77..d40ed96839870 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/MultiProducerSequencer.java @@ -192,29 +192,10 @@ public long remainingCapacity() { * * @param gatingSequences consumer sequences to add */ - public final void addGatingSequences(Sequence... gatingSequences) { + public void addGatingSequences(Sequence... gatingSequences) { SequenceGroups.addSequences(this, this.cursor, gatingSequences); } - /** - * Remove a gating sequence - * - * @param sequence sequence to remove - * @return true if sequence was found and removed - */ - public boolean removeGatingSequence(Sequence sequence) { - return SequenceGroups.removeSequence(this, sequence); - } - - /** - * Get the minimum sequence from all consumers - * - * @return minimum gating sequence - */ - public long getMinimumSequence() { - return Sequence.getMinimumSequence(gatingSequences, cursor.get()); - } - /** * Create a sequence barrier for consumers * diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java index e43200adaffce..2af784b603d6c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/RingBuffer.java @@ -236,29 +236,6 @@ public void addGatingSequences(Sequence... gatingSequences) { sequencer.addGatingSequences(gatingSequences); } - /** - * Remove a gating sequence - * - *

          Should be called when a consumer is shut down - * - * @param sequence sequence to remove - * @return true if sequence was found and removed - */ - public boolean removeGatingSequence(Sequence sequence) { - return sequencer.removeGatingSequence(sequence); - } - - /** - * Get the minimum sequence from all consumers - * - *

          Represents the slowest consumer's progress - * - * @return minimum gating sequence - */ - public long getMinimumGatingSequence() { - return sequencer.getMinimumSequence(); - } - /** * Create a sequence barrier for consumers * diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java index 61571f5757d59..af5039070f0e4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/disruptor/SequenceGroups.java @@ -74,48 +74,4 @@ static void addSequences( sequence.set(cursorSequence); } } - - /** - * Remove sequence from the group - * - * @param sequencer the sequencer - * @param sequence sequence to remove - * @return true if removed - */ - static boolean removeSequence(final MultiProducerSequencer sequencer, final Sequence sequence) { - int numToRemove; - Sequence[] oldSequences; - Sequence[] newSequences; - - do { - oldSequences = sequencer.gatingSequences; - numToRemove = countMatching(oldSequences, sequence); - - if (0 == numToRemove) { - break; - } - - final int oldSize = oldSequences.length; - newSequences = new Sequence[oldSize - numToRemove]; - - for (int i = 0, pos = 0; i < oldSize; i++) { - final Sequence testSequence = oldSequences[i]; - if (sequence != testSequence) { - newSequences[pos++] = testSequence; - } - } - } while (!SEQUENCE_UPDATER.compareAndSet(sequencer, oldSequences, newSequences)); - - return numToRemove != 0; - } - - private static int countMatching(Sequence[] values, final Sequence toMatch) { - int numToRemove = 0; - for (Sequence value : values) { - if (value == toMatch) { - numToRemove++; - } - } - return numToRemove; - } }