diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java index d15d6e365a77f..11707fe5634db 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java @@ -97,7 +97,7 @@ public class IoTConsensus implements IConsensus { new ConcurrentHashMap<>(); private final IoTConsensusRPCService service; private final RegisterManager registerManager = new RegisterManager(); - private IoTConsensusConfig config; + private volatile IoTConsensusConfig config; private final IClientManager clientManager; private final IClientManager syncClientManager; private final ScheduledExecutorService backgroundTaskService; @@ -472,6 +472,11 @@ public String getRegionDirFromConsensusGroupId(ConsensusGroupId groupId) { public void reloadConsensusConfig(ConsensusConfig consensusConfig) { config = consensusConfig.getIotConsensusConfig(); + IoTConsensusMemoryManager.getInstance() + .init( + config.getReplication().getAllocateMemoryForConsensus(), + config.getReplication().getAllocateMemoryForQueue()); + for (IoTConsensusServerImpl impl : stateMachineMap.values()) { impl.reloadConsensusConfig(config); } diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java index 3002b018e3ea8..7033ddf36d960 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java @@ -116,7 +116,7 @@ public class IoTConsensusServerImpl { private final TreeSet configuration; private final AtomicLong searchIndex; private final LogDispatcher logDispatcher; - private IoTConsensusConfig config; + private volatile IoTConsensusConfig config; private final ConsensusReqReader consensusReqReader; private volatile boolean active; private String newSnapshotDirName; @@ -911,6 +911,7 @@ public String getConsensusGroupId() { /** This method is used for hot reload of IoTConsensusConfig. */ public void reloadConsensusConfig(IoTConsensusConfig config) { this.config = config; + logDispatcher.reloadConfig(config); } /** diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java index 22e5484f5a203..d8adec09a7b06 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java @@ -33,8 +33,8 @@ public class IoTConsensusMemoryManager { private final AtomicLong memorySizeInByte = new AtomicLong(0); private final AtomicLong queueMemorySizeInByte = new AtomicLong(0); private final AtomicLong syncMemorySizeInByte = new AtomicLong(0); - private Long maxMemorySizeInByte = Runtime.getRuntime().maxMemory() / 10; - private Long maxMemorySizeForQueueInByte = Runtime.getRuntime().maxMemory() / 100 * 6; + private volatile long maxMemorySizeInByte = Runtime.getRuntime().maxMemory() / 10; + private volatile long maxMemorySizeForQueueInByte = Runtime.getRuntime().maxMemory() / 100 * 6; private IoTConsensusMemoryManager() { MetricService.getInstance().addMetricSet(new IoTConsensusMemoryManagerMetrics(this)); diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java index 374691bf38bf1..3965290f82bd6 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java @@ -179,6 +179,10 @@ public void checkAndFlushIndex() { } } + public synchronized void reloadConfig(IoTConsensusConfig config) { + threads.forEach(thread -> thread.reloadConfig(config)); + } + public void offer(IndexedConsensusRequest request) { // we don't need to serialize and offer request when replicaNum is 1. if (!threads.isEmpty()) { @@ -215,7 +219,7 @@ public class LogDispatcherThread implements Runnable { private static final long PENDING_REQUEST_TAKING_TIME_OUT_IN_SEC = 10; private static final long START_INDEX = 1; - private final IoTConsensusConfig config; + private volatile IoTConsensusConfig config; private final Peer peer; private final IndexController controller; // A sliding window class that manages asynchronous pendingBatches @@ -273,6 +277,11 @@ public IoTConsensusConfig getConfig() { return config; } + private void reloadConfig(IoTConsensusConfig config) { + this.config = config; + syncStatus.reloadConfig(config); + } + public int getPendingEntriesSize() { return pendingEntries.size(); } @@ -358,11 +367,16 @@ public void run() { IndexedConsensusRequest request = pendingEntries.poll(PENDING_REQUEST_TAKING_TIME_OUT_IN_SEC, TimeUnit.SECONDS); 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 (pendingEntries.size() <= config.getReplication().getMaxLogEntriesNumPerBatch() - && bufferedEntries.isEmpty()) { - Thread.sleep(config.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs()); + if (shouldWaitForBatchAccumulation) { + waitForBatchAccumulation( + currentConfig.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs()); } } // Immediately check for interrupts after poll and sleep @@ -392,6 +406,10 @@ public void run() { logger.info("{}: Dispatcher for {} exits", impl.getThisNode(), peer); } + void waitForBatchAccumulation(long waitingTimeInMs) throws InterruptedException { + Thread.sleep(waitingTimeInMs); + } + 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 @@ -406,6 +424,7 @@ public void updateSafelyDeletedSearchIndex() { } public Batch getBatch() { + final IoTConsensusConfig currentConfig = config; long startIndex = syncStatus.getNextSendingIndex(); long maxIndex; synchronized (impl.getIndexObject()) { @@ -420,7 +439,7 @@ public Batch getBatch() { // Use drainTo instead of poll to reduce lock overhead pendingEntries.drainTo( bufferedEntries, - config.getReplication().getMaxLogEntriesNumPerBatch() - bufferedEntries.size()); + currentConfig.getReplication().getMaxLogEntriesNumPerBatch() - bufferedEntries.size()); } // remove all request that searchIndex < startIndex Iterator iterator = bufferedEntries.iterator(); @@ -434,7 +453,7 @@ public Batch getBatch() { } } - Batch batches = new Batch(config); + Batch batches = new Batch(currentConfig); // This condition will be executed in several scenarios: // 1. restart // 2. The getBatch() is invoked immediately at the moment the PendingEntries are consumed diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatus.java index accc9f7667d21..a96abfb1a5a2c 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/SyncStatus.java @@ -31,7 +31,7 @@ public class SyncStatus { private static final Logger LOGGER = LoggerFactory.getLogger(SyncStatus.class); - private final IoTConsensusConfig config; + private IoTConsensusConfig config; private final IndexController controller; private final LinkedList pendingBatches = new LinkedList<>(); private final IoTConsensusMemoryManager iotConsensusMemoryManager = @@ -42,6 +42,11 @@ 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. * diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManagerTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManagerTest.java index f87d8cd7f9887..6d6bae6165e1d 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManagerTest.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManagerTest.java @@ -37,6 +37,20 @@ public class IoTConsensusMemoryManagerTest { + @Test + public void testInitUpdatesMemoryLimits() { + IoTConsensusMemoryManager memoryManager = IoTConsensusMemoryManager.getInstance(); + long previousMaxMemory = memoryManager.getMaxMemorySizeInByte(); + long previousMaxQueueMemory = memoryManager.getMaxMemorySizeForQueueInByte(); + try { + memoryManager.init(1024, 512); + assertEquals(1024L, memoryManager.getMaxMemorySizeInByte().longValue()); + assertEquals(512L, memoryManager.getMaxMemorySizeForQueueInByte().longValue()); + } finally { + memoryManager.init(previousMaxMemory, previousMaxQueueMemory); + } + } + @Test public void testAllocateQueue() { IoTConsensusMemoryManager memoryManager = IoTConsensusMemoryManager.getInstance(); diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java new file mode 100644 index 0000000000000..de7b273bde2ab --- /dev/null +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java @@ -0,0 +1,222 @@ +/* + * 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.consensus.iot.logdispatcher; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.consensus.DataRegionId; +import org.apache.iotdb.consensus.common.Peer; +import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; +import org.apache.iotdb.consensus.config.IoTConsensusConfig; +import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl; +import org.apache.iotdb.consensus.iot.client.DispatchLogHandler; +import org.apache.iotdb.consensus.iot.thrift.TLogEntry; +import org.apache.iotdb.consensus.iot.util.TestEntry; +import org.apache.iotdb.consensus.iot.util.TestStateMachine; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.lang.reflect.Field; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.TreeSet; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +public class LogDispatcherTest { + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void testWaitForBatchAccumulationAfterFirstRequest() throws Exception { + final Peer localPeer = createPeer(1, 6667); + final Peer remotePeer = createPeer(2, 6668); + final IoTConsensusConfig config = IoTConsensusConfig.newBuilder().build(); + final ScheduledExecutorService backgroundTaskService = + Executors.newSingleThreadScheduledExecutor(); + final ExecutorService executorService = Executors.newSingleThreadExecutor(); + LogDispatcher.LogDispatcherThread dispatcherThread = null; + Future dispatcherFuture = null; + try { + final IoTConsensusServerImpl server = + createServer( + localPeer, Collections.singletonList(localPeer), config, backgroundTaskService); + final Batch batch = createBatch(config, 1); + final CountDownLatch accumulationWaitInvoked = new CountDownLatch(1); + final AtomicInteger getBatchInvocations = new AtomicInteger(); + dispatcherThread = + server.getLogDispatcher().new LogDispatcherThread(remotePeer, config, 0) { + @Override + public Batch getBatch() { + return getBatchInvocations.getAndIncrement() == 0 ? new Batch(config) : batch; + } + + @Override + void waitForBatchAccumulation(long waitingTimeInMs) { + accumulationWaitInvoked.countDown(); + } + + @Override + public void sendBatchAsync(Batch sentBatch, DispatchLogHandler handler) { + getSyncStatus().removeBatch(sentBatch); + Thread.currentThread().interrupt(); + } + }; + assertTrue( + dispatcherThread.offer( + new IndexedConsensusRequest( + 1, Collections.singletonList(new TestEntry(1, localPeer))))); + + dispatcherFuture = executorService.submit(dispatcherThread); + + assertTrue(accumulationWaitInvoked.await(5, TimeUnit.SECONDS)); + dispatcherFuture.get(5, TimeUnit.SECONDS); + } finally { + if (dispatcherFuture != null) { + dispatcherFuture.cancel(true); + } + executorService.shutdownNow(); + executorService.awaitTermination(5, TimeUnit.SECONDS); + if (dispatcherThread != null) { + dispatcherThread.stop(); + } + backgroundTaskService.shutdownNow(); + } + } + + @Test + public void testReloadConfigUpdatesExistingDispatcherPipeline() throws Exception { + final Peer localPeer = createPeer(1, 6677); + final Peer remotePeer = createPeer(2, 6678); + final IoTConsensusConfig initialConfig = + IoTConsensusConfig.newBuilder() + .setReplication( + IoTConsensusConfig.Replication.newBuilder() + .setMaxLogEntriesNumPerBatch(1) + .setMaxPendingBatchesNum(1) + .build()) + .build(); + final ScheduledExecutorService backgroundTaskService = + Executors.newSingleThreadScheduledExecutor(); + final ExecutorService executorService = Executors.newSingleThreadExecutor(); + LogDispatcher dispatcher = null; + Future secondBatchFuture = null; + try { + final IoTConsensusServerImpl server = + createServer( + localPeer, + Arrays.asList(localPeer, remotePeer), + initialConfig, + backgroundTaskService); + dispatcher = server.getLogDispatcher(); + final LogDispatcher.LogDispatcherThread dispatcherThread = getOnlyThread(dispatcher); + dispatcher.start(); + + final SyncStatus syncStatus = dispatcherThread.getSyncStatus(); + syncStatus.addNextBatch(createBatch(initialConfig, 1)); + final CountDownLatch secondBatchAttempted = new CountDownLatch(1); + secondBatchFuture = + executorService.submit( + () -> { + secondBatchAttempted.countDown(); + syncStatus.addNextBatch(createBatch(initialConfig, 2)); + return null; + }); + assertTrue(secondBatchAttempted.await(5, TimeUnit.SECONDS)); + Thread.sleep(100); + assertFalse(secondBatchFuture.isDone()); + + final IoTConsensusConfig reloadedConfig = + IoTConsensusConfig.newBuilder() + .setReplication( + IoTConsensusConfig.Replication.newBuilder() + .setMaxLogEntriesNumPerBatch(2) + .setMaxPendingBatchesNum(2) + .build()) + .build(); + server.reloadConsensusConfig(reloadedConfig); + + secondBatchFuture.get(5, TimeUnit.SECONDS); + assertSame(reloadedConfig, dispatcherThread.getConfig()); + assertEquals(2, syncStatus.getPendingBatches().size()); + } finally { + if (secondBatchFuture != null) { + secondBatchFuture.cancel(true); + } + executorService.shutdownNow(); + executorService.awaitTermination(5, TimeUnit.SECONDS); + if (dispatcher != null) { + dispatcher.stop(); + } + backgroundTaskService.shutdownNow(); + } + } + + private IoTConsensusServerImpl createServer( + Peer localPeer, + List configuration, + IoTConsensusConfig config, + ScheduledExecutorService backgroundTaskService) + throws Exception { + return new IoTConsensusServerImpl( + temporaryFolder.newFolder().getAbsolutePath(), + localPeer, + new TreeSet<>(configuration), + new TestStateMachine(), + backgroundTaskService, + null, + null, + config); + } + + private static Peer createPeer(int nodeId, int port) { + return new Peer(new DataRegionId(1), nodeId, new TEndPoint("127.0.0.1", port)); + } + + private static Batch createBatch(IoTConsensusConfig config, long searchIndex) { + final Batch batch = new Batch(config); + batch.addTLogEntry(new TLogEntry().setSearchIndex(searchIndex).setMemorySize(1)); + batch.buildIndex(); + return batch; + } + + @SuppressWarnings("unchecked") + private static LogDispatcher.LogDispatcherThread getOnlyThread(LogDispatcher dispatcher) + throws Exception { + final Field threadsField = LogDispatcher.class.getDeclaredField("threads"); + threadsField.setAccessible(true); + final List threads = + (List) threadsField.get(dispatcher); + assertEquals(1, threads.size()); + return threads.get(0); + } +}