This is an automated email from the ASF dual-hosted git repository. haonan pushed a commit to branch arraysize11 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d162c2669cd822902f9c274152d005f459009909 Author: Haonan <[email protected]> AuthorDate: Sat Jul 3 16:48:49 2021 +0800 [To rel/0.11] Primitive Array Manager v2 --- .../org/apache/iotdb/db/metadata/MManager.java | 16 +- .../iotdb/db/rescon/PrimitiveArrayManager.java | 219 ++++++++++----------- 2 files changed, 108 insertions(+), 127 deletions(-) diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java index 325a2e4..68f78fb 100644 --- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java +++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java @@ -103,8 +103,6 @@ public class MManager { private static final String DEBUG_MSG_1 = "%s: TimeSeries %s's tag info has been removed from tag inverted index "; private static final String PREVIOUS_CONDITION = "before deleting it, tag key is %s, tag value is %s, tlog offset is %d, contains key %b"; - private static final int UPDATE_SCHEMA_MAP_IN_ARRAYPOOL_THRESHOLD = 5000; - private static final Logger logger = LoggerFactory.getLogger(MManager.class); /** @@ -128,8 +126,6 @@ public class MManager { // data type -> number private Map<TSDataType, Integer> schemaDataTypeNumMap = new ConcurrentHashMap<>(); - // reported total series number - private long reportedDataTypeTotalNum; private AtomicLong totalSeriesNumber = new AtomicLong(); private boolean initialized; protected static IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); @@ -224,7 +220,6 @@ public class MManager { mtree = new MTree(); logger.error("Cannot read MTree from file, using an empty new one", e); } - reportedDataTypeTotalNum = 0L; initialized = true; } @@ -293,7 +288,6 @@ public class MManager { tagLogFile = null; } this.schemaDataTypeNumMap.clear(); - this.reportedDataTypeTotalNum = 0L; initialized = false; if (config.isEnableMTreeSnapshot() && timedCreateMTreeSnapshotThread != null) { timedCreateMTreeSnapshotThread.shutdownNow(); @@ -646,15 +640,7 @@ public class MManager { schemaDataTypeNumMap.put(TSDataType.INT64, schemaDataTypeNumMap.getOrDefault(TSDataType.INT64, 0) + num); - // total current DataType Total Num (twice of number of time series) - // used in primitive array manager - long currentDataTypeTotalNum = totalSeriesNumber.get() * 2; - - if (num > 0 && currentDataTypeTotalNum - reportedDataTypeTotalNum - >= UPDATE_SCHEMA_MAP_IN_ARRAYPOOL_THRESHOLD) { - PrimitiveArrayManager.updateSchemaDataTypeNum(schemaDataTypeNumMap, currentDataTypeTotalNum); - reportedDataTypeTotalNum = currentDataTypeTotalNum; - } + PrimitiveArrayManager.updateSchemaDataTypeNum(schemaDataTypeNumMap, totalSeriesNumber.get()); } /** diff --git a/server/src/main/java/org/apache/iotdb/db/rescon/PrimitiveArrayManager.java b/server/src/main/java/org/apache/iotdb/db/rescon/PrimitiveArrayManager.java index 20063fd..78e849e 100644 --- a/server/src/main/java/org/apache/iotdb/db/rescon/PrimitiveArrayManager.java +++ b/server/src/main/java/org/apache/iotdb/db/rescon/PrimitiveArrayManager.java @@ -18,41 +18,32 @@ */ package org.apache.iotdb.db.rescon; -import java.util.ArrayDeque; -import java.util.Arrays; -import java.util.EnumMap; -import java.util.Map; -import java.util.concurrent.atomic.AtomicLong; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.utils.Binary; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; -/** - * Manage all primitive data list in memory, including get and release operation. - */ -public class PrimitiveArrayManager { +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.EnumMap; +import java.util.Map; +import java.util.Map.Entry; +import java.util.concurrent.atomic.AtomicLong; - /** - * data type -> ArrayDeque<Array> - */ - private static final Map<TSDataType, ArrayDeque<Object>> bufferedArraysMap = new EnumMap<>( - TSDataType.class); +/** Manage all primitive data list in memory, including get and release operation. */ +public class PrimitiveArrayManager { - /** - * data type -> current number of buffered arrays - */ - private static final Map<TSDataType, Integer> bufferedArraysNumMap = new EnumMap<>( - TSDataType.class); + /** data type -> ArrayDeque<Array> */ + private static final Map<TSDataType, ArrayDeque<Object>> bufferedArraysMap = + new EnumMap<>(TSDataType.class); - /** - * data type -> ratio of data type in schema, which could be seen as recommended ratio - */ - private static final Map<TSDataType, Double> bufferedArraysNumRatio = new EnumMap<>( - TSDataType.class); + /** data type -> ratio of data type in schema, which could be seen as recommended ratio */ + private static final Map<TSDataType, Double> bufferedArraysNumRatio = + new EnumMap<>(TSDataType.class); private static final Logger logger = LoggerFactory.getLogger(PrimitiveArrayManager.class); @@ -60,21 +51,15 @@ public class PrimitiveArrayManager { public static final int ARRAY_SIZE = config.getPrimitiveArraySize(); - /** - * threshold total size of arrays for all data types - */ + /** threshold total size of arrays for all data types */ private static final double BUFFERED_ARRAY_SIZE_THRESHOLD = config.getAllocateMemoryForWrite() * config.getBufferedArraysMemoryProportion(); - /** - * total size of buffered arrays - */ - private static AtomicLong bufferedArraysRamSize = new AtomicLong(); + /** total size of buffered arrays */ + private static final AtomicLong bufferedArraysRamSize = new AtomicLong(); - /** - * total size of out of buffer arrays - */ - private static AtomicLong outOfBufferArraysRamSize = new AtomicLong(); + /** total size of out of buffer arrays */ + private static final AtomicLong outOfBufferArraysRamSize = new AtomicLong(); static { bufferedArraysMap.put(TSDataType.BOOLEAN, new ArrayDeque<>()); @@ -96,11 +81,12 @@ public class PrimitiveArrayManager { * @return an array */ public static Object getPrimitiveArraysByType(TSDataType dataType) { + long delta = (long) ARRAY_SIZE * dataType.getDataTypeSize(); + // check memory of buffered array, if already full, generate OOB - if (bufferedArraysRamSize.get() + ARRAY_SIZE * dataType.getDataTypeSize() - > BUFFERED_ARRAY_SIZE_THRESHOLD) { + if (bufferedArraysRamSize.get() + delta > BUFFERED_ARRAY_SIZE_THRESHOLD) { // return an out of buffer array - outOfBufferArraysRamSize.addAndGet((long) ARRAY_SIZE * dataType.getDataTypeSize()); + outOfBufferArraysRamSize.addAndGet(delta); return createPrimitiveArray(dataType); } @@ -110,11 +96,10 @@ public class PrimitiveArrayManager { if (dataArray != null) { return dataArray; } - // no buffered array, create one - bufferedArraysNumMap.put(dataType, bufferedArraysNumMap.getOrDefault(dataType, 0) + 1); - bufferedArraysRamSize.addAndGet((long) ARRAY_SIZE * dataType.getDataTypeSize()); } + // no buffered array, create one + bufferedArraysRamSize.addAndGet(delta); return createPrimitiveArray(dataType); } @@ -150,10 +135,10 @@ public class PrimitiveArrayManager { * Get primitive data lists according to data type and size, only for TVList's sorting * * @param dataType data type - * @param size needed capacity + * @param size needed capacity * @return an array of primitive data arrays */ - public static synchronized Object createDataListsByType(TSDataType dataType, int size) { + public static Object createDataListsByType(TSDataType dataType, int size) { int arrayNumber = (int) Math.ceil((float) size / (float) ARRAY_SIZE); switch (dataType) { case BOOLEAN: @@ -200,94 +185,104 @@ public class PrimitiveArrayManager { /** * This method is called when bringing back data array * - * @param dataArray data array + * @param releasingArray data array to be released */ - public static void release(Object dataArray) { - TSDataType dataType; - if (dataArray instanceof boolean[]) { - dataType = TSDataType.BOOLEAN; - } else if (dataArray instanceof int[]) { - dataType = TSDataType.INT32; - } else if (dataArray instanceof long[]) { - dataType = TSDataType.INT64; - } else if (dataArray instanceof float[]) { - dataType = TSDataType.FLOAT; - } else if (dataArray instanceof double[]) { - dataType = TSDataType.DOUBLE; - } else if (dataArray instanceof Binary[]) { - Arrays.fill((Binary[]) dataArray, null); - dataType = TSDataType.TEXT; + public static void release(Object releasingArray) { + TSDataType releasingType; + if (releasingArray instanceof boolean[]) { + releasingType = TSDataType.BOOLEAN; + } else if (releasingArray instanceof int[]) { + releasingType = TSDataType.INT32; + } else if (releasingArray instanceof long[]) { + releasingType = TSDataType.INT64; + } else if (releasingArray instanceof float[]) { + releasingType = TSDataType.FLOAT; + } else if (releasingArray instanceof double[]) { + releasingType = TSDataType.DOUBLE; + } else if (releasingArray instanceof Binary[]) { + Arrays.fill((Binary[]) releasingArray, null); + releasingType = TSDataType.TEXT; } else { throw new UnSupportedDataTypeException("Unknown data array type"); } - // Check out of buffer array num - if (outOfBufferArraysRamSize.get() > 0 && isCurrentDataTypeExceeded(dataType)) { - // release an out of buffer array - bringBackOOBArray(dataType, ARRAY_SIZE); - } else if (outOfBufferArraysRamSize.get() > 0 && !isCurrentDataTypeExceeded(dataType)) { - // if the ratio of buffered arrays of this data type does not exceed the schema ratio, - // choose one replaced array who has larger ratio than schema recommended ratio - TSDataType replacedDataType = null; - for (Map.Entry<TSDataType, Integer> entry : bufferedArraysNumMap.entrySet()) { - if (isCurrentDataTypeExceeded(entry.getKey())) { - replacedDataType = entry.getKey(); - // bring back the replaced array as OOB array - bringBackOOBArray(replacedDataType, ARRAY_SIZE); - break; - } - } - if (replacedDataType != null) { - // if we find a replaced array, bring back the original array as a buffered array - if (logger.isDebugEnabled()) { - logger.debug( - "The ratio of {} in buffered array has not reached the schema ratio. Replaced by {}", - dataType, replacedDataType); + if (outOfBufferArraysRamSize.get() <= 0) { + // if there is no out of buffer array, bring back as buffered array directly + putBackBufferedArray(releasingType, releasingArray); + } else { + // if the system has out of buffer array, we need to release some memory + if (!isCurrentDataTypeExceeded(releasingType)) { + // if the buffered array of the releasingType is less than expected + // choose an array of redundantDataType to release and try to buffer the array of + // releasingType + for (Entry<TSDataType, ArrayDeque<Object>> entry : bufferedArraysMap.entrySet()) { + TSDataType dataType = entry.getKey(); + if (isCurrentDataTypeExceeded(dataType)) { + // if we find a replaced array, bring back the original array as a buffered array + if (logger.isDebugEnabled()) { + logger.debug( + "The ratio of {} in buffered array has not reached the schema ratio. discard a redundant array of {}", + releasingType, + dataType); + } + // bring back the replaced array as OOB array + replaceBufferedArray(releasingType, releasingArray, dataType); + break; + } } - bringBackBufferedArray(dataType, dataArray); - } else { - // or else bring back the original array as OOB array - bringBackOOBArray(dataType, ARRAY_SIZE); } - } else { - // if there is no out of buffer array, bring back as buffered array directly - bringBackBufferedArray(dataType, dataArray); + + releaseOutOfBuffer(releasingType); } } /** * Bring back a buffered array * - * @param dataType data type + * @param dataType data type * @param dataArray data array */ - private static void bringBackBufferedArray(TSDataType dataType, Object dataArray) { + private static void putBackBufferedArray(TSDataType dataType, Object dataArray) { synchronized (bufferedArraysMap.get(dataType)) { bufferedArraysMap.get(dataType).add(dataArray); - bufferedArraysNumMap.put(dataType, bufferedArraysNumMap.getOrDefault(dataType, 0) + 1); } - bufferedArraysRamSize.addAndGet((long) -ARRAY_SIZE * dataType.getDataTypeSize()); } - /** - * Bring back out of buffered array - * - * @param dataType data type - * @param size capacity - */ - private static void bringBackOOBArray(TSDataType dataType, int size) { - outOfBufferArraysRamSize.addAndGet((long) -size * dataType.getDataTypeSize()); + private static void replaceBufferedArray( + TSDataType releasingType, Object releasingArray, TSDataType redundantType) { + synchronized (bufferedArraysMap.get(redundantType)) { + if (bufferedArraysMap.get(redundantType).poll() != null) { + bufferedArraysRamSize.addAndGet((long) -ARRAY_SIZE * redundantType.getDataTypeSize()); + } + } + + if (bufferedArraysRamSize.get() + (long) ARRAY_SIZE * releasingType.getDataTypeSize() + < BUFFERED_ARRAY_SIZE_THRESHOLD) { + ArrayDeque<Object> releasingArrays = bufferedArraysMap.get(releasingType); + synchronized (releasingArrays) { + releasingArrays.add(releasingArray); + } + bufferedArraysRamSize.addAndGet((long) ARRAY_SIZE * releasingType.getDataTypeSize()); + } + } + + private static void releaseOutOfBuffer(TSDataType dataType) { + outOfBufferArraysRamSize.getAndUpdate( + l -> Math.max(0, l - (long) ARRAY_SIZE * dataType.getDataTypeSize())); } /** - * @param schemaDataTypeNumMap schema DataType Num Map (for each series, increase a long and a specific type) - * @param total current DataType Total Num (twice of number of time series) + * @param schemaDataTypeNumMap schema DataType Num Map (for each series, increase a long and a + * specific type) + * @param totalSeries total time series number */ - public static void updateSchemaDataTypeNum(Map<TSDataType, Integer> schemaDataTypeNumMap, - long total) { + public static void updateSchemaDataTypeNum( + Map<TSDataType, Integer> schemaDataTypeNumMap, long totalSeries) { for (Map.Entry<TSDataType, Integer> entry : schemaDataTypeNumMap.entrySet()) { TSDataType dataType = entry.getKey(); - bufferedArraysNumRatio.put(dataType, (double) schemaDataTypeNumMap.get(dataType) / total); + // one time series has 2 columns (time column + value column) + bufferedArraysNumRatio.put( + dataType, (double) schemaDataTypeNumMap.get(dataType) / (totalSeries * 2)); } } @@ -299,13 +294,14 @@ public class PrimitiveArrayManager { * @return true if the buffered array ratio exceeds the recommend ratio */ private static boolean isCurrentDataTypeExceeded(TSDataType dataType) { - int total = 0; - for (int num : bufferedArraysNumMap.values()) { - total += num; + long total = 0; + for (ArrayDeque<Object> value : bufferedArraysMap.values()) { + total += value.size(); } - return total != 0 && - ((double) bufferedArraysNumMap.getOrDefault(dataType, 0) / total > - bufferedArraysNumRatio.getOrDefault(dataType, 0.0)); + long arrayNumInBuffer = + bufferedArraysMap.containsKey(dataType) ? bufferedArraysMap.get(dataType).size() : 0; + return total != 0 + && ((double) arrayNumInBuffer / total > bufferedArraysNumRatio.getOrDefault(dataType, 0.0)); } public static void close() { @@ -313,7 +309,6 @@ public class PrimitiveArrayManager { dataListQueue.clear(); } - bufferedArraysNumMap.clear(); bufferedArraysNumRatio.clear(); bufferedArraysRamSize.set(0);
