This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 25ab7941ceb Limit Edge query threads and remove unused binary
allocator (#18587)
25ab7941ceb is described below
commit 25ab7941cebe6e14f396b5541cdd9daefae0f151
Author: Jackie Tien <[email protected]>
AuthorDate: Wed Sep 9 10:15:38 2026 +0800
Limit Edge query threads and remove unused binary allocator (#18587)
---
.../org/apache/iotdb/edge/it/IoTDBEdgeBasicIT.java | 124 ++++++++
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 30 ++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 26 +-
.../execution/schedule/DriverTaskThread.java | 35 ++-
.../iotdb/db/queryengine/plan/Coordinator.java | 5 +-
.../iotdb/db/conf/QueryThreadPoolConfigTest.java | 29 ++
.../execution/schedule/DriverTaskThreadTest.java | 120 ++++++++
.../resources/conf/edge/iotdb-system.properties | 2 +
.../conf/iotdb-system.properties.template | 49 +---
.../apache/iotdb/commons/i18n/CommonMessages.java | 14 +-
.../apache/iotdb/commons/i18n/CommonMessages.java | 15 +-
.../commons/binaryallocator/BinaryAllocator.java | 311 ---------------------
.../binaryallocator/BinaryAllocatorState.java | 71 -----
.../PooledBinaryPhantomReference.java | 42 ---
.../iotdb/commons/binaryallocator/arena/Arena.java | 273 ------------------
.../binaryallocator/arena/ArenaStrategy.java | 35 ---
.../binaryallocator/autoreleaser/Releaser.java | 93 ------
.../binaryallocator/config/AllocatorConfig.java | 53 ----
.../ema/AdaptiveWeightedAverage.java | 100 -------
.../commons/binaryallocator/evictor/Evictor.java | 102 -------
.../metric/BinaryAllocatorMetrics.java | 138 ---------
.../commons/binaryallocator/utils/SizeClasses.java | 146 ----------
.../iotdb/commons/concurrent/ThreadName.java | 2 -
.../apache/iotdb/commons/conf/CommonConfig.java | 50 ----
.../iotdb/commons/conf/CommonDescriptor.java | 25 --
.../service/metric/JvmGcMonitorMetrics.java | 4 -
.../iotdb/commons/service/metric/enums/Metric.java | 1 -
.../binaryallocator/BinaryAllocatorTest.java | 171 -----------
28 files changed, 369 insertions(+), 1697 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/edge/it/IoTDBEdgeBasicIT.java
b/integration-test/src/test/java/org/apache/iotdb/edge/it/IoTDBEdgeBasicIT.java
index 57a8189aa84..07ab860655d 100644
---
a/integration-test/src/test/java/org/apache/iotdb/edge/it/IoTDBEdgeBasicIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/edge/it/IoTDBEdgeBasicIT.java
@@ -45,6 +45,7 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
@@ -53,6 +54,7 @@ import java.sql.Statement;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashSet;
+import java.util.HexFormat;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -68,6 +70,7 @@ import java.util.stream.Stream;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -112,6 +115,11 @@ public class IoTDBEdgeBasicIT {
ports = EnvUtils.searchAvailablePorts();
rpcPort = ports[2];
configurePorts(edgeHome.resolve("conf/iotdb-system.properties"));
+ // Existing installations may still carry this retired configuration key.
+ Files.writeString(
+ edgeHome.resolve("conf/iotdb-system.properties"),
+ "\nenable_binary_allocator=true\n",
+ StandardOpenOption.APPEND);
runScript(edgeHome.resolve("sbin/start-edge.sh"), START_SCRIPT_LOG);
edgePid =
Long.parseLong(Files.readString(edgeHome.resolve("edge.pid")).trim());
@@ -246,6 +254,120 @@ public class IoTDBEdgeBasicIT {
jdbcUrl(), SessionConfig.DEFAULT_USER, SessionConfig.DEFAULT_PASSWORD);
}
+ @Test
+ public void testConcurrentCrossDatabaseJoinsWithSmallDispatchPool() throws
Exception {
+ try (Connection connection = openTableConnection();
+ Statement statement = connection.createStatement()) {
+ for (String database : new String[] {"edge_it_dispatch_a",
"edge_it_dispatch_b"}) {
+ statement.execute("CREATE DATABASE " + database);
+ statement.execute("USE " + database);
+ statement.execute("CREATE TABLE sensor(device STRING TAG, value INT32
FIELD)");
+ statement.execute("INSERT INTO sensor(time,device,value) VALUES
(1,'d1',42),(2,'d1',84)");
+ }
+ }
+ ExecutorService executor = Executors.newFixedThreadPool(4);
+ CountDownLatch ready = new CountDownLatch(4);
+ CountDownLatch start = new CountDownLatch(1);
+ List<Future<Void>> queries = new ArrayList<>();
+ try {
+ for (int i = 0; i < 4; i++) {
+ queries.add(
+ executor.submit(
+ () -> {
+ try (Connection connection = openTableConnection();
+ Statement statement = connection.createStatement()) {
+ ready.countDown();
+ assertTrue(start.await(30, TimeUnit.SECONDS));
+ for (int iteration = 0; iteration < 10; iteration++) {
+ try (ResultSet result =
+ statement.executeQuery(
+ "SELECT count(*), sum(a.value + b.value)"
+ + " FROM edge_it_dispatch_a.sensor a JOIN
edge_it_dispatch_b.sensor b"
+ + " ON a.device = b.device AND a.time =
b.time")) {
+ assertTrue(result.next());
+ assertEquals(2, result.getLong(1));
+ assertEquals(252.0, result.getDouble(2), 0.0);
+ assertFalse(result.next());
+ }
+ }
+ }
+ return null;
+ }));
+ }
+ assertTrue(ready.await(30, TimeUnit.SECONDS));
+ start.countDown();
+ for (Future<Void> query : queries) {
+ query.get(90, TimeUnit.SECONDS);
+ }
+ } finally {
+ start.countDown();
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
+ }
+ String threads = captureThreadDump("cross-database-joins");
+ long dispatchWorkers =
+ threads
+ .lines()
+ .filter(
+ line -> line.startsWith("\"pool-") &&
line.contains("Fragment-Instance-Dispatch-"))
+ .count();
+ assertTrue("The join must exercise fragment dispatch", dispatchWorkers >
0);
+ assertTrue("Fragment dispatch exceeded its configured pool size",
dispatchWorkers <= 2);
+ }
+
+ @Test
+ public void testBlobReadWriteAfterLegacyAllocatorConfigurationReload()
throws Exception {
+ byte[] expected = new byte[65536];
+ for (int i = 0; i < expected.length; i++) {
+ expected[i] = (byte) i;
+ }
+ try (Connection connection = openTableConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("LOAD CONFIGURATION");
+ statement.execute("CREATE DATABASE edge_it_blob");
+ statement.execute("USE edge_it_blob");
+ statement.execute("CREATE TABLE payloads(device STRING TAG, payload BLOB
FIELD)");
+ statement.execute(
+ "INSERT INTO payloads(time,device,payload) VALUES (1,'d1',X'"
+ + HexFormat.of().formatHex(expected)
+ + "')");
+ for (int round = 0; round < 2; round++) {
+ try (ResultSet result = statement.executeQuery("SELECT payload FROM
payloads")) {
+ assertTrue(result.next());
+ assertArrayEquals(expected, result.getBytes(1));
+ assertFalse(result.next());
+ }
+ if (round == 0) {
+ statement.execute("FLUSH");
+ statement.execute("LOAD CONFIGURATION");
+ }
+ }
+ }
+
assertFalse(captureThreadDump("blob-after-reload").contains("BinaryAllocator-"));
+ }
+
+ private static String captureThreadDump(String name) throws Exception {
+ Path output = WORK_DIR.resolve(name + "-threads.txt");
+ Process process =
+ new ProcessBuilder(
+ Paths.get(System.getProperty("java.home"), "bin",
"jcmd").toString(),
+ Long.toString(edgePid),
+ "Thread.print",
+ "-l")
+ .redirectErrorStream(true)
+ .redirectOutput(output.toFile())
+ .start();
+ try {
+ assertTrue("Thread dump timed out", process.waitFor(30,
TimeUnit.SECONDS));
+ assertEquals("Failed to capture thread dump: " + output, 0,
process.exitValue());
+ return Files.readString(output);
+ } finally {
+ if (process.isAlive()) {
+ process.destroyForcibly();
+ }
+ }
+ }
+
@Test
public void testRatisMetadataConsensus() throws SQLException {
Map<String, String> variables = new LinkedHashMap<>();
@@ -280,6 +402,8 @@ public class IoTDBEdgeBasicIT {
assertEdgeProperty("coordinator_read_executor_size", "2");
assertEdgeProperty("coordinator_scheduled_executor_size", "2");
assertEdgeProperty("fragment_instance_notification_thread_count", "2");
+ assertEdgeProperty("driver_task_scheduler_notification_thread_count", "2");
+ assertEdgeProperty("fragment_instance_dispatch_thread_count", "2");
assertEdgeProperty("cn_load_statistics_publisher_thread_count", "1");
assertEdgeProperty("candidate_compaction_task_queue_size", "10");
assertEdgeProperty("compaction_max_aligned_series_num_in_one_batch", "2");
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index b14f2e1b622..989031fc0fd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -1026,6 +1026,12 @@ public class IoTDBConfig {
/** Thread pool size for fragment instance state change notifications. */
private int fragmentInstanceNotificationThreadCount = 4;
+ /** Zero retains the cached pool used by general deployments. */
+ private int driverTaskSchedulerNotificationThreadCount = 0;
+
+ /** Zero selects max(20, twice the available processors). */
+ private int fragmentInstanceDispatchThreadCount = 0;
+
/** Policy of DataNodeSchemaCache eviction */
private String dataNodeSchemaCacheEvictionPolicy = "FIFO";
@@ -3612,6 +3618,30 @@ public class IoTDBConfig {
this.fragmentInstanceNotificationThreadCount =
fragmentInstanceNotificationThreadCount;
}
+ public int getDriverTaskSchedulerNotificationThreadCount() {
+ return driverTaskSchedulerNotificationThreadCount;
+ }
+
+ public void setDriverTaskSchedulerNotificationThreadCount(int threadCount) {
+ if (threadCount < 0) {
+ throw new IllegalArgumentException(
+
CommonMessages.EXCEPTION_THREAD_COUNT_MUST_BE_GREATER_THAN_OR_EQUAL_TO_0_988EF69B);
+ }
+ this.driverTaskSchedulerNotificationThreadCount = threadCount;
+ }
+
+ public int getFragmentInstanceDispatchThreadCount() {
+ return fragmentInstanceDispatchThreadCount;
+ }
+
+ public void setFragmentInstanceDispatchThreadCount(int threadCount) {
+ if (threadCount < 0) {
+ throw new IllegalArgumentException(
+
CommonMessages.EXCEPTION_THREAD_COUNT_MUST_BE_GREATER_THAN_OR_EQUAL_TO_0_988EF69B);
+ }
+ this.fragmentInstanceDispatchThreadCount = threadCount;
+ }
+
public TEndPoint getAddressAndPort() {
return new TEndPoint(rpcAddress, rpcPort);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 44971c718a1..9f48bac0e0c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.conf;
import org.apache.iotdb.calc.exception.QueryProcessException;
-import org.apache.iotdb.commons.binaryallocator.BinaryAllocator;
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.conf.ConfigurationFileUtils;
@@ -1012,6 +1011,16 @@ public class IoTDBDescriptor {
properties.getProperty(
"fragment_instance_notification_thread_count",
Integer.toString(conf.getFragmentInstanceNotificationThreadCount()))));
+ conf.setDriverTaskSchedulerNotificationThreadCount(
+ Integer.parseInt(
+ properties.getProperty(
+ "driver_task_scheduler_notification_thread_count",
+
Integer.toString(conf.getDriverTaskSchedulerNotificationThreadCount()))));
+ conf.setFragmentInstanceDispatchThreadCount(
+ Integer.parseInt(
+ properties.getProperty(
+ "fragment_instance_dispatch_thread_count",
+
Integer.toString(conf.getFragmentInstanceDispatchThreadCount()))));
conf.setDataNodeTableSchemaCacheSize(
Long.parseLong(
properties.getProperty(
@@ -2308,21 +2317,6 @@ public class IoTDBDescriptor {
// update retry config
commonDescriptor.loadRetryProperties(properties);
- // update binary allocator
- commonDescriptor
- .getConfig()
- .setEnableBinaryAllocator(
- Boolean.parseBoolean(
- properties.getProperty(
- "enable_binary_allocator",
- ConfigurationFileUtils.getConfigurationDefaultValue(
- "enable_binary_allocator"))));
- if (commonDescriptor.getConfig().isEnableBinaryAllocator()) {
- BinaryAllocator.getInstance().start();
- } else {
- BinaryAllocator.getInstance().close(true);
- }
-
// update disk_space_warning_threshold; also refresh the static copy in
JVMCommonUtils that
// the ReadOnly disk guard reads, otherwise the new threshold would not
take effect until
// restart. Parsing / validation is shared with the ConfigNode
hot-reload path.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskThread.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskThread.java
index 912bd6711e3..1c44cb82f0c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskThread.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskThread.java
@@ -23,6 +23,7 @@ import
org.apache.iotdb.calc.execution.schedule.queue.IndexedBlockingQueue;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.execution.driver.IDriver;
import
org.apache.iotdb.db.queryengine.execution.schedule.queue.multilevelqueue.MultilevelPriorityQueue;
import org.apache.iotdb.db.queryengine.execution.schedule.task.DriverTask;
@@ -33,6 +34,7 @@ import com.google.common.util.concurrent.ListenableFuture;
import io.airlift.units.Duration;
import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
@@ -54,10 +56,13 @@ public class DriverTaskThread extends AbstractDriverThread {
(level + 1) * DRIVER_TASK_EXECUTION_TIME_SLICE_IN_MS,
TimeUnit.MILLISECONDS))
.toArray(Duration[]::new);
- // We manage thread pool size directly, so create an unlimited pool
- private static final Executor listeningExecutor =
- IoTDBThreadPoolFactory.newCachedThreadPool(
- ThreadName.DRIVER_TASK_SCHEDULER_NOTIFICATION.getName());
+ private static final Executor NOTIFICATION_EXECUTOR =
+ createNotificationExecutor(
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getDriverTaskSchedulerNotificationThreadCount());
+
+ private final Executor listeningExecutor;
private final Ticker ticker;
@@ -67,10 +72,32 @@ public class DriverTaskThread extends AbstractDriverThread {
IndexedBlockingQueue<DriverTask> queue,
ITaskScheduler scheduler,
ThreadProducer producer) {
+ this(workerId, tg, queue, scheduler, producer, NOTIFICATION_EXECUTOR);
+ }
+
+ DriverTaskThread(
+ String workerId,
+ ThreadGroup tg,
+ IndexedBlockingQueue<DriverTask> queue,
+ ITaskScheduler scheduler,
+ ThreadProducer producer,
+ Executor listeningExecutor) {
super(workerId, tg, queue, scheduler, producer);
+ this.listeningExecutor = listeningExecutor;
this.ticker = Ticker.systemTicker();
}
+ static ExecutorService createNotificationExecutor(int threadCount) {
+ String poolName = ThreadName.DRIVER_TASK_SCHEDULER_NOTIFICATION.getName();
+ if (threadCount == 0) {
+ return IoTDBThreadPoolFactory.newCachedThreadPool(poolName);
+ }
+ // Queue notifications instead of running them on threads completing
driver futures, which
+ // may hold locks. Idle workers can exit after the query workload ends.
+ return IoTDBThreadPoolFactory.newFixedThreadPoolWithIdleThreadTimeout(
+ threadCount, 60, TimeUnit.SECONDS, poolName);
+ }
+
@Override
public void execute(DriverTask task) throws InterruptedException {
long startNanos = ticker.read();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
index 09a35803595..fad2ea1354d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
@@ -284,7 +284,10 @@ public class Coordinator {
this.typeManager = new InternalTypeManager();
this.executor = getQueryExecutor();
this.scheduledExecutor = getScheduledExecutor();
- int dispatchThreadNum = Math.max(20,
Runtime.getRuntime().availableProcessors() * 2);
+ int dispatchThreadNum = CONFIG.getFragmentInstanceDispatchThreadCount();
+ if (dispatchThreadNum == 0) {
+ dispatchThreadNum = Math.max(20,
Runtime.getRuntime().availableProcessors() * 2);
+ }
this.dispatchExecutor =
IoTDBThreadPoolFactory.newCachedThreadPool(
ThreadName.FRAGMENT_INSTANCE_DISPATCH.getName(),
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/QueryThreadPoolConfigTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/QueryThreadPoolConfigTest.java
index 92b6d0fd45f..8ac89c9f6f2 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/QueryThreadPoolConfigTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/QueryThreadPoolConfigTest.java
@@ -32,6 +32,15 @@ public class QueryThreadPoolConfigTest {
public void testDefaultsAndPositiveSizes() throws Exception {
IoTDBConfig config = new IoTDBConfig();
assertSizes(config, 20, 10, 4);
+ assertAdditionalSizes(config, 0, 0);
+ assertEquals(
+ "0",
+ ConfigurationFileUtils.getConfigurationDefaultValue(
+ "driver_task_scheduler_notification_thread_count"));
+ assertEquals(
+ "0",
+ ConfigurationFileUtils.getConfigurationDefaultValue(
+ "fragment_instance_dispatch_thread_count"));
assertEquals(
"20",
ConfigurationFileUtils.getConfigurationDefaultValue("coordinator_read_executor_size"));
@@ -54,6 +63,15 @@ public class QueryThreadPoolConfigTest {
() -> config.setFragmentInstanceNotificationThreadCount(invalid));
}
assertSizes(config, 20, 10, 4);
+ for (int invalid : new int[] {-1, Integer.MIN_VALUE}) {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> config.setDriverTaskSchedulerNotificationThreadCount(invalid));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> config.setFragmentInstanceDispatchThreadCount(invalid));
+ }
+ assertAdditionalSizes(config, 0, 0);
}
@Test
@@ -63,14 +81,25 @@ public class QueryThreadPoolConfigTest {
properties.setProperty("coordinator_read_executor_size", "3");
properties.setProperty("coordinator_scheduled_executor_size", "2");
properties.setProperty("fragment_instance_notification_thread_count", "1");
+ properties.setProperty("driver_task_scheduler_notification_thread_count",
"1");
+ properties.setProperty("fragment_instance_dispatch_thread_count", "2");
descriptor.loadProperties(properties);
assertSizes(descriptor.getConfig(), 3, 2, 1);
+ assertAdditionalSizes(descriptor.getConfig(), 1, 2);
properties.setProperty("coordinator_read_executor_size", "6");
properties.setProperty("coordinator_scheduled_executor_size", "5");
properties.setProperty("fragment_instance_notification_thread_count", "4");
+ properties.setProperty("driver_task_scheduler_notification_thread_count",
"4");
+ properties.setProperty("fragment_instance_dispatch_thread_count", "5");
descriptor.loadHotModifiedProps(properties);
assertSizes(descriptor.getConfig(), 3, 2, 1);
+ assertAdditionalSizes(descriptor.getConfig(), 1, 2);
+ }
+
+ private static void assertAdditionalSizes(IoTDBConfig config, int
notification, int dispatch) {
+ assertEquals(notification,
config.getDriverTaskSchedulerNotificationThreadCount());
+ assertEquals(dispatch, config.getFragmentInstanceDispatchThreadCount());
}
private static void assertSizes(IoTDBConfig config, int read, int scheduled,
int notification) {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskThreadTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskThreadTest.java
new file mode 100644
index 00000000000..5ea8b9af35c
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskThreadTest.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.queryengine.execution.schedule;
+
+import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
+import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.execution.driver.IDriver;
+import org.apache.iotdb.db.queryengine.execution.schedule.task.DriverTask;
+import org.apache.iotdb.db.queryengine.execution.schedule.task.DriverTaskId;
+import
org.apache.iotdb.db.queryengine.execution.schedule.task.DriverTaskStatus;
+
+import com.google.common.util.concurrent.SettableFuture;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class DriverTaskThreadTest {
+
+ @Test
+ public void
testBlockedDriverNotificationsQueueWithoutRunningOnCompletingThread()
+ throws Exception {
+ int taskCount = 12;
+ int threadCount = 2;
+ ExecutorService executor =
DriverTaskThread.createNotificationExecutor(threadCount);
+ CountDownLatch workersStarted = new CountDownLatch(threadCount);
+ CountDownLatch releaseWorkers = new CountDownLatch(1);
+ CountDownLatch notified = new CountDownLatch(taskCount);
+ Set<Thread> callbackThreads = ConcurrentHashMap.newKeySet();
+ ITaskScheduler scheduler = mock(ITaskScheduler.class);
+ when(scheduler.readyToRunning(any())).thenReturn(true);
+ doAnswer(
+ invocation -> {
+ callbackThreads.add(Thread.currentThread());
+ workersStarted.countDown();
+ assertTrue(releaseWorkers.await(10, TimeUnit.SECONDS));
+ notified.countDown();
+ return null;
+ })
+ .when(scheduler)
+ .blockedToReady(any());
+
+ DriverTaskThread worker =
+ new DriverTaskThread("test-worker", null, null, scheduler, null,
executor);
+ List<SettableFuture<Void>> futures = new ArrayList<>();
+ List<DriverTask> tasks = new ArrayList<>();
+ try {
+ for (int i = 0; i < taskCount; i++) {
+ IDriver driver = mock(IDriver.class);
+ DriverTaskId id =
+ new DriverTaskId(
+ new FragmentInstanceId(
+ new PlanFragmentId(new QueryId("notifications"), 0), "i" +
i),
+ 0);
+ when(driver.getDriverTaskId()).thenReturn(id);
+ SettableFuture<Void> future = SettableFuture.create();
+ doReturn(future).when(driver).processFor(any());
+ DriverTask task = new DriverTask(driver, 30000,
DriverTaskStatus.READY, null, 0, false);
+ tasks.add(task);
+ futures.add(future);
+ worker.execute(task);
+ }
+ // Completing futures must remain nonblocking even when all notification
workers are busy.
+ for (SettableFuture<Void> future : futures) {
+ future.set(null);
+ }
+ assertTrue(workersStarted.await(10, TimeUnit.SECONDS));
+ assertEquals(threadCount, callbackThreads.size());
+ assertEquals(taskCount - threadCount, ((ThreadPoolExecutor)
executor).getQueue().size());
+ assertFalse(callbackThreads.contains(Thread.currentThread()));
+
+ releaseWorkers.countDown();
+ assertTrue(notified.await(10, TimeUnit.SECONDS));
+ assertEquals(threadCount, ((ThreadPoolExecutor)
executor).getLargestPoolSize());
+ for (DriverTask task : tasks) {
+ verify(scheduler, times(1)).runningToBlocked(eq(task), any());
+ verify(scheduler, times(1)).blockedToReady(task);
+ }
+ } finally {
+ releaseWorkers.countDown();
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
+ }
+ }
+}
diff --git
a/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties
b/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties
index 47a67d1cf38..82679f99e6a 100644
---
a/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties
+++
b/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties
@@ -82,6 +82,8 @@ dn_metric_prometheus_reporter_port=9092
coordinator_read_executor_size=2
coordinator_scheduled_executor_size=2
fragment_instance_notification_thread_count=2
+driver_task_scheduler_notification_thread_count=2
+fragment_instance_dispatch_thread_count=2
cn_load_statistics_publisher_thread_count=1
query_thread_count=2
degree_of_query_parallelism=1
diff --git
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
index 866dde0a94e..c53c95fd016 100644
---
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
+++
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
@@ -1250,6 +1250,20 @@ coordinator_scheduled_executor_size=10
# Datatype: int
fragment_instance_notification_thread_count=4
+# Maximum number of workers notifying the driver scheduler that blocked tasks
are ready.
+# 0 retains the unbounded cached pool. Positive values limit workers and queue
notifications.
+# Idle workers exit after 60 seconds. Must be non-negative.
+# effectiveMode: restart
+# Datatype: int
+driver_task_scheduler_notification_thread_count=0
+
+# Maximum number of workers dispatching query fragment instances.
+# 0 uses max(20, CPU core number * 2). Must be non-negative.
+# When all workers are busy, dispatch runs on the submitting thread.
+# effectiveMode: restart
+# Datatype: int
+fragment_instance_dispatch_thread_count=0
+
# How many pipeline drivers will be created for one fragment instance. When <=
0, use CPU core number / 2.
# effectiveMode: restart
# Datatype: int
@@ -1847,41 +1861,6 @@
data_region_iot_snapshot_transmission_progress_log_interval_ms = 5000
# Datatype: boolean
keep_same_disk_when_loading_snapshot=true
-####################
-### Blob Allocator Configuration
-####################
-
-# Whether to enable binary allocator.
-# For scenarios where large binary streams cause severe GC, enabling this
parameter significantly improves performance.
-# effectiveMode: hot_reload
-enable_binary_allocator=true
-
-# The size boundaries that allocator is responsible for
-# lower boundary for allocation size
-# unit: bytes
-# Datatype: int
-# effectiveMode: restart
-small_binary_object=4096
-
-# The size boundaries that allocator is responsible for
-# upper boundary for allocation size
-# unit: bytes
-# Datatype: int
-# effectiveMode: restart
-huge_binary_object=1048576
-
-# Number of arena regions in blob allocator, used to control concurrent
performance
-# Datatype: int
-# effectiveMode: restart
-arena_num=4
-
-# Control the number of slabs in allocator
-# The number of different sizes in each power-of-2 interval is
2^LOG2_SIZE_CLASS_GROUP
-# For example: if LOG2_SIZE_CLASS_GROUP=3, between 1024-2048 there will be 8
different sizes
-# Datatype: int
-# effectiveMode: restart
-log2_size_class_group=3
-
####################
### TsFile Configurations
####################
diff --git
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
index a6b958fd142..a05c8f9915a 100644
---
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
@@ -107,6 +107,8 @@ public final class CommonMessages {
public static final String BASE_VALUE_SHOULD_NOT_BE_NULL =
"When comparing, base value should never be null";
public static final String SIZE_MUST_BE_POSITIVE = "Size must be greater
than 0";
+ public static final String
EXCEPTION_THREAD_COUNT_MUST_BE_GREATER_THAN_OR_EQUAL_TO_0_988EF69B =
+ "Thread count must be greater than or equal to 0";
// --- sync ---
public static final String UNEXPECTED_SERIALIZATION_ERROR =
@@ -117,18 +119,6 @@ public final class CommonMessages {
public static final String CLASSLOADER_NOT_DETERMINED =
"A ClassLoader to load the class could not be determined.";
- // --- binaryallocator ---
- public static final String BINARY_ALLOCATOR_RUNNING_GC_EVICTION =
- "Binary allocator running GC eviction";
- public static final String BINARY_ALLOCATOR_SHUTTING_DOWN_HIGH_GC =
- "Binary allocator is shutting down because of high GC time percentage
{}%.";
- public static final String AUTO_RELEASER_EXIT_INTERRUPTED =
- "{} exits due to interruptedException.";
- public static final String STOPPING_COMPONENT = "Stopping {}";
- public static final String UNABLE_TO_STOP_AUTO_RELEASER =
- "unable to stop auto releaser after {} ms";
- public static final String UNABLE_TO_STOP_EVICTOR = "unable to stop evictor
after {} ms";
-
// --- startcheck / system.properties ---
public static final String IOTDB_VERSION_TOO_OLD = "IoTDB version is too
old";
public static final String REPAIR_SYSTEM_PROPERTIES = "repair
system.properties, lack {}";
diff --git
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
index c3af2dbe7e2..c5c6290687e 100644
---
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
@@ -106,6 +106,9 @@ public final class CommonMessages {
"比较时基准值不应为 null";
public static final String SIZE_MUST_BE_POSITIVE = "Size 必须大于 0";
+ public static final String
EXCEPTION_THREAD_COUNT_MUST_BE_GREATER_THAN_OR_EQUAL_TO_0_988EF69B =
+ "线程数必须大于或等于 0";
+
// --- sync ---
public static final String UNEXPECTED_SERIALIZATION_ERROR =
"序列化 PipeInfo 时发生意外错误。";
@@ -114,18 +117,6 @@ public final class CommonMessages {
public static final String ENCRYPT_PASSWORD_ERROR = "加密密码时出错。";
public static final String CLASSLOADER_NOT_DETERMINED = "无法确定用于加载类的
ClassLoader。";
- // --- binaryallocator ---
- public static final String BINARY_ALLOCATOR_RUNNING_GC_EVICTION =
- "二进制分配器正在执行 GC 驱逐";
- public static final String BINARY_ALLOCATOR_SHUTTING_DOWN_HIGH_GC =
- "由于 GC 时间百分比过高 ({}%)，二进制分配器正在关闭。";
- public static final String AUTO_RELEASER_EXIT_INTERRUPTED =
- "{} 因 InterruptedException 退出。";
- public static final String STOPPING_COMPONENT = "正在停止 {}";
- public static final String UNABLE_TO_STOP_AUTO_RELEASER =
- "在 {} 毫秒后仍无法停止自动释放器";
- public static final String UNABLE_TO_STOP_EVICTOR = "在 {} 毫秒后仍无法停止驱逐器";
-
// --- startcheck / system.properties ---
public static final String IOTDB_VERSION_TOO_OLD = "IoTDB 版本过旧";
public static final String REPAIR_SYSTEM_PROPERTIES = "修复
system.properties，缺少 {}";
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocator.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocator.java
deleted file mode 100644
index 175651551d6..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocator.java
+++ /dev/null
@@ -1,311 +0,0 @@
-/*
- * 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.commons.binaryallocator;
-
-import org.apache.iotdb.commons.binaryallocator.arena.Arena;
-import org.apache.iotdb.commons.binaryallocator.arena.ArenaStrategy;
-import org.apache.iotdb.commons.binaryallocator.autoreleaser.Releaser;
-import org.apache.iotdb.commons.binaryallocator.config.AllocatorConfig;
-import org.apache.iotdb.commons.binaryallocator.evictor.Evictor;
-import org.apache.iotdb.commons.binaryallocator.metric.BinaryAllocatorMetrics;
-import org.apache.iotdb.commons.binaryallocator.utils.SizeClasses;
-import org.apache.iotdb.commons.concurrent.ThreadName;
-import org.apache.iotdb.commons.i18n.CommonMessages;
-import org.apache.iotdb.commons.service.metric.MetricService;
-import org.apache.iotdb.commons.utils.TestOnly;
-
-import org.apache.tsfile.utils.PooledBinary;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.lang.ref.ReferenceQueue;
-import java.time.Duration;
-import java.util.Collections;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicReference;
-
-public class BinaryAllocator {
-
- private static final Logger LOGGER =
LoggerFactory.getLogger(BinaryAllocator.class);
-
- private final Arena[] heapArenas;
- private final AllocatorConfig allocatorConfig;
-
- private final ArenaStrategy arenaStrategy = new LeastUsedArenaStrategy();
- private final AtomicReference<BinaryAllocatorState> state =
- new AtomicReference<>(BinaryAllocatorState.UNINITIALIZED);
-
- private final BinaryAllocatorMetrics metrics;
- private Evictor sampleEvictor;
- private Releaser autoReleaser;
- private static final ThreadLocal<ThreadArenaRegistry> arenaRegistry =
- ThreadLocal.withInitial(ThreadArenaRegistry::new);
-
- private static final int WARNING_GC_TIME_PERCENTAGE = 20;
- private static final int HALF_GC_TIME_PERCENTAGE = 25;
- private static final int SHUTDOWN_GC_TIME_PERCENTAGE = 30;
- private static final int RESTART_GC_TIME_PERCENTAGE = 5;
-
- public final ReferenceQueue<PooledBinary> referenceQueue = new
ReferenceQueue<>();
-
- // JDK 9+ Cleaner uses double-linked list and synchronized to manage
references, which has worse
- // performance than lock-free hash set
- public final Set<PooledBinaryPhantomReference> phantomRefs =
- Collections.newSetFromMap(new ConcurrentHashMap<>());
-
- public BinaryAllocator(AllocatorConfig allocatorConfig) {
- this.allocatorConfig = allocatorConfig;
-
- heapArenas = new Arena[allocatorConfig.arenaNum];
- SizeClasses sizeClasses = new SizeClasses(allocatorConfig);
-
- for (int i = 0; i < heapArenas.length; i++) {
- Arena arena = new Arena(this, sizeClasses, i, allocatorConfig);
- heapArenas[i] = arena;
- }
-
- this.metrics = new BinaryAllocatorMetrics(this);
-
- if (allocatorConfig.enableBinaryAllocator) {
- start();
- } else {
- state.set(BinaryAllocatorState.CLOSE);
- }
- }
-
- public synchronized void start() {
- if (state.get() == BinaryAllocatorState.OPEN) {
- return;
- }
-
- state.set(BinaryAllocatorState.OPEN);
- MetricService.getInstance().addMetricSet(this.metrics);
- sampleEvictor =
- new SampleEvictor(
- ThreadName.BINARY_ALLOCATOR_SAMPLE_EVICTOR.getName(),
- allocatorConfig.durationShutdownTimeout,
- allocatorConfig.durationBetweenEvictorRuns);
- sampleEvictor.start();
- autoReleaser =
- new AutoReleaser(
- ThreadName.BINARY_ALLOCATOR_AUTO_RELEASER.getName(),
- allocatorConfig.durationShutdownTimeout);
- autoReleaser.start();
- }
-
- public synchronized void close(boolean forceClose) {
- if (forceClose) {
- state.set(BinaryAllocatorState.CLOSE);
- MetricService.getInstance().removeMetricSet(this.metrics);
- } else {
- state.set(BinaryAllocatorState.PENDING);
- }
-
- sampleEvictor.stop();
- autoReleaser.stop();
- for (Arena arena : heapArenas) {
- arena.close();
- }
- }
-
- public PooledBinary allocateBinary(int reqCapacity, boolean autoRelease) {
- if (reqCapacity < allocatorConfig.minAllocateSize
- || reqCapacity > allocatorConfig.maxAllocateSize
- || state.get() != BinaryAllocatorState.OPEN) {
- return new PooledBinary(new byte[reqCapacity]);
- }
-
- Arena arena = arenaStrategy.choose(heapArenas);
-
- return arena.allocate(reqCapacity, autoRelease);
- }
-
- public void deallocateBinary(PooledBinary binary) {
- if (binary != null
- && binary.getLength() >= allocatorConfig.minAllocateSize
- && binary.getLength() <= allocatorConfig.maxAllocateSize
- && state.get() == BinaryAllocatorState.OPEN) {
- int arenaIndex = binary.getArenaIndex();
- if (arenaIndex != -1) {
- Arena arena = heapArenas[arenaIndex];
- arena.deallocate(binary);
- }
- }
- }
-
- public long getTotalUsedMemory() {
- long totalUsedMemory = 0;
- for (Arena arena : heapArenas) {
- totalUsedMemory += arena.getTotalUsedMemory();
- }
- return totalUsedMemory;
- }
-
- public long getTotalActiveMemory() {
- long totalActiveMemory = 0;
- for (Arena arena : heapArenas) {
- totalActiveMemory += arena.getActiveMemory();
- }
- return totalActiveMemory;
- }
-
- @TestOnly
- public void resetArenaBinding() {
- arenaRegistry.get().unbindArena();
- }
-
- public BinaryAllocatorMetrics getMetrics() {
- return metrics;
- }
-
- private long evict(double ratio) {
- long evictedSize = 0;
- for (Arena arena : heapArenas) {
- evictedSize += arena.evict(ratio);
- }
- return evictedSize;
- }
-
- public static BinaryAllocator getInstance() {
- return BinaryAllocatorHolder.INSTANCE;
- }
-
- private static class BinaryAllocatorHolder {
-
- private static final BinaryAllocator INSTANCE =
- new BinaryAllocator(AllocatorConfig.DEFAULT_CONFIG);
- }
-
- private static class ThreadArenaRegistry {
-
- private Arena threadArenaBinding = null;
-
- public Arena getArena() {
- return threadArenaBinding;
- }
-
- public void bindArena(Arena arena) {
- threadArenaBinding = arena;
- arena.incRegisteredThread();
- }
-
- public void unbindArena() {
- Arena arena = threadArenaBinding;
- if (arena != null) {
- arena.decRegisteredThread();
- threadArenaBinding = null;
- }
- }
-
- @Override
- protected void finalize() {
- unbindArena();
- }
- }
-
- private static class LeastUsedArenaStrategy implements ArenaStrategy {
-
- @Override
- public Arena choose(Arena[] arenas) {
- Arena boundArena = arenaRegistry.get().getArena();
- if (boundArena != null) {
- return boundArena;
- }
-
- Arena minArena = arenas[0];
-
- for (int i = 1; i < arenas.length; i++) {
- Arena arena = arenas[i];
- if (arena.getNumRegisteredThread() <
minArena.getNumRegisteredThread()) {
- minArena = arena;
- }
- }
-
- arenaRegistry.get().bindArena(minArena);
- return minArena;
- }
- }
-
- public void runGcEviction(long curGcTimePercent) {
- if (state.get() == BinaryAllocatorState.CLOSE) {
- return;
- }
-
- LOGGER.debug(CommonMessages.BINARY_ALLOCATOR_RUNNING_GC_EVICTION);
- if (state.get() == BinaryAllocatorState.PENDING) {
- if (curGcTimePercent <= RESTART_GC_TIME_PERCENTAGE) {
- start();
- }
- return;
- }
-
- long evictedSize = 0;
- if (curGcTimePercent > SHUTDOWN_GC_TIME_PERCENTAGE) {
- LOGGER.info(CommonMessages.BINARY_ALLOCATOR_SHUTTING_DOWN_HIGH_GC,
curGcTimePercent);
- evictedSize = evict(1.0);
- close(false);
- } else if (curGcTimePercent > HALF_GC_TIME_PERCENTAGE) {
- evictedSize = evict(0.5);
- } else if (curGcTimePercent > WARNING_GC_TIME_PERCENTAGE) {
- evictedSize = evict(0.2);
- }
- metrics.updateGcEvictionCounter(evictedSize);
- }
-
- public class SampleEvictor extends Evictor {
-
- public SampleEvictor(
- String name, Duration evictorShutdownTimeoutDuration, Duration
durationBetweenEvictorRuns) {
- super(name, evictorShutdownTimeoutDuration, durationBetweenEvictorRuns);
- }
-
- @Override
- public void run() {
- long evictedSize = 0;
- for (Arena arena : heapArenas) {
- evictedSize += arena.runSampleEviction();
- }
- metrics.updateSampleEvictionCounter(evictedSize);
- }
- }
-
- /** Process phantomly reachable objects and return their byte arrays to
pool. */
- public class AutoReleaser extends Releaser {
-
- public AutoReleaser(String name, Duration shutdownTimeoutDuration) {
- super(name, shutdownTimeoutDuration);
- }
-
- @Override
- public void run() {
- PooledBinaryPhantomReference ref;
- try {
- while ((ref = (PooledBinaryPhantomReference) referenceQueue.remove())
!= null) {
- phantomRefs.remove(ref);
- ref.slabRegion.deallocate(ref.byteArray);
- }
- } catch (InterruptedException e) {
- LOGGER.info(CommonMessages.AUTO_RELEASER_EXIT_INTERRUPTED, name);
- Thread.currentThread().interrupt();
- }
- }
- }
-}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocatorState.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocatorState.java
deleted file mode 100644
index 6812bc84b11..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocatorState.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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.commons.binaryallocator;
-
-/**
- * The state transmission of a binary allocator.
- *
- * <pre>
- * ----------------------------------------
- * | |
- * | ---------- |
- * | | | |
- * | v | v
- * UNINITIALIZED --> OPEN ---> PENDING --> CLOSE
- * ^ ^
- * | |
- * -------------------------
- * </pre>
- *
- * State Transition Logic:
- *
- * <ul>
- * <li><b>UNINITIALIZED -> CLOSE</b>: When enable_binary_allocator = false
- * <li><b>UNINITIALIZED -> OPEN</b>: When enable_binary_allocator = true
- * <li><b>OPEN -> CLOSE</b>: When enable_binary_allocator is hot reload to
false
- * <li><b>CLOSE -> OPEN</b>: When enable_binary_allocator is hot reload to
true
- * <li><b>PENDING -> CLOSE</b>: When enable_binary_allocator is hot reload
to false
- * <li><b>OPEN -> PENDING</b>: When in OPEN state and GC time percentage
exceeds 30%, indicating
- * allocator ineffectiveness
- * <li><b>PENDING -> OPEN</b>: When GC time percentage drops below 5%,
returning to normal state
- * and re-enabling the allocator. Or when enable_binary_allocator is hot
reload to true.
- * </ul>
- */
-public enum BinaryAllocatorState {
- /** Binary allocator is open for allocation. */
- OPEN,
-
- /** Binary allocator is close. All allocations are from the JVM heap. */
- CLOSE,
-
- /**
- * Binary allocator is temporarily closed by GC evictor. All allocations are
from the JVM heap.
- * Allocator can be restarted afterward.
- */
- PENDING,
-
- /** The initial state of the allocator. */
- UNINITIALIZED;
-
- @Override
- public String toString() {
- return name();
- }
-}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/PooledBinaryPhantomReference.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/PooledBinaryPhantomReference.java
deleted file mode 100644
index 84541219c08..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/PooledBinaryPhantomReference.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.commons.binaryallocator;
-
-import org.apache.iotdb.commons.binaryallocator.arena.Arena;
-
-import org.apache.tsfile.utils.PooledBinary;
-
-import java.lang.ref.PhantomReference;
-import java.lang.ref.ReferenceQueue;
-
-public class PooledBinaryPhantomReference extends
PhantomReference<PooledBinary> {
- public final byte[] byteArray;
- public Arena.SlabRegion slabRegion;
-
- public PooledBinaryPhantomReference(
- PooledBinary referent,
- ReferenceQueue<? super PooledBinary> q,
- byte[] byteArray,
- Arena.SlabRegion region) {
- super(referent, q);
- this.byteArray = byteArray;
- this.slabRegion = region;
- }
-}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/arena/Arena.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/arena/Arena.java
deleted file mode 100644
index b5d95234aa7..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/arena/Arena.java
+++ /dev/null
@@ -1,273 +0,0 @@
-/*
- * 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.commons.binaryallocator.arena;
-
-import org.apache.iotdb.commons.binaryallocator.BinaryAllocator;
-import org.apache.iotdb.commons.binaryallocator.PooledBinaryPhantomReference;
-import org.apache.iotdb.commons.binaryallocator.config.AllocatorConfig;
-import org.apache.iotdb.commons.binaryallocator.ema.AdaptiveWeightedAverage;
-import org.apache.iotdb.commons.binaryallocator.utils.SizeClasses;
-
-import org.apache.tsfile.utils.PooledBinary;
-
-import java.lang.ref.ReferenceQueue;
-import java.util.Set;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.atomic.AtomicInteger;
-
-public class Arena {
-
- private static final int EVICT_SAMPLE_COUNT = 100;
-
- private final BinaryAllocator binaryAllocator;
- private final SizeClasses sizeClasses;
- private final int arenaID;
- private final AtomicInteger numRegisteredThread;
- private final SlabRegion[] regions;
-
- private int sampleCount;
-
- private final ReferenceQueue<PooledBinary> referenceQueue;
- private final Set<PooledBinaryPhantomReference> phantomRefs;
-
- public Arena(
- BinaryAllocator allocator, SizeClasses sizeClasses, int id,
AllocatorConfig allocatorConfig) {
- this.binaryAllocator = allocator;
- this.sizeClasses = sizeClasses;
- this.arenaID = id;
- this.numRegisteredThread = new AtomicInteger(0);
- regions = new SlabRegion[sizeClasses.getSizeClassNum()];
-
- for (int i = 0; i < regions.length; i++) {
- regions[i] = new SlabRegion(sizeClasses.sizeIdx2size(i),
allocatorConfig);
- }
-
- sampleCount = 0;
- referenceQueue = binaryAllocator.referenceQueue;
- phantomRefs = binaryAllocator.phantomRefs;
- }
-
- public int getArenaID() {
- return arenaID;
- }
-
- public PooledBinary allocate(int reqCapacity, boolean autoRelease) {
- final int sizeIdx = sizeClasses.size2SizeIdx(reqCapacity);
- byte[] data = regions[sizeIdx].allocate();
- if (autoRelease) {
- PooledBinary binary = new PooledBinary(data, reqCapacity, -1);
- PooledBinaryPhantomReference ref =
- new PooledBinaryPhantomReference(binary, referenceQueue, data,
regions[sizeIdx]);
- phantomRefs.add(ref);
- return binary;
- } else {
- return new PooledBinary(data, reqCapacity, arenaID);
- }
- }
-
- public void deallocate(PooledBinary binary) {
- final int sizeIdx = sizeClasses.size2SizeIdx(binary.getLength());
- regions[sizeIdx].deallocate(binary.getValues());
- }
-
- public long evict(double ratio) {
- long evictedSize = 0;
- for (SlabRegion region : regions) {
- evictedSize += region.evict(ratio);
- }
- return evictedSize;
- }
-
- public void close() {
- sampleCount = 0;
- for (SlabRegion region : regions) {
- region.close();
- }
- }
-
- public long getTotalUsedMemory() {
- long totalUsedMemory = 0;
- for (SlabRegion region : regions) {
- totalUsedMemory += region.getTotalUsedMemory();
- }
- return totalUsedMemory;
- }
-
- public long getActiveMemory() {
- long totalActiveMemory = 0;
- for (SlabRegion region : regions) {
- totalActiveMemory += region.getActiveUsedMemory();
- }
- return totalActiveMemory;
- }
-
- public int getNumRegisteredThread() {
- return numRegisteredThread.get();
- }
-
- public void incRegisteredThread() {
- this.numRegisteredThread.incrementAndGet();
- }
-
- public void decRegisteredThread() {
- this.numRegisteredThread.decrementAndGet();
- }
-
- public long runSampleEviction() {
- // update metric
- long allocateFromSlabDelta = 0;
- long allocateFromJVMDelta = 0;
- for (SlabRegion region : regions) {
- allocateFromSlabDelta +=
- (long) region.byteArraySize
- * (region.allocationsFromAllocator.get() -
region.prevAllocations);
- region.prevAllocations = region.allocationsFromAllocator.get();
- allocateFromJVMDelta +=
- (long) region.byteArraySize
- * (region.allocationsFromJVM.get() -
region.prevAllocationsFromJVM);
- region.prevAllocationsFromJVM = region.allocationsFromJVM.get();
- }
- binaryAllocator
- .getMetrics()
- .updateAllocationCounter(allocateFromSlabDelta, allocateFromJVMDelta);
-
- // Start sampling
- for (SlabRegion region : regions) {
- region.updateSample();
- }
-
- sampleCount++;
- long evictedSize = 0;
- if (sampleCount == EVICT_SAMPLE_COUNT) {
- // Evict
- for (SlabRegion region : regions) {
- evictedSize += region.resize();
- }
- sampleCount = 0;
- }
- return evictedSize;
- }
-
- public static class SlabRegion {
- private final int byteArraySize;
-
- // Current implementation uses ConcurrentLinkedQueue for simplicity
- // TODO: Can be optimized with more efficient lock-free approaches:
- // 1. No need for strict FIFO, it's just an object pool
- // 2. Use segmented arrays/queues with per-segment counters to reduce
contention
- private final ConcurrentLinkedQueue<byte[]> queue;
-
- private final AtomicInteger allocationsFromAllocator;
- private final AtomicInteger allocationsFromJVM;
- private final AtomicInteger deAllocationsToAllocator;
- private final AtomicInteger evictions;
-
- public int prevAllocations;
- public int prevAllocationsFromJVM;
- AdaptiveWeightedAverage average;
-
- SlabRegion(int byteArraySize, AllocatorConfig allocatorConfig) {
- this.byteArraySize = byteArraySize;
- this.average = new
AdaptiveWeightedAverage(allocatorConfig.arenaPredictionWeight);
- queue = new ConcurrentLinkedQueue<>();
- allocationsFromAllocator = new AtomicInteger(0);
- allocationsFromJVM = new AtomicInteger(0);
- deAllocationsToAllocator = new AtomicInteger(0);
- evictions = new AtomicInteger(0);
- prevAllocations = 0;
- prevAllocationsFromJVM = 0;
- }
-
- public final byte[] allocate() {
- byte[] bytes = queue.poll();
- if (bytes == null) {
- allocationsFromJVM.incrementAndGet();
- return new byte[this.byteArraySize];
- }
- allocationsFromAllocator.incrementAndGet();
- return bytes;
- }
-
- public void deallocate(byte[] bytes) {
- deAllocationsToAllocator.incrementAndGet();
- queue.add(bytes);
- }
-
- private void updateSample() {
- average.sample(getActiveSize());
- }
-
- private long resize() {
- average.update();
- float averageActiveSize = average.average();
- if (averageActiveSize < 0.0001f) {
- // avoid keeping the last binary forever
- averageActiveSize = 0.0f;
- }
- int needRemain = (int) Math.ceil(averageActiveSize) - getActiveSize();
- return evict(getQueueSize() - needRemain);
- }
-
- private long evict(double ratio) {
- return evict((int) (getQueueSize() * ratio));
- }
-
- private long evict(int num) {
- long evicted = 0;
- while (num > 0 && !queue.isEmpty()) {
- queue.poll();
- evictions.incrementAndGet();
- num--;
- evicted += byteArraySize;
- }
- return evicted;
- }
-
- private long getTotalUsedMemory() {
- return (long) byteArraySize * getQueueSize();
- }
-
- private long getActiveUsedMemory() {
- return (long) byteArraySize * getActiveSize();
- }
-
- // ConcurrentLinkedQueue::size() is O(n)
- private int getQueueSize() {
- return deAllocationsToAllocator.get() - allocationsFromAllocator.get() -
evictions.get();
- }
-
- private int getActiveSize() {
- return allocationsFromAllocator.get()
- + allocationsFromJVM.get()
- - deAllocationsToAllocator.get();
- }
-
- private void close() {
- queue.clear();
- allocationsFromAllocator.set(0);
- allocationsFromJVM.set(0);
- deAllocationsToAllocator.set(0);
- evictions.set(0);
- prevAllocations = 0;
- prevAllocationsFromJVM = 0;
- average.clear();
- }
- }
-}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/arena/ArenaStrategy.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/arena/ArenaStrategy.java
deleted file mode 100644
index ec4af3fbe0a..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/arena/ArenaStrategy.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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.commons.binaryallocator.arena;
-
-/**
- * This interface defines a strategy for choosing a {@link Arena} from an
array of {@link Arena}s.
- * Implementations of this interface can provide various strategies for
selection based on specific
- * criteria.
- */
-public interface ArenaStrategy {
- /**
- * Chooses a {@link Arena} from the given array of {@link Arena}s.
- *
- * @param arenas an array of {@link Arena}s to choose from, should not be
null or length == 0
- * @return the selected {@link Arena}
- */
- Arena choose(Arena[] arenas);
-}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/autoreleaser/Releaser.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/autoreleaser/Releaser.java
deleted file mode 100644
index 26b9607ad9f..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/autoreleaser/Releaser.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * 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.commons.binaryallocator.autoreleaser;
-
-import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
-import org.apache.iotdb.commons.i18n.CommonMessages;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.time.Duration;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-
-public abstract class Releaser implements Runnable {
- private static final Logger LOGGER = LoggerFactory.getLogger(Releaser.class);
-
- private Future<?> future;
- protected final String name;
- private final Duration shutdownTimeoutDuration;
-
- private ExecutorService executor;
-
- public Releaser(String name, Duration shutdownTimeoutDuration) {
- this.name = name;
- this.shutdownTimeoutDuration = shutdownTimeoutDuration;
- }
-
- /** Cancels the future. */
- void cancel() {
- future.cancel(false);
- }
-
- @Override
- public abstract void run();
-
- void setFuture(final Future<?> future) {
- this.future = future;
- }
-
- @Override
- public String toString() {
- return getClass().getName() + " [future=" + future + "]";
- }
-
- public void start() {
- if (null == executor) {
- executor = IoTDBThreadPoolFactory.newSingleThreadExecutor(name);
- }
- final Future<?> future = executor.submit(this);
- this.setFuture(future);
- }
-
- public void stop() {
- if (executor == null) {
- return;
- }
-
- LOGGER.info(CommonMessages.STOPPING_COMPONENT, name);
-
- cancel();
- executor.shutdown();
- try {
- boolean result =
- executor.awaitTermination(shutdownTimeoutDuration.toMillis(),
TimeUnit.MILLISECONDS);
- if (!result) {
- LOGGER.info(
- CommonMessages.UNABLE_TO_STOP_AUTO_RELEASER,
shutdownTimeoutDuration.toMillis());
- }
- } catch (final InterruptedException ignored) {
- Thread.currentThread().interrupt();
- }
- executor = null;
- }
-}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/config/AllocatorConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/config/AllocatorConfig.java
deleted file mode 100644
index 3bbbbc9a210..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/config/AllocatorConfig.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.commons.binaryallocator.config;
-
-import org.apache.iotdb.commons.conf.CommonDescriptor;
-
-import java.time.Duration;
-
-public class AllocatorConfig {
-
- public int minAllocateSize =
CommonDescriptor.getInstance().getConfig().getMinAllocateSize();
-
- public int maxAllocateSize =
CommonDescriptor.getInstance().getConfig().getMaxAllocateSize();
-
- public int arenaNum =
CommonDescriptor.getInstance().getConfig().getArenaNum();
-
- public int log2ClassSizeGroup =
- CommonDescriptor.getInstance().getConfig().getLog2SizeClassGroup();
-
- public boolean enableBinaryAllocator =
- CommonDescriptor.getInstance().getConfig().isEnableBinaryAllocator();
-
- /** Maximum wait time in milliseconds when shutting down the evictor and
autoReleaser */
- public Duration durationShutdownTimeout = Duration.ofMillis(1000L);
-
- /** Time interval in milliseconds between two consecutive evictor runs */
- public Duration durationBetweenEvictorRuns = Duration.ofMillis(1000L);
-
- public int arenaPredictionWeight = 35;
-
- public static final AllocatorConfig DEFAULT_CONFIG = new AllocatorConfig();
-
- public void setTimeBetweenEvictorRunsMillis(long time) {
- this.durationBetweenEvictorRuns = Duration.ofMillis(time);
- }
-}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/ema/AdaptiveWeightedAverage.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/ema/AdaptiveWeightedAverage.java
deleted file mode 100644
index 201f2f63ae1..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/ema/AdaptiveWeightedAverage.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * 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.commons.binaryallocator.ema;
-
-import static java.lang.Math.max;
-
-/**
- * This file is modified from <a
- *
href="https://github.com/openjdk/jdk17/blob/master/src/hotspot/share/gc/shared/gcUtil.hpp";>JDK17
- * AdaptiveWeightedAverage</a>. But some necessary modifications are made to
adapt to the usage of
- * binary allocator:
- *
- * <p>Adaptive weighted average implementation for memory allocation tracking.
During each eviction
- * cycle, records the peak memory allocation size via sampling, then uses this
peak to calculate a
- * weighted moving average.
- */
-public class AdaptiveWeightedAverage {
-
- private static final int OLD_THRESHOLD = 100;
-
- private float average;
- private int sampleCount;
- private int tmpMaxSample;
- private final int weight;
- private boolean isOld; // Enable to have enough historical data
-
- public AdaptiveWeightedAverage(int weight) {
- this.weight = weight;
- average = 0f;
- sampleCount = 0;
- tmpMaxSample = 0;
- }
-
- public void sample(int newSample) {
- tmpMaxSample = max(tmpMaxSample, newSample);
- }
-
- // called at the end of each eviction cycle
- public void update() {
- incrementCount();
-
- // Compute the new weighted average
- int newSample = tmpMaxSample;
- tmpMaxSample = 0;
- average = computeAdaptiveAverage(newSample, average);
- }
-
- public float average() {
- return average;
- }
-
- public void clear() {
- average = 0f;
- sampleCount = 0;
- tmpMaxSample = 0;
- isOld = false;
- }
-
- void incrementCount() {
- sampleCount++;
-
- if (!isOld && sampleCount > OLD_THRESHOLD) {
- isOld = true;
- }
- }
-
- float computeAdaptiveAverage(int newSample, float average) {
- // We smooth the samples by not using weight() directly until we've
- // had enough data to make it meaningful. We'd like the first weight
- // used to be 1, the second to be 1/2, etc until we have
- // OLD_THRESHOLD/weight samples.
- int countWeight = 0;
-
- // Avoid division by zero if the counter wraps
- if (!isOld) {
- countWeight = OLD_THRESHOLD / sampleCount;
- }
-
- int adaptiveWeight = max(weight, countWeight);
-
- return (100.0f - adaptiveWeight) * average / 100.0f + adaptiveWeight *
newSample / 100.0f;
- }
-}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/evictor/Evictor.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/evictor/Evictor.java
deleted file mode 100644
index d310ef1f15e..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/evictor/Evictor.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/*
- * 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.commons.binaryallocator.evictor;
-
-import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
-import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
-import org.apache.iotdb.commons.i18n.CommonMessages;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.time.Duration;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-
-public abstract class Evictor implements Runnable {
- private static final Logger LOGGER = LoggerFactory.getLogger(Evictor.class);
-
- private ScheduledFuture<?> scheduledFuture;
- private final String name;
- private final Duration shutdownTimeoutDuration;
- private final Duration durationBetweenEvictorRuns;
-
- private ScheduledExecutorService executor;
-
- public Evictor(
- String name, Duration shutdownTimeoutDuration, Duration
durationBetweenEvictorRuns) {
- this.name = name;
- this.shutdownTimeoutDuration = shutdownTimeoutDuration;
- this.durationBetweenEvictorRuns = durationBetweenEvictorRuns;
- }
-
- /** Cancels the scheduled future. */
- void cancel() {
- scheduledFuture.cancel(false);
- }
-
- @Override
- public abstract void run();
-
- void setScheduledFuture(final ScheduledFuture<?> scheduledFuture) {
- this.scheduledFuture = scheduledFuture;
- }
-
- @Override
- public String toString() {
- return getClass().getName() + " [scheduledFuture=" + scheduledFuture + "]";
- }
-
- public void start() {
- if (null == executor) {
- executor = IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(name);
- }
- final ScheduledFuture<?> scheduledFuture =
- ScheduledExecutorUtil.safelyScheduleAtFixedRate(
- executor,
- this,
- durationBetweenEvictorRuns.toMillis(),
- durationBetweenEvictorRuns.toMillis(),
- TimeUnit.MILLISECONDS);
- this.setScheduledFuture(scheduledFuture);
- }
-
- public void stop() {
- if (executor == null) {
- return;
- }
-
- LOGGER.info(CommonMessages.STOPPING_COMPONENT, name);
-
- cancel();
- executor.shutdown();
- try {
- boolean result =
- executor.awaitTermination(shutdownTimeoutDuration.toMillis(),
TimeUnit.MILLISECONDS);
- if (!result) {
- LOGGER.info(CommonMessages.UNABLE_TO_STOP_EVICTOR,
shutdownTimeoutDuration.toMillis());
- }
- } catch (final InterruptedException ignored) {
- Thread.currentThread().interrupt();
- }
- executor = null;
- }
-}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/metric/BinaryAllocatorMetrics.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/metric/BinaryAllocatorMetrics.java
deleted file mode 100644
index 305dadc432b..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/metric/BinaryAllocatorMetrics.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * 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.commons.binaryallocator.metric;
-
-import org.apache.iotdb.commons.binaryallocator.BinaryAllocator;
-import org.apache.iotdb.commons.service.metric.enums.Metric;
-import org.apache.iotdb.commons.service.metric.enums.Tag;
-import org.apache.iotdb.metrics.AbstractMetricService;
-import org.apache.iotdb.metrics.metricsets.IMetricSet;
-import org.apache.iotdb.metrics.type.Counter;
-import org.apache.iotdb.metrics.utils.MetricLevel;
-import org.apache.iotdb.metrics.utils.MetricType;
-
-public class BinaryAllocatorMetrics implements IMetricSet {
-
- private static final String TOTAL_MEMORY = "total-memory";
- private static final String ALLOCATE_FROM_SLAB = "allocate-from-slab";
- private static final String ALLOCATE_FROM_JVM = "allocate-from-jvm";
- private static final String ACTIVE_MEMORY = "active-memory";
- private static final String EVICTED_BY_SAMPLE_EVICTION =
"evicted-by-sample-eviction";
- private static final String EVICTED_BY_GC_EVICTION =
"evicted-by-gc-eviction";
-
- private final BinaryAllocator binaryAllocator;
- private Counter allocateFromSlab;
- private Counter allocateFromJVM;
- private Counter evictedBySampleEviction;
- private Counter evictedByGcEviction;
-
- public BinaryAllocatorMetrics(final BinaryAllocator binaryAllocator) {
- this.binaryAllocator = binaryAllocator;
- }
-
- @Override
- public void bindTo(AbstractMetricService metricService) {
- metricService.createAutoGauge(
- Metric.BINARY_ALLOCATOR.toString(),
- MetricLevel.IMPORTANT,
- binaryAllocator,
- BinaryAllocator::getTotalUsedMemory,
- Tag.NAME.toString(),
- TOTAL_MEMORY);
- metricService.createAutoGauge(
- Metric.BINARY_ALLOCATOR.toString(),
- MetricLevel.IMPORTANT,
- binaryAllocator,
- BinaryAllocator::getTotalActiveMemory,
- Tag.NAME.toString(),
- ACTIVE_MEMORY);
- allocateFromSlab =
- metricService.getOrCreateCounter(
- Metric.BINARY_ALLOCATOR.toString(),
- MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- ALLOCATE_FROM_SLAB);
- allocateFromJVM =
- metricService.getOrCreateCounter(
- Metric.BINARY_ALLOCATOR.toString(),
- MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- ALLOCATE_FROM_JVM);
- evictedBySampleEviction =
- metricService.getOrCreateCounter(
- Metric.BINARY_ALLOCATOR.toString(),
- MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- EVICTED_BY_SAMPLE_EVICTION);
- evictedByGcEviction =
- metricService.getOrCreateCounter(
- Metric.BINARY_ALLOCATOR.toString(),
- MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- EVICTED_BY_GC_EVICTION);
- }
-
- @Override
- public void unbindFrom(AbstractMetricService metricService) {
- metricService.remove(
- MetricType.AUTO_GAUGE,
- Metric.BINARY_ALLOCATOR.toString(),
- Tag.NAME.toString(),
- TOTAL_MEMORY);
- metricService.remove(
- MetricType.AUTO_GAUGE,
- Metric.BINARY_ALLOCATOR.toString(),
- Tag.NAME.toString(),
- ACTIVE_MEMORY);
- metricService.remove(
- MetricType.COUNTER,
- Metric.BINARY_ALLOCATOR.toString(),
- Tag.NAME.toString(),
- ALLOCATE_FROM_SLAB);
- metricService.remove(
- MetricType.COUNTER,
- Metric.BINARY_ALLOCATOR.toString(),
- Tag.NAME.toString(),
- ALLOCATE_FROM_JVM);
- metricService.remove(
- MetricType.COUNTER,
- Metric.BINARY_ALLOCATOR.toString(),
- Tag.NAME.toString(),
- EVICTED_BY_SAMPLE_EVICTION);
- metricService.remove(
- MetricType.COUNTER,
- Metric.BINARY_ALLOCATOR.toString(),
- Tag.NAME.toString(),
- EVICTED_BY_GC_EVICTION);
- }
-
- public void updateAllocationCounter(long allocateFromSlabDelta, long
allocateFromJVMDelta) {
- allocateFromSlab.inc(allocateFromSlabDelta);
- allocateFromJVM.inc(allocateFromJVMDelta);
- }
-
- public void updateGcEvictionCounter(long evictedByGcEvictionDelta) {
- evictedByGcEviction.inc(evictedByGcEvictionDelta);
- }
-
- public void updateSampleEvictionCounter(long evictedBySampleEvictionDelta) {
- evictedBySampleEviction.inc(evictedBySampleEvictionDelta);
- }
-}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/utils/SizeClasses.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/utils/SizeClasses.java
deleted file mode 100644
index e13cc54a9f1..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/utils/SizeClasses.java
+++ /dev/null
@@ -1,146 +0,0 @@
-/*
- * 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.commons.binaryallocator.utils;
-
-import org.apache.iotdb.commons.binaryallocator.config.AllocatorConfig;
-
-/**
- * SizeClasses class manages different size classes of memory blocks in a
memory allocator. It
- * optimizes the memory allocation process by precomputing the block sizes and
mapping them to
- * indices.
- */
-public final class SizeClasses {
-
- // Integer size in bits minus 1, used for log2 calculations
- private static final int INTEGER_SIZE_MINUS_ONE = Integer.SIZE - 1;
-
- // Mapping from size class index to actual memory block size
- private final int[] sizeIdx2sizeTab;
-
- // Log2 value of the minimum memory block size
- private final int log2MinSize;
-
- // Log2 value of the size class group
- private final int log2SizeClassGroup;
-
- /**
- * Constructor that initializes the size class table based on the allocator
configuration.
- *
- * @param allocatorConfig The allocator configuration containing minimum and
maximum allocation
- * sizes.
- */
- public SizeClasses(AllocatorConfig allocatorConfig) {
- this.log2SizeClassGroup = allocatorConfig.log2ClassSizeGroup;
- this.log2MinSize = log2(allocatorConfig.minAllocateSize);
-
- int maxSize = allocatorConfig.maxAllocateSize;
- int sizeClassGroupCount = log2(maxSize) - log2MinSize;
-
- // Initialize the sizeIdx2sizeTab array based on the number of size class
groups
- sizeIdx2sizeTab = new int[(sizeClassGroupCount << log2SizeClassGroup) + 1];
-
- // Calculate the size of each size class and populate the table
- initializeSizeClasses(allocatorConfig.minAllocateSize, maxSize);
- }
-
- /**
- * Returns the memory block size for a given size class index.
- *
- * @param sizeIdx The index of the size class.
- * @return The memory block size corresponding to the size class index.
- */
- public int sizeIdx2size(int sizeIdx) {
- return sizeIdx2sizeTab[sizeIdx];
- }
-
- /**
- * Returns the size class index for a given memory block size.
- *
- * @param size The memory block size.
- * @return The corresponding size class index.
- */
- public int size2SizeIdx(int size) {
- int log2Size = log2((size << 1) - 1); // Calculate the approximate log2
value
- int shift = log2Size - log2MinSize - 1;
-
- // Calculate the size class group
- int group = shift << log2SizeClassGroup;
- int log2Delta = log2Size - 1 - log2SizeClassGroup;
-
- // Calculate the index within the size class group
- int mod = (size - 1) >> log2Delta & (1 << log2SizeClassGroup) - 1;
- return group + mod + 1;
- }
-
- /**
- * Returns the total number of size classes.
- *
- * @return The total number of size classes.
- */
- public int getSizeClassNum() {
- return sizeIdx2sizeTab.length;
- }
-
- /**
- * Calculates the memory block size for a given log2 group, delta, and log2
delta.
- *
- * @param log2Group The log2 value of the current size class group.
- * @param delta The delta value for the size class.
- * @param log2Delta The log2 value of the delta.
- * @return The calculated memory block size.
- */
- private static int calculateSize(int log2Group, int delta, int log2Delta) {
- return (1 << log2Group) + (delta << log2Delta);
- }
-
- /**
- * Calculates the log2 value of a given integer.
- *
- * @param val The value to calculate the log2 for.
- * @return The log2 value of the given integer.
- */
- private static int log2(int val) {
- return INTEGER_SIZE_MINUS_ONE - Integer.numberOfLeadingZeros(val);
- }
-
- /**
- * Initializes the size class table by calculating the memory block sizes
for each size class.
- *
- * @param minSize The minimum memory block size.
- * @param maxSize The maximum memory block size.
- */
- private void initializeSizeClasses(int minSize, int maxSize) {
- int nDeltaLimit = 1 << log2SizeClassGroup;
- int log2Group = log2MinSize;
- int log2Delta = log2MinSize - log2SizeClassGroup;
-
- int sizeCount = 0;
- int size = calculateSize(log2Group, 0, log2Delta);
- sizeIdx2sizeTab[sizeCount++] = size; // Initial size
-
- // Iterate through the remaining size classes and calculate their sizes
- for (; size < maxSize; log2Group++, log2Delta++) {
- for (int nDelta = 1; nDelta <= nDeltaLimit && size <= maxSize; nDelta++)
{
- size = calculateSize(log2Group, nDelta, log2Delta);
- sizeIdx2sizeTab[sizeCount++] = size;
- }
- }
- }
-}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
index a0b62cd6a62..f0d3b02dcee 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
@@ -210,8 +210,6 @@ public enum ThreadName {
STORAGE_ENGINE_RECOVER_TRIGGER("StorageEngine-RecoverTrigger"),
FILE_TIME_INDEX_RECORD("FileTimeIndexRecord"),
TABLE_SIZE_INDEX_RECORD("TableSizeIndexRecord"),
- BINARY_ALLOCATOR_SAMPLE_EVICTOR("BinaryAllocator-SampleEvictor"),
- BINARY_ALLOCATOR_AUTO_RELEASER("BinaryAllocator-Auto-Releaser"),
FIND_EARLIEST_TIME_SLOT_PARALLEL_POOL("FindEarliestTimeSlot-Parallel-Pool"),
DATA_PARTITION_RECOVER_PARALLEL_POOL("DataPartitionRecover-Parallel-Pool"),
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index cb5902986b8..6b26b9d544d 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -467,16 +467,6 @@ public class CommonConfig {
private long seriesLimitThreshold = -1;
private long deviceLimitThreshold = -1;
- private boolean enableBinaryAllocator = true;
-
- private int arenaNum = 4;
-
- private int minAllocateSize = 4096;
-
- private int maxAllocateSize = 1024 * 1024;
-
- private int log2SizeClassGroup = 3;
-
// time in nanosecond precision when starting up
private final long startUpNanosecond = System.nanoTime();
@@ -3031,46 +3021,6 @@ public class CommonConfig {
this.remoteWriteMaxRetryDurationInMs = remoteWriteMaxRetryDurationInMs;
}
- public int getArenaNum() {
- return arenaNum;
- }
-
- public void setArenaNum(int arenaNum) {
- this.arenaNum = arenaNum;
- }
-
- public int getMinAllocateSize() {
- return minAllocateSize;
- }
-
- public void setMinAllocateSize(int minAllocateSize) {
- this.minAllocateSize = minAllocateSize;
- }
-
- public int getMaxAllocateSize() {
- return maxAllocateSize;
- }
-
- public void setMaxAllocateSize(int maxAllocateSize) {
- this.maxAllocateSize = maxAllocateSize;
- }
-
- public boolean isEnableBinaryAllocator() {
- return enableBinaryAllocator;
- }
-
- public void setEnableBinaryAllocator(boolean enableBinaryAllocator) {
- this.enableBinaryAllocator = enableBinaryAllocator;
- }
-
- public int getLog2SizeClassGroup() {
- return log2SizeClassGroup;
- }
-
- public void setLog2SizeClassGroup(int log2SizeClassGroup) {
- this.log2SizeClassGroup = log2SizeClassGroup;
- }
-
public int getPathLogMaxSize() {
return pathLogMaxSize;
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
index b249dea07ee..cd187a98bed 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
@@ -342,7 +342,6 @@ public class CommonDescriptor {
"path_log_max_size",
String.valueOf(config.getPathLogMaxSize()))));
loadRetryProperties(properties);
- loadBinaryAllocatorProps(properties);
}
private void loadSubscriptionProps(TrimProperties properties) {
@@ -647,30 +646,6 @@ public class CommonDescriptor {
String.valueOf(config.getSubscriptionConsensusWalRetentionTimeMs()))));
}
- public void loadBinaryAllocatorProps(TrimProperties properties) {
- config.setEnableBinaryAllocator(
- Boolean.parseBoolean(
- properties.getProperty(
- "enable_binary_allocator",
Boolean.toString(config.isEnableBinaryAllocator()))));
- config.setMinAllocateSize(
- Integer.parseInt(
- properties.getProperty(
- "small_blob_object",
String.valueOf(config.getMinAllocateSize()))));
- config.setMaxAllocateSize(
- Integer.parseInt(
- properties.getProperty(
- "huge_blob_object",
String.valueOf(config.getMaxAllocateSize()))));
- int arenaNum =
- Integer.parseInt(properties.getProperty("arena_num",
String.valueOf(config.getArenaNum())));
- if (arenaNum > 0) {
- config.setArenaNum(arenaNum);
- }
- config.setLog2SizeClassGroup(
- Integer.parseInt(
- properties.getProperty(
- "log2_size_class_group",
String.valueOf(config.getLog2SizeClassGroup()))));
- }
-
public void loadGlobalConfig(TGlobalConfig globalConfig) {
config.setTimestampPrecision(globalConfig.timestampPrecision);
config.setTimePartitionOrigin(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/JvmGcMonitorMetrics.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/JvmGcMonitorMetrics.java
index c28f9f34a71..0bdd863950d 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/JvmGcMonitorMetrics.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/JvmGcMonitorMetrics.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.commons.service.metric;
-import org.apache.iotdb.commons.binaryallocator.BinaryAllocator;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
@@ -121,9 +120,6 @@ public class JvmGcMonitorMetrics implements IMetricSet {
if (alertHandler != null && curData.getGcTimePercentage() >
MAX_GC_TIME_PERCENTAGE) {
alertHandler.alert(curData.clone());
}
-
- // Run GC eviction
- BinaryAllocator.getInstance().runGcEviction(curData.getGcTimePercentage());
}
private long getTotalGCTime() {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
index b35d2bc3575..c6e0213630f 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
@@ -237,7 +237,6 @@ public enum Metric {
LOAD_TIME_COST("load_time_cost"),
LOAD_POINT_COUNT("load_point_count"),
MEMTABLE_POINT_COUNT("memtable_point_count"),
- BINARY_ALLOCATOR("binary_allocator"),
// memory related
MEMORY_THRESHOLD_SIZE("memory_threshold_size"),
MEMORY_ACTUAL_SIZE("memory_actual_size"),
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocatorTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocatorTest.java
deleted file mode 100644
index df11f871118..00000000000
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocatorTest.java
+++ /dev/null
@@ -1,171 +0,0 @@
-/*
- * 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.commons.binaryallocator;
-
-import org.apache.iotdb.commons.binaryallocator.config.AllocatorConfig;
-import org.apache.iotdb.commons.binaryallocator.utils.SizeClasses;
-
-import org.apache.tsfile.utils.PooledBinary;
-import org.awaitility.Awaitility;
-import org.junit.Test;
-
-import java.util.Collections;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-public class BinaryAllocatorTest {
-
- @Test
- public void testAllocateBinary() {
- AllocatorConfig config = new AllocatorConfig();
- config.arenaNum = 1;
- BinaryAllocator binaryAllocator = new BinaryAllocator(config);
- binaryAllocator.resetArenaBinding();
-
- PooledBinary binary = binaryAllocator.allocateBinary(255, false);
- assertNotNull(binary);
- assertEquals(binary.getArenaIndex(), -1);
- assertEquals(binary.getLength(), 255);
- binaryAllocator.deallocateBinary(binary);
-
- binary = binaryAllocator.allocateBinary(65536, false);
- assertNotNull(binary);
- assertEquals(binary.getArenaIndex(), 0);
- assertEquals(binary.getLength(), 65536);
- binaryAllocator.deallocateBinary(binary);
-
- binary = binaryAllocator.allocateBinary(65535, false);
- assertNotNull(binary);
- assertEquals(binary.getArenaIndex(), 0);
- assertEquals(binary.getLength(), 65535);
- assertEquals(binary.getValues().length, 65536);
- binaryAllocator.deallocateBinary(binary);
- }
-
- @Test
- public void testStrategy() throws InterruptedException {
- BinaryAllocator binaryAllocator = new
BinaryAllocator(AllocatorConfig.DEFAULT_CONFIG);
- binaryAllocator.resetArenaBinding();
-
- PooledBinary binary1 = binaryAllocator.allocateBinary(4096, false);
- PooledBinary binary2 = binaryAllocator.allocateBinary(4096, false);
- assertEquals(binary1.getArenaIndex(), binary2.getArenaIndex());
- binaryAllocator.deallocateBinary(binary1);
- binaryAllocator.deallocateBinary(binary2);
-
- int threadCount = 4;
- CountDownLatch latch = new CountDownLatch(threadCount);
- Map<Integer, Integer> arenaUsageCount = new ConcurrentHashMap<>();
- for (int i = 0; i < threadCount; i++) {
- Thread thread =
- new Thread(
- () -> {
- try {
- PooledBinary firstBinary =
binaryAllocator.allocateBinary(2048, false);
- int arenaId = firstBinary.getArenaIndex();
- arenaUsageCount.merge(arenaId, 1, Integer::sum);
- binaryAllocator.deallocateBinary(firstBinary);
- } finally {
- latch.countDown();
- }
- });
- thread.start();
- }
-
- latch.await();
- int maxUsage = Collections.max(arenaUsageCount.values());
- int minUsage = Collections.min(arenaUsageCount.values());
- assertEquals(maxUsage, minUsage);
- }
-
- @Test
- public void testEviction() {
- AllocatorConfig config = new AllocatorConfig();
- config.arenaNum = 1;
- config.minAllocateSize = config.maxAllocateSize = 4096;
- config.setTimeBetweenEvictorRunsMillis(1);
- BinaryAllocator binaryAllocator = new BinaryAllocator(config);
- binaryAllocator.resetArenaBinding();
-
- PooledBinary binary = binaryAllocator.allocateBinary(4096, false);
- binaryAllocator.deallocateBinary(binary);
- assertEquals(4096, binaryAllocator.getTotalUsedMemory());
- Awaitility.await()
- .atMost(20, TimeUnit.SECONDS)
- .until(() -> binaryAllocator.getTotalUsedMemory() == 0);
- }
-
- @Test
- public void testSizeMapping() {
- AllocatorConfig config = new AllocatorConfig();
- config.minAllocateSize = 4096;
- config.maxAllocateSize = 65536;
- SizeClasses sizeClasses = new SizeClasses(config);
-
- assertEquals(sizeClasses.getSizeClassNum(), 33);
- int[] testSizes = {4607, 8191, 16383, 32767, 65535};
-
- for (int size : testSizes) {
- int sizeIdx = sizeClasses.size2SizeIdx(size);
- int mappedSize = sizeClasses.sizeIdx2size(sizeIdx);
-
- assertEquals("Mapped size should be >= original size", mappedSize, size
+ 1);
-
- if (sizeIdx > 0) {
- int previousSize = sizeClasses.sizeIdx2size(sizeIdx - 1);
- assertTrue("Previous size should be < original size", previousSize <
size);
- }
- }
- }
-
- @Test
- public void testAutoRelease() throws InterruptedException {
- AllocatorConfig config = new AllocatorConfig();
- config.minAllocateSize = 4096;
- config.maxAllocateSize = 65536;
- BinaryAllocator binaryAllocator = new BinaryAllocator(config);
- binaryAllocator.resetArenaBinding();
-
- PooledBinary binary = binaryAllocator.allocateBinary(4096, true);
- assertNotNull(binary);
- assertEquals(binary.getArenaIndex(), -1);
- assertEquals(binary.getLength(), 4096);
- assertEquals(binaryAllocator.getTotalUsedMemory(), 0);
-
- // reference count is 0
- binary = null;
- System.gc();
- long startTime = System.currentTimeMillis();
- while (System.currentTimeMillis() - startTime <=
TimeUnit.MINUTES.toMillis(1)) {
- if (binaryAllocator.getTotalUsedMemory() == 4096) {
- return;
- }
- Thread.sleep(100);
- }
- fail("Can not auto release PoolBinary in binary allocator");
- }
-}
