Caideyipi commented on code in PR #18014:
URL: https://github.com/apache/iotdb/pull/18014#discussion_r3464410711
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java:
##########
@@ -46,13 +48,66 @@
import java.time.ZoneId;
import java.util.Optional;
+import java.util.concurrent.Semaphore;
public class LoadTsFileDataTypeConverter {
private static final Logger LOGGER =
LoggerFactory.getLogger(LoadTsFileDataTypeConverter.class);
private static final SessionManager SESSION_MANAGER =
SessionManager.getInstance();
+ private static Semaphore getTabletConversionSemaphore() {
+ return TabletConversionSemaphoreHolder.INSTANCE;
+ }
+
+ private static int getTabletConversionPermitCount() {
+ final int configuredThreadCount =
+ Math.max(
+ 1,
+
IoTDBDescriptor.getInstance().getConfig().getLoadTsFileTabletConversionThreadCount());
+ if (!PipeConfig.getInstance().getPipeMemoryManagementEnabled()) {
+ return configuredThreadCount;
+ }
+ final long memorySafePermitCount =
+ getAllowedPipeTabletMemorySizeInBytes() /
estimatePipeTabletMemorySizePerConversion();
+ return (int) Math.max(1, Math.min((long) configuredThreadCount,
memorySafePermitCount));
+ }
+
+ private static long estimatePipeTabletMemorySizePerConversion() {
+ final PipeConfig pipeConfig = PipeConfig.getInstance();
+ final long tabletSize =
+ Math.max(
+ 1L,
IoTDBDescriptor.getInstance().getConfig().getPipeDataStructureTabletSizeInBytes());
+ final long maxReaderChunkSize = Math.max(0L,
pipeConfig.getPipeMaxReaderChunkSize());
+ final long tableSize =
+ Math.max(
+ 1L,
+ Math.min(tabletSize,
IoTDBDescriptor.getInstance().getConfig().getTargetChunkSize()));
+
+ final long treeParserMemorySize = 2L * tabletSize + maxReaderChunkSize;
+ final long tableParserMemorySize = 2L * tabletSize + 2L * tableSize +
maxReaderChunkSize;
Review Comment:
Added comments explaining the per-conversion memory estimate and why we use
the larger tree/table parser working set.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java:
##########
@@ -46,13 +48,66 @@
import java.time.ZoneId;
import java.util.Optional;
+import java.util.concurrent.Semaphore;
public class LoadTsFileDataTypeConverter {
private static final Logger LOGGER =
LoggerFactory.getLogger(LoadTsFileDataTypeConverter.class);
private static final SessionManager SESSION_MANAGER =
SessionManager.getInstance();
+ private static Semaphore getTabletConversionSemaphore() {
+ return TabletConversionSemaphoreHolder.INSTANCE;
+ }
+
+ private static int getTabletConversionPermitCount() {
+ final int configuredThreadCount =
+ Math.max(
+ 1,
+
IoTDBDescriptor.getInstance().getConfig().getLoadTsFileTabletConversionThreadCount());
+ if (!PipeConfig.getInstance().getPipeMemoryManagementEnabled()) {
+ return configuredThreadCount;
+ }
+ final long memorySafePermitCount =
+ getAllowedPipeTabletMemorySizeInBytes() /
estimatePipeTabletMemorySizePerConversion();
+ return (int) Math.max(1, Math.min((long) configuredThreadCount,
memorySafePermitCount));
+ }
+
+ private static long estimatePipeTabletMemorySizePerConversion() {
+ final PipeConfig pipeConfig = PipeConfig.getInstance();
+ final long tabletSize =
+ Math.max(
+ 1L,
IoTDBDescriptor.getInstance().getConfig().getPipeDataStructureTabletSizeInBytes());
+ final long maxReaderChunkSize = Math.max(0L,
pipeConfig.getPipeMaxReaderChunkSize());
+ final long tableSize =
+ Math.max(
+ 1L,
+ Math.min(tabletSize,
IoTDBDescriptor.getInstance().getConfig().getTargetChunkSize()));
+
+ final long treeParserMemorySize = 2L * tabletSize + maxReaderChunkSize;
+ final long tableParserMemorySize = 2L * tabletSize + 2L * tableSize +
maxReaderChunkSize;
+ return Math.max(1L, Math.max(treeParserMemorySize, tableParserMemorySize));
+ }
+
+ private static long getAllowedPipeTabletMemorySizeInBytes() {
+ final PipeConfig pipeConfig = PipeConfig.getInstance();
+ return (long)
+
((pipeConfig.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold()
+ +
pipeConfig.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() / 2)
+ *
PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes());
+ }
+
+ private static Optional<TSStatus> getInterruptedConversionStatus(final
InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return Optional.of(
+ new
TSStatus(TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode())
+ .setMessage("Interrupted while waiting for tablet conversion slot:
" + e.getMessage()));
Review Comment:
Done. Moved the interrupted tablet conversion slot message to
StorageEngineMessages and added both en/zh i18n entries.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
