#9922

Committed 25 Aug 2023 08:15AM UTC coverage: 47.788% (-0.01%) from 47.802%

Build # #9922

Build Type

push

travis_ci

Committed by

web-flow

Commit Message

[IOTDB-6126] Pipe: Added PipeHeartbeatEvent to trigger delay reporting, batch transferring and async retrying (#10940) (#10954)

Co-authored-by: Steve Yurong Su <rong@apache.org>
(cherry picked from commit c4b72fc76)

Run Details

92 of 92 new or added lines in 16 files covered. (100.0%)

80029 of 167468 relevant lines covered (47.79%)

0.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.46

/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/SystemInfo.java

/*
 * 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.storageengine.rescon.memory;

import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.WriteProcessRejectException;
import org.apache.iotdb.db.storageengine.dataregion.DataRegionInfo;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionFileCountExceededException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionMemoryNotEnoughException;
import org.apache.iotdb.db.storageengine.dataregion.flush.FlushManager;
import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class SystemInfo {

  private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
  private static final Logger logger = LoggerFactory.getLogger(SystemInfo.class);

  private long totalStorageGroupMemCost = 0L;
  private volatile boolean rejected = false;

  private long memorySizeForMemtable;
  private long memorySizeForCompaction;
  private Map<DataRegionInfo, Long> reportedStorageGroupMemCostMap = new HashMap<>();

  private long flushingMemTablesCost = 0L;
  private AtomicLong compactionMemoryCost = new AtomicLong(0L);

  private AtomicInteger compactionFileNumCost = new AtomicInteger(0);

  private int totalFileLimitForCrossTask = config.getTotalFileLimitForCrossTask();

  private ExecutorService flushTaskSubmitThreadPool =
      IoTDBThreadPoolFactory.newSingleThreadExecutor(ThreadName.FLUSH_TASK_SUBMIT.getName());
  private double FLUSH_THERSHOLD = memorySizeForMemtable * config.getFlushProportion();
  private double REJECT_THERSHOLD = memorySizeForMemtable * config.getRejectProportion();

  private volatile boolean isEncodingFasterThanIo = true;

  private SystemInfo() {
    allocateWriteMemory();
  }

  /**
   * Report current mem cost of database to system. Called when the memory of database newly
   * accumulates to IoTDBConfig.getStorageGroupSizeReportThreshold()
   *
   * @param dataRegionInfo database
   * @throws WriteProcessRejectException
   */
  public synchronized boolean reportStorageGroupStatus(
      DataRegionInfo dataRegionInfo, TsFileProcessor tsFileProcessor)
      throws WriteProcessRejectException {
    long currentDataRegionMemCost = dataRegionInfo.getMemCost();
    long delta =
        currentDataRegionMemCost - reportedStorageGroupMemCostMap.getOrDefault(dataRegionInfo, 0L);
    totalStorageGroupMemCost += delta;
    if (logger.isDebugEnabled()) {
      logger.debug(
          "Report database Status to the system. " + "After adding {}, current sg mem cost is {}.",
          delta,
          totalStorageGroupMemCost);
    }
    reportedStorageGroupMemCostMap.put(dataRegionInfo, currentDataRegionMemCost);
    dataRegionInfo.setLastReportedSize(currentDataRegionMemCost);
    if (totalStorageGroupMemCost < FLUSH_THERSHOLD) {
      return true;
    } else if (totalStorageGroupMemCost >= FLUSH_THERSHOLD
        && totalStorageGroupMemCost < REJECT_THERSHOLD) {
      logger.debug(
          "The total database mem costs are too large, call for flushing. "
              + "Current sg cost is {}",
          totalStorageGroupMemCost);
      chooseMemTablesToMarkFlush(tsFileProcessor);
      return true;
    } else {
      logger.info(
          "Change system to reject status. Triggered by: logical SG ({}), mem cost delta ({}), totalSgMemCost ({}), REJECT_THERSHOLD ({})",
          dataRegionInfo.getDataRegion().getDatabaseName(),
          delta,
          totalStorageGroupMemCost,
          REJECT_THERSHOLD);
      rejected = true;
      if (chooseMemTablesToMarkFlush(tsFileProcessor)) {
        if (totalStorageGroupMemCost < memorySizeForMemtable) {
          return true;
        } else {
          throw new WriteProcessRejectException(
              "Total database MemCost "
                  + totalStorageGroupMemCost
                  + " is over than memorySizeForWriting "
                  + memorySizeForMemtable);
        }
      } else {
        return false;
      }
    }
  }

  /**
   * Report resetting the mem cost of sg to system. It will be called after flushing, closing and
   * failed to insert
   *
   * @param dataRegionInfo database
   */
  public synchronized void resetStorageGroupStatus(DataRegionInfo dataRegionInfo) {
    long currentDataRegionMemCost = dataRegionInfo.getMemCost();
    long delta = 0;
    if (reportedStorageGroupMemCostMap.containsKey(dataRegionInfo)) {
      delta = reportedStorageGroupMemCostMap.get(dataRegionInfo) - currentDataRegionMemCost;
      this.totalStorageGroupMemCost -= delta;
      dataRegionInfo.setLastReportedSize(currentDataRegionMemCost);
      // report after reset sg status, because slow write may not reach the report threshold
      dataRegionInfo.setNeedToReportToSystem(true);
      reportedStorageGroupMemCostMap.put(dataRegionInfo, currentDataRegionMemCost);
    }

    if (totalStorageGroupMemCost >= FLUSH_THERSHOLD
        && totalStorageGroupMemCost < REJECT_THERSHOLD) {
      logger.debug(
          "SG ({}) released memory (delta: {}) but still exceeding flush proportion (totalSgMemCost: {}), call flush.",
          dataRegionInfo.getDataRegion().getDatabaseName(),
          delta,
          totalStorageGroupMemCost);
      if (rejected) {
        logger.info(
            "SG ({}) released memory (delta: {}), set system to normal status (totalSgMemCost: {}).",
            dataRegionInfo.getDataRegion().getDatabaseName(),
            delta,
            totalStorageGroupMemCost);
      }
      logCurrentTotalSGMemory();
      rejected = false;
    } else if (totalStorageGroupMemCost >= REJECT_THERSHOLD) {
      logger.warn(
          "SG ({}) released memory (delta: {}), but system is still in reject status (totalSgMemCost: {}).",
          dataRegionInfo.getDataRegion().getDatabaseName(),
          delta,
          totalStorageGroupMemCost);
      logCurrentTotalSGMemory();
      rejected = true;
    } else {
      logger.debug(
          "SG ({}) released memory (delta: {}), system is in normal status (totalSgMemCost: {}).",
          dataRegionInfo.getDataRegion().getDatabaseName(),
          delta,
          totalStorageGroupMemCost);
      logCurrentTotalSGMemory();
      rejected = false;
    }
  }

  public synchronized void addFlushingMemTableCost(long flushingMemTableCost) {
    this.flushingMemTablesCost += flushingMemTableCost;
  }

  public synchronized void resetFlushingMemTableCost(long flushingMemTableCost) {
    this.flushingMemTablesCost -= flushingMemTableCost;
  }

  public void addCompactionFileNum(int fileNum, long timeOutInSecond)
      throws InterruptedException, CompactionFileCountExceededException {
    if (fileNum > totalFileLimitForCrossTask) {
      // source file num is greater than the max file num for compaction
      throw new CompactionFileCountExceededException(
          String.format(
              "Required file num %d is greater than the max file num %d for compaction.",
              fileNum, totalFileLimitForCrossTask));
    }
    long startTime = System.currentTimeMillis();
    int originFileNum = this.compactionFileNumCost.get();
    while (originFileNum + fileNum > totalFileLimitForCrossTask
        || !compactionFileNumCost.compareAndSet(originFileNum, originFileNum + fileNum)) {
      if (System.currentTimeMillis() - startTime >= timeOutInSecond * 1000L) {
        throw new CompactionFileCountExceededException(
            String.format(
                "Failed to allocate %d files for compaction after %d seconds, max file num for compaction module is %d, %d files is used.",
                fileNum, timeOutInSecond, totalFileLimitForCrossTask, originFileNum));
      }
      Thread.sleep(100);
      originFileNum = this.compactionFileNumCost.get();
    }
  }

  public void addCompactionMemoryCost(long memoryCost, long timeOutInSecond)
      throws InterruptedException, CompactionMemoryNotEnoughException {
    if (!config.isEnableCompactionMemControl()) {
      return;
    }
    if (memoryCost > memorySizeForCompaction) {
      // required memory cost is greater than the total memory budget for compaction
      throw new CompactionMemoryNotEnoughException(
          String.format(
              "Required memory cost %d bytes is greater than "
                  + "the total memory budget for compaction %d bytes",
              memoryCost, memorySizeForCompaction));
    }
    long startTime = System.currentTimeMillis();
    long originSize = this.compactionMemoryCost.get();
    while (originSize + memoryCost > memorySizeForCompaction
        || !compactionMemoryCost.compareAndSet(originSize, originSize + memoryCost)) {
      if (System.currentTimeMillis() - startTime >= timeOutInSecond * 1000L) {
        throw new CompactionMemoryNotEnoughException(
            String.format(
                "Failed to allocate %d bytes memory for compaction after %d seconds, "
                    + "total memory budget for compaction module is %d bytes, %d bytes is used",
                memoryCost, timeOutInSecond, memorySizeForCompaction, originSize));
      }
      Thread.sleep(100);
      originSize = this.compactionMemoryCost.get();
    }
  }

  public synchronized void resetCompactionMemoryCost(long compactionMemoryCost) {
    this.compactionMemoryCost.addAndGet(-compactionMemoryCost);
  }

  public synchronized void decreaseCompactionFileNumCost(int fileNum) {
    this.compactionFileNumCost.addAndGet(-fileNum);
  }

  public long getMemorySizeForCompaction() {
    if (config.isEnableMemControl()) {
      return memorySizeForCompaction;
    } else {
      return Long.MAX_VALUE;
    }
  }

  public void allocateWriteMemory() {
    memorySizeForMemtable =
        (long)
            (config.getAllocateMemoryForStorageEngine() * config.getWriteProportionForMemtable());
    memorySizeForCompaction =
        (long) (config.getAllocateMemoryForStorageEngine() * config.getCompactionProportion());
    FLUSH_THERSHOLD = memorySizeForMemtable * config.getFlushProportion();
    REJECT_THERSHOLD = memorySizeForMemtable * config.getRejectProportion();
  }

  @TestOnly
  public void setMemorySizeForCompaction(long size) {
    memorySizeForCompaction = size;
  }

  @TestOnly
  public void setTotalFileLimitForCrossTask(int totalFileLimitForCrossTask) {
    this.totalFileLimitForCrossTask = totalFileLimitForCrossTask;
  }

  @TestOnly
  public int getTotalFileLimitForCrossTask() {
    return totalFileLimitForCrossTask;
  }

  @TestOnly
  public AtomicLong getCompactionMemoryCost() {
    return compactionMemoryCost;
  }

  @TestOnly
  public AtomicInteger getCompactionFileNumCost() {
    return compactionFileNumCost;
  }

  private void logCurrentTotalSGMemory() {
    logger.debug("Current Sg cost is {}", totalStorageGroupMemCost);
  }

  /**
   * Order all working memtables in system by memory cost of actual data points in memtable. Mark
   * the top K TSPs as to be flushed, so that after flushing the K TSPs, the memory cost should be
   * less than FLUSH_THRESHOLD
   */
  private boolean chooseMemTablesToMarkFlush(TsFileProcessor currentTsFileProcessor) {
    // If invoke flush by replaying logs, do not flush now!
    if (reportedStorageGroupMemCostMap.size() == 0) {
      return false;
    }
    PriorityQueue<TsFileProcessor> allTsFileProcessors =
        new PriorityQueue<>(
            (o1, o2) -> Long.compare(o2.getWorkMemTableRamCost(), o1.getWorkMemTableRamCost()));
    for (DataRegionInfo dataRegionInfo : reportedStorageGroupMemCostMap.keySet()) {
      allTsFileProcessors.addAll(dataRegionInfo.getAllReportedTsp());
    }
    boolean isCurrentTsFileProcessorSelected = false;
    long memCost = 0;
    long activeMemSize = totalStorageGroupMemCost - flushingMemTablesCost;
    while (activeMemSize - memCost > FLUSH_THERSHOLD) {
      if (allTsFileProcessors.isEmpty()
          || allTsFileProcessors.peek().getWorkMemTableRamCost() == 0) {
        return false;
      }
      TsFileProcessor selectedTsFileProcessor = allTsFileProcessors.peek();
      memCost += selectedTsFileProcessor.getWorkMemTableRamCost();
      selectedTsFileProcessor.setWorkMemTableShouldFlush();
      flushTaskSubmitThreadPool.submit(
          () -> {
            selectedTsFileProcessor.submitAFlushTask();
          });
      if (selectedTsFileProcessor == currentTsFileProcessor) {
        isCurrentTsFileProcessorSelected = true;
      }
      allTsFileProcessors.poll();
    }
    return isCurrentTsFileProcessorSelected;
  }

  public boolean isRejected() {
    return rejected;
  }

  public void setEncodingFasterThanIo(boolean isEncodingFasterThanIo) {
    this.isEncodingFasterThanIo = isEncodingFasterThanIo;
  }

  public boolean isEncodingFasterThanIo() {
    return isEncodingFasterThanIo;
  }

  public void close() {
    reportedStorageGroupMemCostMap.clear();
    totalStorageGroupMemCost = 0;
    rejected = false;
  }

  public static SystemInfo getInstance() {
    return InstanceHolder.instance;
  }

  private static class InstanceHolder {

    private InstanceHolder() {}

    private static SystemInfo instance = new SystemInfo();
  }

  public synchronized void applyTemporaryMemoryForFlushing(long estimatedTemporaryMemSize) {
    memorySizeForMemtable -= estimatedTemporaryMemSize;
    FLUSH_THERSHOLD = memorySizeForMemtable * config.getFlushProportion();
    REJECT_THERSHOLD = memorySizeForMemtable * config.getRejectProportion();
  }

  public synchronized void releaseTemporaryMemoryForFlushing(long estimatedTemporaryMemSize) {
    memorySizeForMemtable += estimatedTemporaryMemSize;
    FLUSH_THERSHOLD = memorySizeForMemtable * config.getFlushProportion();
    REJECT_THERSHOLD = memorySizeForMemtable * config.getRejectProportion();
  }

  public long getTotalMemTableSize() {
    return totalStorageGroupMemCost;
  }

  public double getFlushThershold() {
    return FLUSH_THERSHOLD;
  }

  public double getRejectThershold() {
    return REJECT_THERSHOLD;
  }

  public int flushingMemTableNum() {
    return FlushManager.getInstance().getNumberOfWorkingTasks();
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one
3	* or more contributor license agreements. See the NOTICE file
4	* distributed with this work for additional information
5	* regarding copyright ownership. The ASF licenses this file
6	* to you under the Apache License, Version 2.0 (the
7	* "License"); you may not use this file except in compliance
8	* with the License. You may obtain a copy of the License at
9	*
10	* http://www.apache.org/licenses/LICENSE-2.0
11	*
12	* Unless required by applicable law or agreed to in writing,
13	* software distributed under the License is distributed on an
14	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15	* KIND, either express or implied. See the License for the
16	* specific language governing permissions and limitations
17	* under the License.
18	*/
19
20	package org.apache.iotdb.db.storageengine.rescon.memory;
21
22	import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
23	import org.apache.iotdb.commons.concurrent.ThreadName;
24	import org.apache.iotdb.commons.utils.TestOnly;
25	import org.apache.iotdb.db.conf.IoTDBConfig;
26	import org.apache.iotdb.db.conf.IoTDBDescriptor;
27	import org.apache.iotdb.db.exception.WriteProcessRejectException;
28	import org.apache.iotdb.db.storageengine.dataregion.DataRegionInfo;
29	import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionFileCountExceededException;
30	import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionMemoryNotEnoughException;
31	import org.apache.iotdb.db.storageengine.dataregion.flush.FlushManager;
32	import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
33
34	import org.slf4j.Logger;
35	import org.slf4j.LoggerFactory;
36
37	import java.util.HashMap;
38	import java.util.Map;
39	import java.util.PriorityQueue;
40	import java.util.concurrent.ExecutorService;
41	import java.util.concurrent.atomic.AtomicInteger;
42	import java.util.concurrent.atomic.AtomicLong;
43
44	public class SystemInfo {
45
46	private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();	1✔
47	private static final Logger logger = LoggerFactory.getLogger(SystemInfo.class);	1✔
48
49	private long totalStorageGroupMemCost = 0L;	1✔
50	private volatile boolean rejected = false;	1✔
51
52	private long memorySizeForMemtable;
53	private long memorySizeForCompaction;
54	private Map<DataRegionInfo, Long> reportedStorageGroupMemCostMap = new HashMap<>();	1✔
55
56	private long flushingMemTablesCost = 0L;	1✔
57	private AtomicLong compactionMemoryCost = new AtomicLong(0L);	1✔
58
59	private AtomicInteger compactionFileNumCost = new AtomicInteger(0);	1✔
60
61	private int totalFileLimitForCrossTask = config.getTotalFileLimitForCrossTask();	1✔
62
63	private ExecutorService flushTaskSubmitThreadPool =	1✔
64	IoTDBThreadPoolFactory.newSingleThreadExecutor(ThreadName.FLUSH_TASK_SUBMIT.getName());	1✔
65	private double FLUSH_THERSHOLD = memorySizeForMemtable * config.getFlushProportion();	1✔
66	private double REJECT_THERSHOLD = memorySizeForMemtable * config.getRejectProportion();	1✔
67
68	private volatile boolean isEncodingFasterThanIo = true;	1✔
69
70	private SystemInfo() {	1✔
71	allocateWriteMemory();	1✔
72	}	1✔
73
74	/**
75	* Report current mem cost of database to system. Called when the memory of database newly
76	* accumulates to IoTDBConfig.getStorageGroupSizeReportThreshold()
77	*
78	* @param dataRegionInfo database
79	* @throws WriteProcessRejectException
80	*/
81	public synchronized boolean reportStorageGroupStatus(
82	DataRegionInfo dataRegionInfo, TsFileProcessor tsFileProcessor)
83	throws WriteProcessRejectException {
84	long currentDataRegionMemCost = dataRegionInfo.getMemCost();	1✔
85	long delta =	1✔
86	currentDataRegionMemCost - reportedStorageGroupMemCostMap.getOrDefault(dataRegionInfo, 0L);	1✔
87	totalStorageGroupMemCost += delta;	1✔
88	if (logger.isDebugEnabled()) {	1✔
89	logger.debug(	×
90	"Report database Status to the system. " + "After adding {}, current sg mem cost is {}.",
91	delta,	×
92	totalStorageGroupMemCost);	×
93	}
94	reportedStorageGroupMemCostMap.put(dataRegionInfo, currentDataRegionMemCost);	1✔
95	dataRegionInfo.setLastReportedSize(currentDataRegionMemCost);	1✔
96	if (totalStorageGroupMemCost < FLUSH_THERSHOLD) {	1✔
97	return true;	1✔
98	} else if (totalStorageGroupMemCost >= FLUSH_THERSHOLD	×
99	&& totalStorageGroupMemCost < REJECT_THERSHOLD) {
100	logger.debug(	×
101	"The total database mem costs are too large, call for flushing. "
102	+ "Current sg cost is {}",
103	totalStorageGroupMemCost);	×
104	chooseMemTablesToMarkFlush(tsFileProcessor);	×
105	return true;	×
106	} else {
107	logger.info(	×
108	"Change system to reject status. Triggered by: logical SG ({}), mem cost delta ({}), totalSgMemCost ({}), REJECT_THERSHOLD ({})",
109	dataRegionInfo.getDataRegion().getDatabaseName(),	×
110	delta,	×
111	totalStorageGroupMemCost,	×
112	REJECT_THERSHOLD);	×
113	rejected = true;	×
114	if (chooseMemTablesToMarkFlush(tsFileProcessor)) {	×
115	if (totalStorageGroupMemCost < memorySizeForMemtable) {	×
116	return true;	×
117	} else {
118	throw new WriteProcessRejectException(	×
119	"Total database MemCost "
120	+ totalStorageGroupMemCost
121	+ " is over than memorySizeForWriting "
122	+ memorySizeForMemtable);
123	}
124	} else {
125	return false;	×
126	}
127	}
128	}
129
130	/**
131	* Report resetting the mem cost of sg to system. It will be called after flushing, closing and
132	* failed to insert
133	*
134	* @param dataRegionInfo database
135	*/
136	public synchronized void resetStorageGroupStatus(DataRegionInfo dataRegionInfo) {
137	long currentDataRegionMemCost = dataRegionInfo.getMemCost();	1✔
138	long delta = 0;	1✔
139	if (reportedStorageGroupMemCostMap.containsKey(dataRegionInfo)) {	1✔
140	delta = reportedStorageGroupMemCostMap.get(dataRegionInfo) - currentDataRegionMemCost;	1✔
141	this.totalStorageGroupMemCost -= delta;	1✔
142	dataRegionInfo.setLastReportedSize(currentDataRegionMemCost);	1✔
143	// report after reset sg status, because slow write may not reach the report threshold
144	dataRegionInfo.setNeedToReportToSystem(true);	1✔
145	reportedStorageGroupMemCostMap.put(dataRegionInfo, currentDataRegionMemCost);	1✔
146	}
147
148	if (totalStorageGroupMemCost >= FLUSH_THERSHOLD	1✔
149	&& totalStorageGroupMemCost < REJECT_THERSHOLD) {
150	logger.debug(	×
151	"SG ({}) released memory (delta: {}) but still exceeding flush proportion (totalSgMemCost: {}), call flush.",
152	dataRegionInfo.getDataRegion().getDatabaseName(),	×
153	delta,	×
154	totalStorageGroupMemCost);	×
155	if (rejected) {	×
156	logger.info(	×
157	"SG ({}) released memory (delta: {}), set system to normal status (totalSgMemCost: {}).",
158	dataRegionInfo.getDataRegion().getDatabaseName(),	×
159	delta,	×
160	totalStorageGroupMemCost);	×
161	}
162	logCurrentTotalSGMemory();	×
163	rejected = false;	×
164	} else if (totalStorageGroupMemCost >= REJECT_THERSHOLD) {	1✔
165	logger.warn(	×
166	"SG ({}) released memory (delta: {}), but system is still in reject status (totalSgMemCost: {}).",
167	dataRegionInfo.getDataRegion().getDatabaseName(),	×
168	delta,	×
169	totalStorageGroupMemCost);	×
170	logCurrentTotalSGMemory();	×
171	rejected = true;	×
172	} else {
173	logger.debug(	1✔
174	"SG ({}) released memory (delta: {}), system is in normal status (totalSgMemCost: {}).",
175	dataRegionInfo.getDataRegion().getDatabaseName(),	1✔
176	delta,	1✔
177	totalStorageGroupMemCost);	1✔
178	logCurrentTotalSGMemory();	1✔
179	rejected = false;	1✔
180	}
181	}	1✔
182
183	public synchronized void addFlushingMemTableCost(long flushingMemTableCost) {
184	this.flushingMemTablesCost += flushingMemTableCost;	1✔
185	}	1✔
186
187	public synchronized void resetFlushingMemTableCost(long flushingMemTableCost) {
188	this.flushingMemTablesCost -= flushingMemTableCost;	1✔
189	}	1✔
190
191	public void addCompactionFileNum(int fileNum, long timeOutInSecond)
192	throws InterruptedException, CompactionFileCountExceededException {
193	if (fileNum > totalFileLimitForCrossTask) {	1✔
194	// source file num is greater than the max file num for compaction
195	throw new CompactionFileCountExceededException(	1✔
196	String.format(	1✔
197	"Required file num %d is greater than the max file num %d for compaction.",
198	fileNum, totalFileLimitForCrossTask));	1✔
199	}
200	long startTime = System.currentTimeMillis();	1✔
201	int originFileNum = this.compactionFileNumCost.get();	1✔
202	while (originFileNum + fileNum > totalFileLimitForCrossTask	1✔
203	\|\| !compactionFileNumCost.compareAndSet(originFileNum, originFileNum + fileNum)) {	1✔
204	if (System.currentTimeMillis() - startTime >= timeOutInSecond * 1000L) {	×
205	throw new CompactionFileCountExceededException(	×
206	String.format(	×
207	"Failed to allocate %d files for compaction after %d seconds, max file num for compaction module is %d, %d files is used.",
208	fileNum, timeOutInSecond, totalFileLimitForCrossTask, originFileNum));	×
209	}
210	Thread.sleep(100);	×
211	originFileNum = this.compactionFileNumCost.get();	×
212	}
213	}	1✔
214
215	public void addCompactionMemoryCost(long memoryCost, long timeOutInSecond)
216	throws InterruptedException, CompactionMemoryNotEnoughException {
217	if (!config.isEnableCompactionMemControl()) {	1✔
218	return;	1✔
219	}
220	if (memoryCost > memorySizeForCompaction) {	1✔
221	// required memory cost is greater than the total memory budget for compaction
222	throw new CompactionMemoryNotEnoughException(	1✔
223	String.format(	1✔
224	"Required memory cost %d bytes is greater than "
225	+ "the total memory budget for compaction %d bytes",
226	memoryCost, memorySizeForCompaction));	1✔
227	}
228	long startTime = System.currentTimeMillis();	1✔
229	long originSize = this.compactionMemoryCost.get();	1✔
230	while (originSize + memoryCost > memorySizeForCompaction	1✔
231	\|\| !compactionMemoryCost.compareAndSet(originSize, originSize + memoryCost)) {	1✔
232	if (System.currentTimeMillis() - startTime >= timeOutInSecond * 1000L) {	×
233	throw new CompactionMemoryNotEnoughException(	×
234	String.format(	×
235	"Failed to allocate %d bytes memory for compaction after %d seconds, "
236	+ "total memory budget for compaction module is %d bytes, %d bytes is used",
237	memoryCost, timeOutInSecond, memorySizeForCompaction, originSize));	×
238	}
239	Thread.sleep(100);	×
240	originSize = this.compactionMemoryCost.get();	×
241	}
242	}	1✔
243
244	public synchronized void resetCompactionMemoryCost(long compactionMemoryCost) {
245	this.compactionMemoryCost.addAndGet(-compactionMemoryCost);	1✔
246	}	1✔
247
248	public synchronized void decreaseCompactionFileNumCost(int fileNum) {
249	this.compactionFileNumCost.addAndGet(-fileNum);	1✔
250	}	1✔
251
252	public long getMemorySizeForCompaction() {
253	if (config.isEnableMemControl()) {	1✔
254	return memorySizeForCompaction;	1✔
255	} else {
256	return Long.MAX_VALUE;	×
257	}
258	}
259
260	public void allocateWriteMemory() {
261	memorySizeForMemtable =	1✔
262	(long)
263	(config.getAllocateMemoryForStorageEngine() * config.getWriteProportionForMemtable());	1✔
264	memorySizeForCompaction =	1✔
265	(long) (config.getAllocateMemoryForStorageEngine() * config.getCompactionProportion());	1✔
266	FLUSH_THERSHOLD = memorySizeForMemtable * config.getFlushProportion();	1✔
267	REJECT_THERSHOLD = memorySizeForMemtable * config.getRejectProportion();	1✔
268	}	1✔
269
270	@TestOnly
271	public void setMemorySizeForCompaction(long size) {
272	memorySizeForCompaction = size;	1✔
273	}	1✔
274
275	@TestOnly
276	public void setTotalFileLimitForCrossTask(int totalFileLimitForCrossTask) {
277	this.totalFileLimitForCrossTask = totalFileLimitForCrossTask;	1✔
278	}	1✔
279
280	@TestOnly
281	public int getTotalFileLimitForCrossTask() {
282	return totalFileLimitForCrossTask;	1✔
283	}
284
285	@TestOnly
286	public AtomicLong getCompactionMemoryCost() {
287	return compactionMemoryCost;	1✔
288	}
289
290	@TestOnly
291	public AtomicInteger getCompactionFileNumCost() {
292	return compactionFileNumCost;	1✔
293	}
294
295	private void logCurrentTotalSGMemory() {
296	logger.debug("Current Sg cost is {}", totalStorageGroupMemCost);	1✔
297	}	1✔
298
299	/**
300	* Order all working memtables in system by memory cost of actual data points in memtable. Mark
301	* the top K TSPs as to be flushed, so that after flushing the K TSPs, the memory cost should be
302	* less than FLUSH_THRESHOLD
303	*/
304	private boolean chooseMemTablesToMarkFlush(TsFileProcessor currentTsFileProcessor) {
305	// If invoke flush by replaying logs, do not flush now!
306	if (reportedStorageGroupMemCostMap.size() == 0) {	×
307	return false;	×
308	}
309	PriorityQueue<TsFileProcessor> allTsFileProcessors =	×
310	new PriorityQueue<>(
311	(o1, o2) -> Long.compare(o2.getWorkMemTableRamCost(), o1.getWorkMemTableRamCost()));	×
312	for (DataRegionInfo dataRegionInfo : reportedStorageGroupMemCostMap.keySet()) {	×
313	allTsFileProcessors.addAll(dataRegionInfo.getAllReportedTsp());	×
314	}	×
315	boolean isCurrentTsFileProcessorSelected = false;	×
316	long memCost = 0;	×
317	long activeMemSize = totalStorageGroupMemCost - flushingMemTablesCost;	×
318	while (activeMemSize - memCost > FLUSH_THERSHOLD) {	×
319	if (allTsFileProcessors.isEmpty()	×
320	\|\| allTsFileProcessors.peek().getWorkMemTableRamCost() == 0) {	×
321	return false;	×
322	}
323	TsFileProcessor selectedTsFileProcessor = allTsFileProcessors.peek();	×
324	memCost += selectedTsFileProcessor.getWorkMemTableRamCost();	×
325	selectedTsFileProcessor.setWorkMemTableShouldFlush();	×
326	flushTaskSubmitThreadPool.submit(	×
327	() -> {
328	selectedTsFileProcessor.submitAFlushTask();	×
329	});	×
330	if (selectedTsFileProcessor == currentTsFileProcessor) {	×
331	isCurrentTsFileProcessorSelected = true;	×
332	}
333	allTsFileProcessors.poll();	×
334	}	×
335	return isCurrentTsFileProcessorSelected;	×
336	}
337
338	public boolean isRejected() {
339	return rejected;	1✔
340	}
341
342	public void setEncodingFasterThanIo(boolean isEncodingFasterThanIo) {
343	this.isEncodingFasterThanIo = isEncodingFasterThanIo;	1✔
344	}	1✔
345
346	public boolean isEncodingFasterThanIo() {
347	return isEncodingFasterThanIo;	1✔
348	}
349
350	public void close() {
351	reportedStorageGroupMemCostMap.clear();	1✔
352	totalStorageGroupMemCost = 0;	1✔
353	rejected = false;	1✔
354	}	1✔
355
356	public static SystemInfo getInstance() {
357	return InstanceHolder.instance;	1✔
358	}
359
360	private static class InstanceHolder {
361
362	private InstanceHolder() {}
363
364	private static SystemInfo instance = new SystemInfo();	1✔
365	}
366
367	public synchronized void applyTemporaryMemoryForFlushing(long estimatedTemporaryMemSize) {
368	memorySizeForMemtable -= estimatedTemporaryMemSize;	1✔
369	FLUSH_THERSHOLD = memorySizeForMemtable * config.getFlushProportion();	1✔
370	REJECT_THERSHOLD = memorySizeForMemtable * config.getRejectProportion();	1✔
371	}	1✔
372
373	public synchronized void releaseTemporaryMemoryForFlushing(long estimatedTemporaryMemSize) {
374	memorySizeForMemtable += estimatedTemporaryMemSize;	1✔
375	FLUSH_THERSHOLD = memorySizeForMemtable * config.getFlushProportion();	1✔
376	REJECT_THERSHOLD = memorySizeForMemtable * config.getRejectProportion();	1✔
377	}	1✔
378
379	public long getTotalMemTableSize() {
380	return totalStorageGroupMemCost;	1✔
381	}
382
383	public double getFlushThershold() {
384	return FLUSH_THERSHOLD;	1✔
385	}
386
387	public double getRejectThershold() {
388	return REJECT_THERSHOLD;	1✔
389	}
390
391	public int flushingMemTableNum() {
392	return FlushManager.getInstance().getNumberOfWorkingTasks();	×
393	}
394	}

apache / iotdb / #9922

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous