This is an automated email from the ASF dual-hosted git repository.

jt2594838 pushed a commit to branch add_rpc_memory_control
in repository https://gitbox.apache.org/repos/asf/iotdb.git

commit f8566275c2aa4de5097e1faa75eeaefe703d0526
Author: Tian Jiang <[email protected]>
AuthorDate: Wed Jun 10 17:47:57 2026 +0800

    ver1
---
 .../org/apache/iotdb/rpc/AutoResizingBuffer.java   | 53 ++++++++++++++++++++--
 .../iotdb/rpc/AutoResizingBufferMemoryControl.java | 25 ++--------
 .../iotdb/rpc/AutoScalingBufferReadTransport.java  | 18 ++++++--
 .../iotdb/rpc/AutoScalingBufferWriteTransport.java | 20 ++++++--
 .../rpc/TCompressedElasticFramedTransport.java     | 17 +++++--
 .../apache/iotdb/rpc/TElasticFramedTransport.java  | 22 ++++++---
 .../iotdb/rpc/TSnappyElasticFramedTransport.java   |  8 ++--
 .../rpc/TimeoutChangeableTFastFramedTransport.java |  6 ++-
 .../TimeoutChangeableTSnappyFramedTransport.java   |  6 ++-
 .../apache/iotdb/consensus/ratis/utils/Utils.java  | 12 ++++-
 .../apache/iotdb/commons/memory/MemoryConfig.java  | 47 ++++++++++++++++++-
 11 files changed, 183 insertions(+), 51 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 180d05b3d93..397a5843171 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
@@ -19,6 +19,7 @@
 
 package org.apache.iotdb.rpc;
 
+import java.io.IOException;
 import java.util.Arrays;
 
 /**
@@ -31,33 +32,38 @@ import java.util.Arrays;
  * required size < current capacity * 0.6, and such small requests last for 
more than 5 times,
  * shrink to the middle of the required size and current capacity.
  */
