This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 0d3304a106a Fix RPC buffer memory control activation (#18194)
0d3304a106a is described below
commit 0d3304a106a4ef2ea168682485ac0b7027f93f95
Author: Caideyipi <[email protected]>
AuthorDate: Tue Jul 14 09:32:30 2026 +0800
Fix RPC buffer memory control activation (#18194)
Avoid activating DataNode RPC buffer memory control during configuration
loading, and bind auto-resizing buffers to the memory account used for their
allocation.
---
.../org/apache/iotdb/rpc/AutoResizingBuffer.java | 24 +++++++++----
.../iotdb/rpc/AutoResizingBufferMemoryManager.java | 25 ++++++++++++--
.../apache/iotdb/rpc/AutoResizingBufferTest.java | 37 ++++++++++++++++++++
.../iotdb/confignode/service/ConfigNode.java | 2 --
.../apache/iotdb/db/conf/DataNodeMemoryConfig.java | 12 +++++--
.../java/org/apache/iotdb/db/service/DataNode.java | 1 +
.../iotdb/db/conf/DataNodeMemoryConfigTest.java | 40 ++++++++++++++++++++++
.../apache/iotdb/commons/memory/MemoryConfig.java | 27 ++++++---------
8 files changed, 137 insertions(+), 31 deletions(-)
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoResizingBuffer.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoResizingBuffer.java
index 397a5843171..aa0ea40541c 100644
---
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoResizingBuffer.java
+++
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoResizingBuffer.java
@@ -40,11 +40,18 @@ class AutoResizingBuffer implements AutoCloseable {
private long lastShrinkTime;
private int accountedCapacity;
private boolean closed;
+ private AutoResizingBufferMemoryManager.MemoryAccount memoryAccount;
public AutoResizingBuffer(int initialCapacity) throws IOException {
- AutoResizingBufferMemoryManager.allocate(initialCapacity);
- this.array = new byte[initialCapacity];
this.initialCapacity = initialCapacity;
+ this.memoryAccount = AutoResizingBufferMemoryManager.createMemoryAccount();
+ memoryAccount.allocate(initialCapacity);
+ try {
+ this.array = new byte[initialCapacity];
+ } catch (RuntimeException | Error e) {
+ memoryAccount.release(initialCapacity);
+ throw e;
+ }
this.accountedCapacity = initialCapacity;
}
@@ -76,7 +83,7 @@ class AutoResizingBuffer implements AutoCloseable {
@Override
public void close() {
if (!closed) {
- AutoResizingBufferMemoryManager.release(accountedCapacity);
+ memoryAccount.release(accountedCapacity);
accountedCapacity = 0;
closed = true;
}
@@ -86,25 +93,28 @@ class AutoResizingBuffer implements AutoCloseable {
final int currentCapacity = array.length;
if (newCapacity > currentCapacity) {
final int delta = newCapacity - currentCapacity;
- AutoResizingBufferMemoryManager.allocate(delta);
+ memoryAccount.allocate(delta);
try {
array = Arrays.copyOf(array, newCapacity);
accountedCapacity += delta;
} catch (RuntimeException | Error e) {
- AutoResizingBufferMemoryManager.release(delta);
+ memoryAccount.release(delta);
throw e;
}
} else if (newCapacity < currentCapacity) {
array = Arrays.copyOf(array, newCapacity);
final int delta = currentCapacity - newCapacity;
- AutoResizingBufferMemoryManager.release(delta);
+ memoryAccount.release(delta);
accountedCapacity -= delta;
}
}
private void reserveCurrentCapacityIfReleased() throws IOException {
if (closed) {
- AutoResizingBufferMemoryManager.allocate(array.length);
+ AutoResizingBufferMemoryManager.MemoryAccount newMemoryAccount =
+ AutoResizingBufferMemoryManager.createMemoryAccount();
+ newMemoryAccount.allocate(array.length);
+ memoryAccount = newMemoryAccount;
accountedCapacity = array.length;
closed = false;
}
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoResizingBufferMemoryManager.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoResizingBufferMemoryManager.java
index d95d51adb48..d33a15b44cb 100644
---
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoResizingBufferMemoryManager.java
+++
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoResizingBufferMemoryManager.java
@@ -68,7 +68,12 @@ public final class AutoResizingBufferMemoryManager {
memoryAllocateRetryIntervalInMs;
}
- static void allocate(long sizeInBytes) throws IOException {
+ static MemoryAccount createMemoryAccount() {
+ return new MemoryAccount(memoryControl);
+ }
+
+ private static void allocate(AutoResizingBufferMemoryControl memoryControl,
long sizeInBytes)
+ throws IOException {
if (sizeInBytes <= 0) {
return;
}
@@ -93,7 +98,7 @@ public final class AutoResizingBufferMemoryManager {
MEMORY_ALLOCATE_MAX_RETRIES));
}
- static void release(long sizeInBytes) {
+ private static void release(AutoResizingBufferMemoryControl memoryControl,
long sizeInBytes) {
if (sizeInBytes <= 0) {
return;
}
@@ -107,4 +112,20 @@ public final class AutoResizingBufferMemoryManager {
public static long getMemoryAllocationFailureCount() {
return memoryAllocationFailureCount.get();
}
+
+ static final class MemoryAccount {
+ private final AutoResizingBufferMemoryControl memoryControl;
+
+ private MemoryAccount(AutoResizingBufferMemoryControl memoryControl) {
+ this.memoryControl = memoryControl;
+ }
+
+ void allocate(long sizeInBytes) throws IOException {
+ AutoResizingBufferMemoryManager.allocate(memoryControl, sizeInBytes);
+ }
+
+ void release(long sizeInBytes) {
+ AutoResizingBufferMemoryManager.release(memoryControl, sizeInBytes);
+ }
+ }
}
diff --git
a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/AutoResizingBufferTest.java
b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/AutoResizingBufferTest.java
index 800881208ee..164d0f9e287 100644
---
a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/AutoResizingBufferTest.java
+++
b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/AutoResizingBufferTest.java
@@ -152,6 +152,43 @@ public class AutoResizingBufferTest {
}
}
+ @Test
+ public void testBufferReleasesMemoryToTheControlUsedForAllocation() throws
Exception {
+ TestMemoryControl firstMemoryControl = new TestMemoryControl();
+ AutoResizingBufferMemoryManager.setMemoryControl(firstMemoryControl);
+ AutoResizingBuffer firstBuffer = new AutoResizingBuffer(100);
+
+ TestMemoryControl secondMemoryControl = new TestMemoryControl();
+ AutoResizingBufferMemoryManager.setMemoryControl(secondMemoryControl);
+ AutoResizingBuffer secondBuffer = new AutoResizingBuffer(200);
+
+ firstBuffer.close();
+ Assert.assertEquals(0, firstMemoryControl.getUsedMemoryInBytes());
+ Assert.assertEquals(200, secondMemoryControl.getUsedMemoryInBytes());
+
+ secondBuffer.close();
+ Assert.assertEquals(0, secondMemoryControl.getUsedMemoryInBytes());
+ }
+
+ @Test
+ public void testBufferCreatedWithoutControlDoesNotReleaseToNewControl()
throws Exception {
+ AutoResizingBuffer bufferWithoutControl = new AutoResizingBuffer(100);
+
+ AutoResizingBufferMemoryManager.setMemoryControl(memoryControl);
+ AutoResizingBuffer bufferWithControl = new AutoResizingBuffer(100);
+
+ bufferWithoutControl.close();
+ Assert.assertEquals(100, memoryControl.getUsedMemoryInBytes());
+
+ bufferWithoutControl.resizeIfNecessary(100);
+ Assert.assertEquals(200, memoryControl.getUsedMemoryInBytes());
+ bufferWithoutControl.close();
+ Assert.assertEquals(100, memoryControl.getUsedMemoryInBytes());
+
+ bufferWithControl.close();
+ Assert.assertEquals(0, memoryControl.getUsedMemoryInBytes());
+ }
+
private void setLastShrinkTime(AutoResizingBuffer buffer, long
lastShrinkTime) throws Exception {
Field field = AutoResizingBuffer.class.getDeclaredField("lastShrinkTime");
field.setAccessible(true);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java
index c5e0a95294b..ffd5ec2d33b 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java
@@ -34,7 +34,6 @@ import
org.apache.iotdb.commons.exception.ConfigurationException;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.exception.StartupException;
-import org.apache.iotdb.commons.memory.MemoryConfig;
import org.apache.iotdb.commons.service.JMXService;
import org.apache.iotdb.commons.service.RegisterManager;
import org.apache.iotdb.commons.service.ServiceType;
@@ -143,7 +142,6 @@ public class ConfigNode extends ServerCommandLine
implements ConfigNodeMBean {
LOGGER.info(ConfigNodeMessages.STARTING_IOTDB,
IoTDBConstant.VERSION_WITH_BUILD);
ConfigNodeStartupCheck checks = new
ConfigNodeStartupCheck(IoTDBConstant.CN_ROLE);
checks.startUpCheck();
- MemoryConfig.getInstance();
} catch (StartupException | ConfigurationException | IOException e) {
LOGGER.error(ConfigNodeMessages.MEET_ERROR_WHEN_DOING_START_CHECKING, e);
throw new IoTDBException(ConfigNodeMessages.ERROR_STARTING, -1);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java
index 6612732441e..9476e84e15e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java
@@ -77,6 +77,9 @@ public class DataNodeMemoryConfig {
/** The memory manager of on heap */
private MemoryManager onHeapMemoryManager;
+ /** Memory budget for RPC auto-resizing buffers */
+ private long autoResizingBufferMemorySize;
+
/** Memory manager for the write process */
private MemoryManager storageEngineMemoryManager;
@@ -168,7 +171,7 @@ public class DataNodeMemoryConfig {
long schemaEngineMemorySize = Runtime.getRuntime().maxMemory() / 10;
long consensusMemorySize = Runtime.getRuntime().maxMemory() / 10;
long pipeMemorySize = Runtime.getRuntime().maxMemory() / 10;
- long autoResizingBufferMemorySize = Runtime.getRuntime().maxMemory() / 20;
+ autoResizingBufferMemorySize = Runtime.getRuntime().maxMemory() / 20;
if (memoryAllocateProportion != null) {
String[] proportions = memoryAllocateProportion.split(":");
int proportionSum = 0;
@@ -217,8 +220,6 @@ public class DataNodeMemoryConfig {
consensusMemoryManager =
onHeapMemoryManager.getOrCreateMemoryManager("Consensus",
consensusMemorySize);
pipeMemoryManager = onHeapMemoryManager.getOrCreateMemoryManager("Pipe",
pipeMemorySize);
- MemoryConfig.getInstance()
- .setAutoResizingBufferMemoryControl(onHeapMemoryManager,
autoResizingBufferMemorySize);
LOGGER.info(
DataNodeMiscMessages.MISC_LOG_INITIAL_ALLOCATEMEMORYFORWRITE_B90EC7D9,
storageEngineMemoryManager.getTotalMemorySizeInBytes());
@@ -260,6 +261,11 @@ public class DataNodeMemoryConfig {
"DirectBuffer", totalDirectBufferMemorySizeLimit);
}
+ public void activateAutoResizingBufferMemoryControl() {
+ MemoryConfig.getInstance()
+ .setAutoResizingBufferMemoryControl(onHeapMemoryManager,
autoResizingBufferMemorySize);
+ }
+
@SuppressWarnings("squid:S3518")
private void initSchemaMemoryAllocate(
MemoryManager schemaEngineMemoryManager, TrimProperties properties) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
index 5fc67446925..0489adb7646 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -253,6 +253,7 @@ public class DataNode extends ServerCommandLine implements
DataNodeMBean {
logger.info(DataNodeMiscMessages.STARTING_DATANODE);
boolean isFirstStart;
try {
+
IoTDBDescriptor.getInstance().getMemoryConfig().activateAutoResizingBufferMemoryControl();
// Check if this DataNode is start for the first time and do other
pre-checks
isFirstStart = prepareDataNode();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/DataNodeMemoryConfigTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/DataNodeMemoryConfigTest.java
new file mode 100644
index 00000000000..0d3447bbefb
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/DataNodeMemoryConfigTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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.conf;
+
+import org.apache.iotdb.commons.memory.MemoryConfig;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class DataNodeMemoryConfigTest {
+
+ @Test
+ public void testRpcMemoryControlIsActivatedOnlyExplicitly() {
+ DataNodeMemoryConfig memoryConfig =
IoTDBDescriptor.getInstance().getMemoryConfig();
+
+ Assert.assertEquals(
+ 0,
MemoryConfig.getInstance().getAutoResizingBufferMemoryTotalSizeInBytes());
+
+ memoryConfig.activateAutoResizingBufferMemoryControl();
+
+
Assert.assertTrue(MemoryConfig.getInstance().getAutoResizingBufferMemoryTotalSizeInBytes()
> 0);
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryConfig.java
index c828120eab5..52b08556e3f 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryConfig.java
@@ -35,7 +35,6 @@ public class MemoryConfig {
private boolean isAutoResizingBufferMemoryControlEnabled;
private MemoryConfig() {
- initAutoResizingBufferMemoryControl();
MetricService.getInstance().addMetricSet(new
AutoResizingBufferMemoryMetrics(this));
}
@@ -53,39 +52,33 @@ public class MemoryConfig {
private MemoryConfigHolder() {}
}
- private void initAutoResizingBufferMemoryControl() {
+ private void initAutoResizingBufferMemoryControl(IMemoryBlock memoryBlock) {
AutoResizingBufferMemoryManager.setMemoryControl(
new AutoResizingBufferMemoryControl() {
@Override
- public synchronized boolean allocate(long sizeInBytes) {
- IMemoryBlock memoryBlock = getAutoResizingBufferMemoryBlock();
- if (memoryBlock == null) {
- return true;
- }
+ public boolean allocate(long sizeInBytes) {
return memoryBlock.allocate(sizeInBytes);
}
@Override
- public synchronized void release(long sizeInBytes) {
- IMemoryBlock memoryBlock = getAutoResizingBufferMemoryBlock();
- if (memoryBlock != null) {
- memoryBlock.release(sizeInBytes);
- }
+ public void release(long sizeInBytes) {
+ memoryBlock.release(sizeInBytes);
}
});
}
+ /**
+ * Installs RPC auto-resizing buffer memory control once. Replacing a live
control would orphan
+ * the memory blocks retained by existing RPC buffers.
+ */
public synchronized void setAutoResizingBufferMemoryControl(
MemoryManager parentMemoryManager, long memorySizeInBytes) {
if (autoResizingBufferMemoryManagerParent != null) {
- autoResizingBufferMemoryManagerParent.releaseChildMemoryManager(
- AUTO_RESIZING_BUFFER_MEMORY_MANAGER_NAME);
+ return;
}
autoResizingBufferMemoryManagerParent = parentMemoryManager;
- autoResizingBufferMemoryBlock = null;
if (memorySizeInBytes <= 0) {
- isAutoResizingBufferMemoryControlEnabled = false;
return;
}
@@ -93,13 +86,13 @@ public class MemoryConfig {
parentMemoryManager.getOrCreateMemoryManager(
AUTO_RESIZING_BUFFER_MEMORY_MANAGER_NAME, memorySizeInBytes, true);
if (autoResizingBufferMemoryManager == null) {
- isAutoResizingBufferMemoryControlEnabled = false;
return;
}
autoResizingBufferMemoryBlock =
autoResizingBufferMemoryManager.exactAllocate(
AUTO_RESIZING_BUFFER_MEMORY_BLOCK_NAME, memorySizeInBytes,
MemoryBlockType.DYNAMIC);
isAutoResizingBufferMemoryControlEnabled = true;
+ initAutoResizingBufferMemoryControl(autoResizingBufferMemoryBlock);
}
private synchronized IMemoryBlock getAutoResizingBufferMemoryBlock() {
