From 4f2eb0c790484827f5ca1a00e5573940d64c957f Mon Sep 17 00:00:00 2001 From: JackieTien97 Date: Sun, 6 Sep 2026 08:47:32 +0800 Subject: [PATCH] Limit Edge query threads and remove unused binary allocator --- .../iotdb/edge/it/IoTDBEdgeBasicIT.java | 124 +++++++ .../org/apache/iotdb/db/conf/IoTDBConfig.java | 30 ++ .../apache/iotdb/db/conf/IoTDBDescriptor.java | 26 +- .../execution/schedule/DriverTaskThread.java | 35 +- .../db/queryengine/plan/Coordinator.java | 5 +- .../db/conf/QueryThreadPoolConfigTest.java | 29 ++ .../schedule/DriverTaskThreadTest.java | 120 +++++++ .../conf/edge/iotdb-system.properties | 2 + .../conf/iotdb-system.properties.template | 49 +-- .../iotdb/commons/i18n/CommonMessages.java | 14 +- .../iotdb/commons/i18n/CommonMessages.java | 15 +- .../binaryallocator/BinaryAllocator.java | 311 ------------------ .../binaryallocator/BinaryAllocatorState.java | 71 ---- .../PooledBinaryPhantomReference.java | 42 --- .../commons/binaryallocator/arena/Arena.java | 273 --------------- .../binaryallocator/arena/ArenaStrategy.java | 35 -- .../autoreleaser/Releaser.java | 93 ------ .../config/AllocatorConfig.java | 53 --- .../ema/AdaptiveWeightedAverage.java | 100 ------ .../binaryallocator/evictor/Evictor.java | 102 ------ .../metric/BinaryAllocatorMetrics.java | 138 -------- .../binaryallocator/utils/SizeClasses.java | 146 -------- .../iotdb/commons/concurrent/ThreadName.java | 2 - .../iotdb/commons/conf/CommonConfig.java | 50 --- .../iotdb/commons/conf/CommonDescriptor.java | 25 -- .../service/metric/JvmGcMonitorMetrics.java | 4 - .../commons/service/metric/enums/Metric.java | 1 - .../binaryallocator/BinaryAllocatorTest.java | 171 ---------- 28 files changed, 369 insertions(+), 1697 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskThreadTest.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocator.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocatorState.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/PooledBinaryPhantomReference.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/arena/Arena.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/arena/ArenaStrategy.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/autoreleaser/Releaser.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/config/AllocatorConfig.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/ema/AdaptiveWeightedAverage.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/evictor/Evictor.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/metric/BinaryAllocatorMetrics.java delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/binaryallocator/utils/SizeClasses.java delete mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/binaryallocator/BinaryAllocatorTest.java 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 57a8189aa8425..07ab860655d6d 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.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.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.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 static void setUp() throws Exception { 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 @@ private static Connection openTreeConnection() throws SQLException { 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> 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 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 variables = new LinkedHashMap<>(); @@ -280,6 +402,8 @@ public void testPackagedConfiguration() throws Exception { 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 b14f2e1b6222b..989031fc0fd12 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 void setFragmentInstanceNotificationThreadCount( 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 44971c718a104..9f48bac0e0cd5 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 void loadProperties(TrimProperties properties) throws BadNodeUrlException 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 synchronized void loadHotModifiedProps(TrimProperties properties) // 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 912bd6711e34e..1c44cb82f0cd9 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.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 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 DriverTaskThread( IndexedBlockingQueue queue, ITaskScheduler scheduler, ThreadProducer producer) { + this(workerId, tg, queue, scheduler, producer, NOTIFICATION_EXECUTOR); + } + + DriverTaskThread( + String workerId, + ThreadGroup tg, + IndexedBlockingQueue 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 09a35803595f8..fad2ea1354d60 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 @@ private 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 92b6d0fd45ffa..8ac89c9f6f2cf 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 void testDefaultsAndPositiveSizes() throws Exception { () -> 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 void testStartupOverridesAreRestartOnly() throws Exception { 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 0000000000000..5ea8b9af35c35 --- /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 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> futures = new ArrayList<>(); + List 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 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 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 47a67d1cf3860..82679f99e6a39 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 95a8154bba4d4..97fd112c2751a 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 @@ -1244,6 +1244,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 @@ -1841,41 +1855,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 a6b958fd1420f..a05c8f9915a1d 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 c3af2dbe7e24a..c5c6290687e67 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 175651551d66a..0000000000000 --- 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 state = - new AtomicReference<>(BinaryAllocatorState.UNINITIALIZED); - - private final BinaryAllocatorMetrics metrics; - private Evictor sampleEvictor; - private Releaser autoReleaser; - private static final ThreadLocal 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 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 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 6812bc84b1161..0000000000000 --- 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. - * - * 
- *     ----------------------------------------
- *     |                                      |
- *     |              ----------              |
- *     |              |        |              |
- *     |              v        |              v
- * UNINITIALIZED --> OPEN ---> PENDING -->  CLOSE
- *                    ^                       ^
- *                    |                       |
- *                    -------------------------
- *
- * - * State Transition Logic: - * - *
    - *
  • UNINITIALIZED -> CLOSE: When enable_binary_allocator = false - *
  • UNINITIALIZED -> OPEN: When enable_binary_allocator = true - *
  • OPEN -> CLOSE: When enable_binary_allocator is hot reload to false - *
  • CLOSE -> OPEN: When enable_binary_allocator is hot reload to true - *
  • PENDING -> CLOSE: When enable_binary_allocator is hot reload to false - *
  • OPEN -> PENDING: When in OPEN state and GC time percentage exceeds 30%, indicating - * allocator ineffectiveness - *
  • PENDING -> OPEN: 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. - *
- */ -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 84541219c082f..0000000000000 --- 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 { - public final byte[] byteArray; - public Arena.SlabRegion slabRegion; - - public PooledBinaryPhantomReference( - PooledBinary referent, - ReferenceQueue 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 b5d95234aa73a..0000000000000 --- 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 referenceQueue; - private final Set 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 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 ec4af3fbe0a38..0000000000000 --- 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 26b9607ad9fd7..0000000000000 --- 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 3bbbbc9a2104d..0000000000000 --- 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 201f2f63ae10e..0000000000000 --- 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 JDK17 - * AdaptiveWeightedAverage. But some necessary modifications are made to adapt to the usage of - * binary allocator: - * - *

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 d310ef1f15e6c..0000000000000 --- 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 305dadc432b09..0000000000000 --- 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 e13cc54a9f14e..0000000000000 --- 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 606a08bde3322..6a0912223aeff 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 @@ -209,8 +209,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 cb5902986b8e8..6b26b9d544dd9 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 void setRemoteWriteMaxRetryDurationInMs(long remoteWriteMaxRetryDurationI 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 b249dea07ee16..cd187a98bede2 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 void loadCommonProps(TrimProperties properties) throws IOException { "path_log_max_size", String.valueOf(config.getPathLogMaxSize())))); loadRetryProperties(properties); - loadBinaryAllocatorProps(properties); } private void loadSubscriptionProps(TrimProperties properties) { @@ -647,30 +646,6 @@ public void loadHotModifiedSubscriptionConsensusProps(final TrimProperties prope 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 c28f9f34a71f0..0bdd863950dd1 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 @@ private void scheduledMonitoring() { 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 b35d2bc35752b..c6e0213630fa6 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 df11f87111881..0000000000000 --- 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 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"); - } -}