-class AutoResizingBuffer {
+class AutoResizingBuffer implements AutoCloseable {
 
   private byte[] array;
   private int bufTooLargeCounter = RpcUtils.MAX_BUFFER_OVERSIZE_TIME;
   private final int initialCapacity;
   private long lastShrinkTime;
+  private int accountedCapacity;
+  private boolean closed;
 
-  public AutoResizingBuffer(int initialCapacity) {
+  public AutoResizingBuffer(int initialCapacity) throws IOException {
+    AutoResizingBufferMemoryManager.allocate(initialCapacity);
     this.array = new byte[initialCapacity];
     this.initialCapacity = initialCapacity;
+    this.accountedCapacity = initialCapacity;
   }
 
-  public void resizeIfNecessary(int size) {
+  public void resizeIfNecessary(int size) throws IOException {
+    reserveCurrentCapacityIfReleased();
     final int currentCapacity = this.array.length;
     final double loadFactor = 0.6;
     if (currentCapacity < size) {
       // Increase by a factor of 1.5x
       int growCapacity = currentCapacity + (currentCapacity >> 1);
       int newCapacity = Math.max(growCapacity, size);
-      this.array = Arrays.copyOf(array, newCapacity);
+      resize(newCapacity);
       bufTooLargeCounter = RpcUtils.MAX_BUFFER_OVERSIZE_TIME;
     } else if (size > initialCapacity
         && currentCapacity * loadFactor > size
         && bufTooLargeCounter-- <= 0
         && System.currentTimeMillis() - lastShrinkTime > 
RpcUtils.MIN_SHRINK_INTERVAL) {
       // do not resize if it is reading the request size and do not shrink too 
often
-      array = Arrays.copyOf(array, size + (currentCapacity - size) / 2);
+      resize(size + (currentCapacity - size) / 2);
       bufTooLargeCounter = RpcUtils.MAX_BUFFER_OVERSIZE_TIME;
       lastShrinkTime = System.currentTimeMillis();
     }
@@ -66,4 +72,41 @@ class AutoResizingBuffer {
   public byte[] array() {
     return this.array;
   }
+
+  @Override
+  public void close() {
+    if (!closed) {
+      AutoResizingBufferMemoryManager.release(accountedCapacity);
+      accountedCapacity = 0;
+      closed = true;
+    }
+  }
+
+  private void resize(int newCapacity) throws IOException {
+    final int currentCapacity = array.length;
+    if (newCapacity > currentCapacity) {
+      final int delta = newCapacity - currentCapacity;
+      AutoResizingBufferMemoryManager.allocate(delta);
+      try {
+        array = Arrays.copyOf(array, newCapacity);
+        accountedCapacity += delta;
+      } catch (RuntimeException | Error e) {
+        AutoResizingBufferMemoryManager.release(delta);
+        throw e;
+      }
+    } else if (newCapacity < currentCapacity) {
+      array = Arrays.copyOf(array, newCapacity);
+      final int delta = currentCapacity - newCapacity;
+      AutoResizingBufferMemoryManager.release(delta);
+      accountedCapacity -= delta;
+    }
+  }
+
+  private void reserveCurrentCapacityIfReleased() throws IOException {
+    if (closed) {
+      AutoResizingBufferMemoryManager.allocate(array.length);
+      accountedCapacity = array.length;
+      closed = false;
+    }
+  }
 }
diff --git 
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryConfig.java
 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoResizingBufferMemoryControl.java
similarity index 57%
copy from 
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryConfig.java
copy to 
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoResizingBufferMemoryControl.java
index dc1ca4e5473..49128727a0b 100644
--- 
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryConfig.java
+++ 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoResizingBufferMemoryControl.java
@@ -17,27 +17,12 @@
  * under the License.
  */
 
-package org.apache.iotdb.commons.memory;
+package org.apache.iotdb.rpc;
 
-public class MemoryConfig {
-  private final MemoryManager globalMemoryManager =
-      new MemoryManager("GlobalMemoryManager", null, 
Runtime.getRuntime().totalMemory());
+/** Memory accounting hook for {@link AutoResizingBuffer}. */
+public interface AutoResizingBufferMemoryControl {
 
-  private MemoryConfig() {
-    // singleton
-  }
+  boolean allocate(long sizeInBytes);
 
-  public static MemoryManager global() {
-    return MemoryConfigHolder.INSTANCE.globalMemoryManager;
-  }
-
-  public static MemoryConfig getInstance() {
-    return MemoryConfigHolder.INSTANCE;
-  }
-
-  private static class MemoryConfigHolder {
-    private static final MemoryConfig INSTANCE = new MemoryConfig();
-
-    private MemoryConfigHolder() {}
-  }
+  void release(long sizeInBytes);
 }
diff --git 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoScalingBufferReadTransport.java
 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoScalingBufferReadTransport.java
index 35833321ba3..035f8753de1 100644
--- 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoScalingBufferReadTransport.java
+++ 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoScalingBufferReadTransport.java
@@ -23,18 +23,24 @@ import org.apache.thrift.TConfiguration;
 import org.apache.thrift.transport.TTransport;
 import org.apache.thrift.transport.TTransportException;
 
+import java.io.IOException;
+
 public class AutoScalingBufferReadTransport extends NonOpenTransport {
 
   private final AutoResizingBuffer buf;
   private int pos = 0;
   private int limit = 0;
 
-  public AutoScalingBufferReadTransport(int initialCapacity) {
+  public AutoScalingBufferReadTransport(int initialCapacity) throws 
IOException {
     this.buf = new AutoResizingBuffer(initialCapacity);
   }
 
   public void fill(TTransport inTrans, int length) throws TTransportException {
-    buf.resizeIfNecessary(length);
+    try {
+      buf.resizeIfNecessary(length);
+    } catch (IOException e) {
+      throw new TTransportException(e);
+    }
     inTrans.readAll(buf.array(), 0, length);
     pos = 0;
     limit = length;
@@ -89,10 +95,16 @@ public class AutoScalingBufferReadTransport extends 
NonOpenTransport {
     return limit - pos;
   }
 
-  public void resizeIfNecessary(int size) {
+  public void resizeIfNecessary(int size) throws IOException {
     buf.resizeIfNecessary(size);
   }
 
+  @Override
+  public void close() {
+    super.close();
+    buf.close();
+  }
+
   public void limit(int newLimit) {
     this.limit = newLimit;
   }
diff --git 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoScalingBufferWriteTransport.java
 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoScalingBufferWriteTransport.java
index ae833fad239..24278067fcd 100644
--- 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoScalingBufferWriteTransport.java
+++ 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/AutoScalingBufferWriteTransport.java
@@ -22,6 +22,8 @@ package org.apache.iotdb.rpc;
 import org.apache.thrift.TConfiguration;
 import org.apache.thrift.transport.TTransportException;
 
+import java.io.IOException;
+
 /**
  * Note that this class is mainly copied from class {@link
  * org.apache.thrift.transport.AutoExpandingBufferWriteTransport}. since that 
class does not support
@@ -32,7 +34,7 @@ public class AutoScalingBufferWriteTransport extends 
NonOpenTransport {
   private final AutoResizingBuffer buf;
   private int pos;
 
-  public AutoScalingBufferWriteTransport(int initialCapacity) {
+  public AutoScalingBufferWriteTransport(int initialCapacity) throws 
IOException {
     this.buf = new AutoResizingBuffer(initialCapacity);
     this.pos = 0;
   }
@@ -43,8 +45,12 @@ public class AutoScalingBufferWriteTransport extends 
NonOpenTransport {
   }
 
   @Override
-  public void write(byte[] toWrite, int off, int len) {
-    buf.resizeIfNecessary(pos + len);
+  public void write(byte[] toWrite, int off, int len) throws 
TTransportException {
+    try {
+      buf.resizeIfNecessary(pos + len);
+    } catch (IOException e) {
+      throw new TTransportException(e);
+    }
     System.arraycopy(toWrite, off, buf.array(), pos, len);
     pos += len;
   }
@@ -57,7 +63,13 @@ public class AutoScalingBufferWriteTransport extends 
NonOpenTransport {
     pos = 0;
   }
 
-  public void resizeIfNecessary(int size) {
+  @Override
+  public void close() {
+    super.close();
+    buf.close();
+  }
+
+  public void resizeIfNecessary(int size) throws IOException {
     buf.resizeIfNecessary(size);
   }
 
diff --git 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TCompressedElasticFramedTransport.java
 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TCompressedElasticFramedTransport.java
index 62abc28e470..3f423771dda 100644
--- 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TCompressedElasticFramedTransport.java
+++ 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TCompressedElasticFramedTransport.java
@@ -34,10 +34,15 @@ public abstract class TCompressedElasticFramedTransport 
extends TElasticFramedTr
       TTransport underlying,
       int thriftDefaultBufferSize,
       int thriftMaxFrameSize,
-      boolean copyBinary) {
+      boolean copyBinary)
+      throws TTransportException {
     super(underlying, thriftDefaultBufferSize, thriftMaxFrameSize, copyBinary);
-    writeCompressBuffer = new 
AutoScalingBufferWriteTransport(thriftDefaultBufferSize);
-    readCompressBuffer = new 
AutoScalingBufferReadTransport(thriftDefaultBufferSize);
+    try {
+      writeCompressBuffer = new 
AutoScalingBufferWriteTransport(thriftDefaultBufferSize);
+      readCompressBuffer = new 
AutoScalingBufferReadTransport(thriftDefaultBufferSize);
+    } catch (IOException e) {
+      throw new TTransportException(e);
+    }
   }
 
   @Override
@@ -80,7 +85,11 @@ public abstract class TCompressedElasticFramedTransport 
extends TElasticFramedTr
 
     writeBuffer.reset();
     if (thriftDefaultBufferSize < length) {
-      writeBuffer.resizeIfNecessary(thriftDefaultBufferSize);
+      try {
+        writeBuffer.resizeIfNecessary(thriftDefaultBufferSize);
+      } catch (IOException e) {
+        throw new TTransportException(e);
+      }
     }
     underlying.flush();
   }
diff --git 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
index 6a174f86b6a..f3208b07234 100644
--- 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
+++ 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
@@ -32,6 +32,7 @@ import javax.net.ssl.SSLException;
 import javax.net.ssl.SSLHandshakeException;
 
 import java.io.EOFException;
+import java.io.IOException;
 import java.net.SocketAddress;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
@@ -74,7 +75,7 @@ public class TElasticFramedTransport extends TTransport {
     }
 
     @Override
-    public TTransport getTransport(TTransport trans) {
+    public TTransport getTransport(TTransport trans) throws 
TTransportException {
       return new TElasticFramedTransport(
           trans, thriftDefaultBufferSize, thriftMaxFrameSize, copyBinary);
     }
@@ -84,13 +85,18 @@ public class TElasticFramedTransport extends TTransport {
       TTransport underlying,
       int thriftDefaultBufferSize,
       int thriftMaxFrameSize,
-      boolean copyBinary) {
+      boolean copyBinary)
+      throws TTransportException {
     this.underlying = underlying;
     this.thriftDefaultBufferSize = thriftDefaultBufferSize;
     this.thriftMaxFrameSize = thriftMaxFrameSize;
     this.copyBinary = copyBinary;
-    readBuffer = new AutoScalingBufferReadTransport(thriftDefaultBufferSize);
-    writeBuffer = new AutoScalingBufferWriteTransport(thriftDefaultBufferSize);
+    try {
+      readBuffer = new AutoScalingBufferReadTransport(thriftDefaultBufferSize);
+      writeBuffer = new 
AutoScalingBufferWriteTransport(thriftDefaultBufferSize);
+    } catch (IOException e) {
+      throw new TTransportException(e);
+    }
   }
 
   protected final int thriftDefaultBufferSize;
@@ -263,7 +269,11 @@ public class TElasticFramedTransport extends TTransport {
     underlying.write(writeBuffer.getBuffer(), 0, length);
     writeBuffer.reset();
     if (length > thriftDefaultBufferSize) {
-      writeBuffer.resizeIfNecessary(thriftDefaultBufferSize);
+      try {
+        writeBuffer.resizeIfNecessary(thriftDefaultBufferSize);
+      } catch (IOException e) {
+        throw new TTransportException(e);
+      }
     }
     underlying.flush();
   }
@@ -292,7 +302,7 @@ public class TElasticFramedTransport extends TTransport {
   }
 
   @Override
-  public void write(byte[] buf, int off, int len) {
+  public void write(byte[] buf, int off, int len) throws TTransportException {
     writeBuffer.write(buf, off, len);
   }
 
diff --git 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSnappyElasticFramedTransport.java
 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSnappyElasticFramedTransport.java
index 79ebc7983e3..fd9d4ffbba3 100644
--- 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSnappyElasticFramedTransport.java
+++ 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSnappyElasticFramedTransport.java
@@ -20,6 +20,7 @@
 package org.apache.iotdb.rpc;
 
 import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
 import org.xerial.snappy.Snappy;
 
 import java.io.IOException;
@@ -41,13 +42,13 @@ public class TSnappyElasticFramedTransport extends 
TCompressedElasticFramedTrans
     }
 
     @Override
-    public TTransport getTransport(TTransport trans) {
+    public TTransport getTransport(TTransport trans) throws 
TTransportException {
       return new TSnappyElasticFramedTransport(
           trans, thriftDefaultBufferSize, thriftMaxFrameSize, copyBinary);
     }
   }
 
-  public TSnappyElasticFramedTransport(TTransport underlying) {
+  public TSnappyElasticFramedTransport(TTransport underlying) throws 
TTransportException {
     this(underlying, RpcUtils.THRIFT_DEFAULT_BUF_CAPACITY, 
RpcUtils.THRIFT_FRAME_MAX_SIZE, true);
   }
 
@@ -55,7 +56,8 @@ public class TSnappyElasticFramedTransport extends 
TCompressedElasticFramedTrans
       TTransport underlying,
       int thriftDefaultBufferSize,
       int thriftMaxFrameSize,
-      boolean copyBinary) {
+      boolean copyBinary)
+      throws TTransportException {
     super(underlying, thriftDefaultBufferSize, thriftMaxFrameSize, copyBinary);
   }
 
diff --git 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TimeoutChangeableTFastFramedTransport.java
 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TimeoutChangeableTFastFramedTransport.java
index 77c2bbf6b97..505e1a99405 100644
--- 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TimeoutChangeableTFastFramedTransport.java
+++ 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TimeoutChangeableTFastFramedTransport.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.rpc;
 
 import org.apache.thrift.transport.TSocket;
 import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
 import org.apache.thrift.transport.TTransportFactory;
 
 import java.net.SocketException;
@@ -31,7 +32,8 @@ public class TimeoutChangeableTFastFramedTransport extends 
TElasticFramedTranspo
   private final TSocket underlyingSocket;
 
   public TimeoutChangeableTFastFramedTransport(
-      TSocket underlying, int thriftDefaultBufferSize, int thriftMaxFrameSize, 
boolean copyBinary) {
+      TSocket underlying, int thriftDefaultBufferSize, int thriftMaxFrameSize, 
boolean copyBinary)
+      throws TTransportException {
     super(underlying, thriftDefaultBufferSize, thriftMaxFrameSize, copyBinary);
     this.underlyingSocket = underlying;
   }
@@ -65,7 +67,7 @@ public class TimeoutChangeableTFastFramedTransport extends 
TElasticFramedTranspo
     }
 
     @Override
-    public TTransport getTransport(TTransport trans) {
+    public TTransport getTransport(TTransport trans) throws 
TTransportException {
       if (trans instanceof TSocket) {
         return new TimeoutChangeableTFastFramedTransport(
             (TSocket) trans, thriftDefaultBufferSize, thriftMaxFrameSize, 
copyBinary);
diff --git 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TimeoutChangeableTSnappyFramedTransport.java
 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TimeoutChangeableTSnappyFramedTransport.java
index 168f52662aa..ecd807e38d6 100644
--- 
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TimeoutChangeableTSnappyFramedTransport.java
+++ 
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TimeoutChangeableTSnappyFramedTransport.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.rpc;
 
 import org.apache.thrift.transport.TSocket;
 import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
 import org.apache.thrift.transport.TTransportFactory;
 
 import java.net.SocketException;
@@ -31,7 +32,8 @@ public class TimeoutChangeableTSnappyFramedTransport extends 
TSnappyElasticFrame
   private final TSocket underlyingSocket;
 
   public TimeoutChangeableTSnappyFramedTransport(
-      TSocket underlying, int thriftDefaultBufferSize, int thriftMaxFrameSize, 
boolean copyBinary) {
+      TSocket underlying, int thriftDefaultBufferSize, int thriftMaxFrameSize, 
boolean copyBinary)
+      throws TTransportException {
     super(underlying, thriftDefaultBufferSize, thriftMaxFrameSize, copyBinary);
     this.underlyingSocket = underlying;
   }
@@ -65,7 +67,7 @@ public class TimeoutChangeableTSnappyFramedTransport extends 
TSnappyElasticFrame
     }
 
     @Override
-    public TTransport getTransport(TTransport trans) {
+    public TTransport getTransport(TTransport trans) throws 
TTransportException {
       if (trans instanceof TSocket) {
         return new TimeoutChangeableTSnappyFramedTransport(
             (TSocket) trans, thriftDefaultBufferSize, thriftMaxFrameSize, 
copyBinary);
diff --git 
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java
 
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java
index 2c24ff12b6d..8244cdf1bc5 100644
--- 
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java
+++ 
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java
@@ -60,6 +60,7 @@ import javax.net.ssl.X509TrustManager;
 
 import java.io.File;
 import java.io.FileNotFoundException;
+import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.nio.file.AccessDeniedException;
@@ -190,12 +191,21 @@ public class Utils {
 
   public static ByteBuffer serializeTSStatus(TSStatus status) throws 
TException {
     AutoScalingBufferWriteTransport byteBuffer =
-        new AutoScalingBufferWriteTransport(TEMP_BUFFER_SIZE);
+        createAutoScalingBufferWriteTransport(TEMP_BUFFER_SIZE);
     TCompactProtocol protocol = new TCompactProtocol(byteBuffer);
     status.write(protocol);
     return ByteBuffer.wrap(byteBuffer.getBuffer());
   }
 
+  private static AutoScalingBufferWriteTransport 
createAutoScalingBufferWriteTransport(
+      int initialCapacity) throws TException {
+    try {
+      return new AutoScalingBufferWriteTransport(initialCapacity);
+    } catch (IOException e) {
+      throw new TException(e);
+    }
+  }
+
   public static TSStatus deserializeFrom(ByteBuffer buffer) throws TException {
     TSStatus status = new TSStatus();
     TByteBuffer byteBuffer = new TByteBuffer(buffer);
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 dc1ca4e5473..cde5882ba59 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
@@ -19,12 +19,19 @@
 
 package org.apache.iotdb.commons.memory;
 
+import org.apache.iotdb.rpc.AutoResizingBufferMemoryControl;
+import org.apache.iotdb.rpc.AutoResizingBufferMemoryManager;
+
 public class MemoryConfig {
+  private static final String AUTO_RESIZING_BUFFER_MEMORY_MANAGER_NAME = 
"AutoResizingBuffer";
+  private static final String AUTO_RESIZING_BUFFER_MEMORY_BLOCK_NAME = 
"AutoResizingBufferBlock";
+  private static final int AUTO_RESIZING_BUFFER_MEMORY_RATIO_DENOMINATOR = 100;
+
   private final MemoryManager globalMemoryManager =
       new MemoryManager("GlobalMemoryManager", null, 
Runtime.getRuntime().totalMemory());
 
   private MemoryConfig() {
-    // singleton
+    initAutoResizingBufferMemoryControl();
   }
 
   public static MemoryManager global() {
@@ -40,4 +47,42 @@ public class MemoryConfig {
 
     private MemoryConfigHolder() {}
   }
+
+  private void initAutoResizingBufferMemoryControl() {
+    AutoResizingBufferMemoryManager.setMemoryControl(
+        new AutoResizingBufferMemoryControl() {
+          private IMemoryBlock autoResizingBufferMemoryBlock;
+
+          @Override
+          public synchronized boolean allocate(long sizeInBytes) {
+            return getAutoResizingBufferMemoryBlock().allocate(sizeInBytes);
+          }
+
+          @Override
+          public synchronized void release(long sizeInBytes) {
+            if (autoResizingBufferMemoryBlock != null
+                && !autoResizingBufferMemoryBlock.isReleased()) {
+              autoResizingBufferMemoryBlock.release(sizeInBytes);
+            }
+          }
+
+          private IMemoryBlock getAutoResizingBufferMemoryBlock() {
+            if (autoResizingBufferMemoryBlock == null
+                || autoResizingBufferMemoryBlock.isReleased()) {
+              long autoResizingBufferMemorySize =
+                  globalMemoryManager.getTotalMemorySizeInBytes()
+                      / AUTO_RESIZING_BUFFER_MEMORY_RATIO_DENOMINATOR;
+              MemoryManager autoResizingBufferMemoryManager =
+                  globalMemoryManager.getOrCreateMemoryManager(
+                      AUTO_RESIZING_BUFFER_MEMORY_MANAGER_NAME, 
autoResizingBufferMemorySize, true);
+              autoResizingBufferMemoryBlock =
+                  autoResizingBufferMemoryManager.exactAllocate(
+                      AUTO_RESIZING_BUFFER_MEMORY_BLOCK_NAME,
+                      autoResizingBufferMemorySize,
+                      MemoryBlockType.DYNAMIC);
+            }
+            return autoResizingBufferMemoryBlock;
+          }
+        });
+  }
 }

Reply via email to