Revert IoTConsensus batch accumulation changes

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java

@@ -108,7 +108,7 @@ public class IoTConsensus implements IConsensus {
   private final RegisterManager registerManager = new RegisterManager();
   private final UserDataTransferAuditHandler userDataTransferAuditHandler;
   private final UserDataTransferAuditClassifier userDataTransferAuditClassifier;
-  private volatile IoTConsensusConfig config;
+  private IoTConsensusConfig config;
 
   /**
    * Optional callback invoked after a new local peer is created via {@link #createLocalPeer}. Used
@@ -549,9 +549,6 @@ public String getRegionDirFromConsensusGroupId(ConsensusGroupId groupId) {
   public void reloadConsensusConfig(ConsensusConfig consensusConfig) {
     config = consensusConfig.getIotConsensusConfig();
 
-    IoTConsensusMemoryManager.getInstance()
-        .updateMaxMemoryRatioForQueue(config.getReplication().getMaxMemoryRatioForQueue());
-
     for (IoTConsensusServerImpl impl : stateMachineMap.values()) {
       impl.reloadConsensusConfig(config);
     }

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java

@@ -152,7 +152,7 @@ public class IoTConsensusServerImpl {
   private final Set<Peer> configuration = ConcurrentHashMap.newKeySet();
   private final AtomicLong searchIndex;
   private final LogDispatcher logDispatcher;
-  private volatile IoTConsensusConfig config;
+  private IoTConsensusConfig config;
   private final ConsensusReqReader consensusReqReader;
   private volatile boolean active;
   private String newSnapshotDirName;
@@ -1474,7 +1474,6 @@ public String getConsensusGroupId() {
   /** This method is used for hot reload of IoTConsensusConfig. */
   public void reloadConsensusConfig(IoTConsensusConfig config) {
     this.config = config;
-    logDispatcher.reloadConfig(config);
   }
 
   /**

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/Batch.java

@@ -69,10 +69,6 @@ public void addTLogEntry(TLogEntry entry, boolean containsUserData) {
   }
 
   public boolean canAccumulate() {
-    return canAccumulate(config, logEntries.size(), memorySize);
-  }
-
-  static boolean canAccumulate(IoTConsensusConfig config, int logEntriesSize, long memorySize) {
     // When reading entries from the WAL, the memory size is calculated based on the serialized
     // size, which can be significantly smaller than the actual size.
     // Thus, we add a multiplier to sender's memory size to estimate the receiver's memory cost.
@@ -81,7 +77,7 @@ static boolean canAccumulate(IoTConsensusConfig config, int logEntriesSize, long
     long senderMemSize = LogDispatcher.getSenderMemSizeSum().get();
     double multiplier = senderMemSize > 0 ? (double) receiverMemSize / senderMemSize : 1.0;
     multiplier = Math.max(multiplier, 1.0);
-    return logEntriesSize < config.getReplication().getMaxLogEntriesNumPerBatch()
+    return logEntries.size() < config.getReplication().getMaxLogEntriesNumPerBatch()
         && ((long) (memorySize * multiplier)) < config.getReplication().getMaxSizePerBatch();
   }
 

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java

@@ -37,7 +37,7 @@ public class IoTConsensusMemoryManager {
   private final AtomicLong syncMemorySizeInByte = new AtomicLong(0);
   private IMemoryBlock memoryBlock =
       new AtomicLongMemoryBlock("Consensus-Default", null, Runtime.getRuntime().maxMemory() / 10);
-  private volatile double maxMemoryRatioForQueue = 0.6;
+  private Double maxMemoryRatioForQueue = 0.6;
 
   private IoTConsensusMemoryManager() {
     MetricService.getInstance().addMetricSet(new IoTConsensusMemoryManagerMetrics(this));
@@ -158,10 +158,6 @@ public void init(IMemoryBlock memoryBlock, double maxMemoryRatioForQueue) {
     this.maxMemoryRatioForQueue = maxMemoryRatioForQueue;
   }
 
-  public void updateMaxMemoryRatioForQueue(double maxMemoryRatioForQueue) {
-    this.maxMemoryRatioForQueue = maxMemoryRatioForQueue;
-  }
-
   @TestOnly
   public void reset() {
     this.memoryBlock.release(this.memoryBlock.getUsedMemoryInBytes());

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java

@@ -183,10 +183,6 @@ public synchronized void checkAndFlushIndex() {
     impl.checkAndUpdateSafeDeletedSearchIndex();
   }
 
-  public synchronized void reloadConfig(IoTConsensusConfig config) {
-    threads.forEach(thread -> thread.reloadConfig(config));
-  }
-
   public void offer(IndexedConsensusRequest request) {
     offer(request, true);
   }
@@ -234,7 +230,7 @@ public class LogDispatcherThread implements Runnable {
 
     private static final long PENDING_REQUEST_TAKING_TIME_OUT_IN_MS = 10_000L;
     private static final long START_INDEX = 1;
-    private volatile IoTConsensusConfig config;
+    private final IoTConsensusConfig config;
     private final Peer peer;
     private final IndexController controller;
     // A sliding window class that manages asynchronous pendingBatches
@@ -293,11 +289,6 @@ public IoTConsensusConfig getConfig() {
       return config;
     }
 
-    private void reloadConfig(IoTConsensusConfig config) {
-      this.config = config;
-      syncStatus.reloadConfig(config);
-    }
-
     public int getPendingEntriesSize() {
       return pendingEntries.size();
     }
@@ -384,16 +375,11 @@ public void run() {
             IndexedConsensusRequest request =
                 pendingEntries.poll(calculateIdlePollTimeoutInMs(), TimeUnit.MILLISECONDS);
             if (request != null) {
-              final IoTConsensusConfig currentConfig = config;
-              final boolean shouldWaitForBatchAccumulation =
-                  pendingEntries.size()
-                          <= currentConfig.getReplication().getMaxLogEntriesNumPerBatch()
-                      && bufferedEntries.isEmpty();
               bufferedEntries.add(request);
               // If write pressure is low, we simply sleep a little to reduce the number of RPC
-              if (shouldWaitForBatchAccumulation) {
-                waitForBatchAccumulation(
-                    currentConfig.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs());
+              if (pendingEntries.size() <= config.getReplication().getMaxLogEntriesNumPerBatch()
+                  && bufferedEntries.isEmpty()) {
+                Thread.sleep(config.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs());
               }
             } else {
               maybeSendIdleWriterSafeTimeBarrier();
@@ -426,38 +412,6 @@ public void run() {
       logger.info(IoTConsensusMessages.DISPATCHER_EXITS, impl.getThisNode(), peer);
     }
 
-    void waitForBatchAccumulation(long waitingTimeInMs) throws InterruptedException {
-      if (waitingTimeInMs <= 0) {
-        return;
-      }
-
-      final long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(waitingTimeInMs);
-      final IoTConsensusConfig currentConfig = config;
-      int accumulatedEntries = bufferedEntries.size();
-      long accumulatedMemorySize =
-          bufferedEntries.stream().mapToLong(IndexedConsensusRequest::getMemorySize).sum();
-
-      // Keep collecting while the batch is below both its entry and memory limits. A plain sleep,
-      // or checking only the entry limit, makes the dispatcher wait for the full accumulation
-      // interval after a batch has already reached its memory limit. This unnecessarily throttles
-      // IoTConsensus when each request contains a large tablet.
-      while (Batch.canAccumulate(currentConfig, accumulatedEntries, accumulatedMemorySize)) {
-        final long remainingNanos = deadlineNanos - System.nanoTime();
-        if (remainingNanos <= 0) {
-          return;
-        }
-
-        final IndexedConsensusRequest request =
-            pendingEntries.poll(remainingNanos, TimeUnit.NANOSECONDS);
-        if (request == null) {
-          return;
-        }
-        bufferedEntries.add(request);
-        accumulatedEntries++;
-        accumulatedMemorySize += request.getMemorySize();
-      }
-    }
-
     public void updateSafelyDeletedSearchIndex() {
       // update safely deleted search index to delete outdated info,
       // indicating that insert nodes whose search index are before this value can be deleted
@@ -474,7 +428,6 @@ public void updateSafelyDeletedSearchIndex() {
 
     public Batch getBatch() {
 
-      final IoTConsensusConfig currentConfig = config;
       long startIndex = syncStatus.getNextSendingIndex();
       long maxIndex;
       synchronized (impl.getIndexObject()) {
@@ -489,7 +442,7 @@ public Batch getBatch() {
         // Use drainTo instead of poll to reduce lock overhead
         pendingEntries.drainTo(
             bufferedEntries,
-            currentConfig.getReplication().getMaxLogEntriesNumPerBatch() - bufferedEntries.size());
+            config.getReplication().getMaxLogEntriesNumPerBatch() - bufferedEntries.size());
       }
       // remove all request that searchIndex < startIndex
       Iterator<IndexedConsensusRequest> iterator = bufferedEntries.iterator();
@@ -503,7 +456,7 @@ public Batch getBatch() {
         }
       }
 
-      Batch batches = new Batch(currentConfig);
+      Batch batches = new Batch(config);
       // This condition will be executed in several scenarios:
       // 1. restart
       // 2. The getBatch() is invoked immediately at the moment the PendingEntries are consumed

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatus.java

@@ -32,7 +32,7 @@
 public class SyncStatus {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(SyncStatus.class);
-  private IoTConsensusConfig config;
+  private final IoTConsensusConfig config;
   private final IndexController controller;
   private final LinkedList<Batch> pendingBatches = new LinkedList<>();
   private final IoTConsensusMemoryManager iotConsensusMemoryManager =
@@ -43,11 +43,6 @@ public SyncStatus(IndexController controller, IoTConsensusConfig config) {
     this.config = config;
   }
 
-  public synchronized void reloadConfig(IoTConsensusConfig config) {
-    this.config = config;
-    notifyAll();
-  }
-
   /**
    * we may block here if the synchronization pipeline is full.
    *

iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManagerTest.java

@@ -41,14 +41,11 @@
 public class IoTConsensusMemoryManagerTest {
 
   private IMemoryBlock previousMemoryBlock;
-  private double previousMaxMemoryRatioForQueue;
   private long memoryBlockSize = 16 * 1024L;
 
   @Before
   public void setUp() throws Exception {
     previousMemoryBlock = IoTConsensusMemoryManager.getInstance().getMemoryBlock();
-    previousMaxMemoryRatioForQueue =
-        IoTConsensusMemoryManager.getInstance().getMaxMemoryRatioForQueue();
     IoTConsensusMemoryManager.getInstance()
         .setMemoryBlock(new AtomicLongMemoryBlock("Test", null, memoryBlockSize));
     IoTConsensusMemoryManager.getInstance().reset();
@@ -58,8 +55,6 @@ public void setUp() throws Exception {
   public void tearDown() throws Exception {
     IoTConsensusMemoryManager.getInstance().reset();
     IoTConsensusMemoryManager.getInstance().setMemoryBlock(previousMemoryBlock);
-    IoTConsensusMemoryManager.getInstance()
-        .updateMaxMemoryRatioForQueue(previousMaxMemoryRatioForQueue);
   }
 
   @Test
@@ -126,23 +121,6 @@ public void testClearUnserializedRequest() {
     assertEquals(0L, request.getRetainedMemorySize());
   }
 
-  @Test
-  public void testUpdateMaxMemoryRatioForQueue() {
-    final IndexedConsensusRequest request =
-        new IndexedConsensusRequest(
-            1,
-            Collections.singletonList(
-                new ByteBufferConsensusRequest(ByteBuffer.allocate((int) (memoryBlockSize / 3)))));
-    request.buildSerializedRequests();
-
-    IoTConsensusMemoryManager.getInstance().updateMaxMemoryRatioForQueue(0.25);
-    assertFalse(IoTConsensusMemoryManager.getInstance().reserve(request));
-
-    IoTConsensusMemoryManager.getInstance().updateMaxMemoryRatioForQueue(0.5);
-    assertTrue(IoTConsensusMemoryManager.getInstance().reserve(request));
-    IoTConsensusMemoryManager.getInstance().free(request);
-  }
-
   private void testReserveAndRelease(int numReservation) {
     int allocationSize = 1;
     long allocatedSize = 0;