This is an automated email from the ASF dual-hosted git repository. hxd pushed a commit to branch feature_async_close_tsfile in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 474f9602025dd5dddf2c9ac61eebe67bc1bbba59 Author: xiangdong huang <[email protected]> AuthorDate: Thu Jun 6 01:35:59 2019 +0800 enable async closing bufferwrite for writing. Not support query in this case now. --- .../engine/bufferwrite/BufferWriteProcessor.java | 84 +++++--- .../db/engine/filenode/CopyOnWriteLinkedList.java | 57 ++++++ .../iotdb/db/engine/filenode/FileNodeManager.java | 38 +++- .../db/engine/filenode/FileNodeProcessor.java | 217 ++++++++++++++------- .../db/engine/overflow/io/OverflowProcessor.java | 11 +- .../db/writelog/node/ExclusiveWriteLogNode.java | 38 +++- .../iotdb/db/writelog/node/WriteLogNode.java | 7 +- .../recover/ExclusiveLogRecoverPerformer.java | 30 +-- .../engine/bufferwrite/BufferWriteBenchmark.java | 10 +- .../bufferwrite/BufferWriteProcessorNewTest.java | 12 +- .../bufferwrite/BufferWriteProcessorTest.java | 32 ++- .../memcontrol/BufferwriteFileSizeControlTest.java | 7 +- .../memcontrol/BufferwriteMetaSizeControlTest.java | 4 +- .../apache/iotdb/db/writelog/WriteLogNodeTest.java | 6 +- 14 files changed, 395 insertions(+), 158 deletions(-) diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessor.java index a8e1e58..987b35d 100644 --- a/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessor.java +++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessor.java @@ -29,10 +29,12 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Consumer; import org.apache.iotdb.db.conf.IoTDBConstant; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.engine.Processor; import org.apache.iotdb.db.engine.filenode.FileNodeManager; +import org.apache.iotdb.db.engine.filenode.TsFileResource; import org.apache.iotdb.db.engine.memcontrol.BasicMemController; import org.apache.iotdb.db.engine.memtable.IMemTable; import org.apache.iotdb.db.engine.memtable.MemSeriesLazyMerger; @@ -69,7 +71,8 @@ public class BufferWriteProcessor extends Processor { private IMemTable workMemTable; private IMemTable flushMemTable; private Action bufferwriteFlushAction; - private Action bufferwriteCloseAction; + //private Action bufferwriteCloseAction; + private Consumer<BufferWriteProcessor> bufferwriteCloseConsumer; private Action filenodeFlushAction; //lastFlushTime time unit: nanosecond @@ -86,6 +89,10 @@ public class BufferWriteProcessor extends Processor { private boolean isClosed = true; private boolean isFlush = false; + + + private TsFileResource currentTsFileResource; + /** * constructor of BufferWriteProcessor. * @@ -97,16 +104,17 @@ public class BufferWriteProcessor extends Processor { * @throws BufferWriteProcessorException BufferWriteProcessorException */ public BufferWriteProcessor(String baseDir, String processorName, String fileName, - Map<String, Action> parameters, VersionController versionController, + Map<String, Action> parameters, Consumer<BufferWriteProcessor> bufferwriteCloseConsumer, + VersionController versionController, FileSchema fileSchema) throws BufferWriteProcessorException { super(processorName); this.fileSchema = fileSchema; this.baseDir = baseDir; bufferwriteFlushAction = parameters.get(FileNodeConstants.BUFFERWRITE_FLUSH_ACTION); - bufferwriteCloseAction = parameters.get(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION); + //bufferwriteCloseAction = parameters.get(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION); filenodeFlushAction = parameters.get(FileNodeConstants.FILENODE_PROCESSOR_FLUSH_ACTION); - + this.bufferwriteCloseConsumer = bufferwriteCloseConsumer; reopen(fileName); try { getLogNode(); @@ -292,9 +300,10 @@ public class BufferWriteProcessor extends Processor { * @param displayMessage message that will appear in system log. * @param version the operation version that will tagged on the to be flushed memtable * (i.e., ChunkGroup) + * @param walTaskId used for declaring what the wal file name suffix is. * @return true if successfully. */ - private boolean flushTask(String displayMessage, long version) { + private boolean flushTask(String displayMessage, long version, long walTaskId) { boolean result; long flushStartTime = System.currentTimeMillis(); LOGGER.info("The bufferwrite processor {} starts flushing {}.", getProcessorName(), @@ -310,7 +319,7 @@ public class BufferWriteProcessor extends Processor { filenodeFlushAction.act(); if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) { - logNode.notifyEndFlush(null); + logNode.notifyEndFlush(null, walTaskId); } result = true; } catch (Exception e) { @@ -358,7 +367,12 @@ public class BufferWriteProcessor extends Processor { // check value count // waiting for the end of last flush operation. try { + long startTime = System.currentTimeMillis(); flushFuture.get(); + long timeCost = System.currentTimeMillis() - startTime; + if (timeCost > 10) { + LOGGER.info("wait for the previous flushing task for {} ms.", timeCost); + } } catch (InterruptedException | ExecutionException e) { throw new IOException(e); } @@ -370,8 +384,11 @@ public class BufferWriteProcessor extends Processor { LOGGER.error("Failed to flush bufferwrite row group when calling the action function."); throw new IOException(e); } + final long walTaskId; if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) { - logNode.notifyStartFlush(); + walTaskId = logNode.notifyStartFlush(); + } else { + walTaskId = 0; } valueCount = 0; switchWorkToFlush(); @@ -380,7 +397,7 @@ public class BufferWriteProcessor extends Processor { memSize.set(0); // switch flushFuture = FlushManager.getInstance().submit(() -> flushTask("asynchronously", - version)); + version, walTaskId)); } else { flushFuture = new ImmediateFuture<>(true); } @@ -398,16 +415,36 @@ public class BufferWriteProcessor extends Processor { return; } try { - long closeStartTime = System.currentTimeMillis(); - // flush data and wait for finishing flush - flush().get(); + + // flush data (if there are flushing task, flush() will be blocked) + Future<Boolean> flush = flush(); + //and wait for finishing flush async + flushFuture = FlushManager.getInstance().submit(() -> closeTask(flush)); + //now, we omit the future of the closeTask. + } catch (IOException e) { + LOGGER.error("Close the bufferwrite processor error, the bufferwrite is {}.", + getProcessorName(), e); + throw new BufferWriteProcessorException(e); + } catch (Exception e) { + LOGGER + .error("Failed to close the bufferwrite processor when calling the action function.", e); + throw new BufferWriteProcessorException(e); + } + } + + private boolean closeTask(Future<Boolean> flush) { + long closeStartTime = System.currentTimeMillis(); + try { + flush.get(); // end file writer.endFile(fileSchema); writer = null; + isClosed = true; + //A BUG may appears here: workMemTable has been cleared, + // but the corresponding TsFile is not maintained in Processor. workMemTable.clear(); - // update the IntervalFile for interval list - bufferwriteCloseAction.act(); + bufferwriteCloseConsumer.accept(this); // flush the changed information for filenode filenodeFlushAction.act(); // delete the restore for this bufferwrite processor @@ -421,16 +458,12 @@ public class BufferWriteProcessor extends Processor { DatetimeUtils.convertMillsecondToZonedDateTime(closeEndTime), closeEndTime - closeStartTime); } - isClosed = true; - } catch (IOException e) { - LOGGER.error("Close the bufferwrite processor error, the bufferwrite is {}.", - getProcessorName(), e); - throw new BufferWriteProcessorException(e); - } catch (Exception e) { - LOGGER - .error("Failed to close the bufferwrite processor when calling the action function.", e); - throw new BufferWriteProcessorException(e); + + }catch (IOException | InterruptedException | ExecutionException|ActionException e) { + LOGGER.error("Close bufferwrite processor {} failed.", getProcessorName(), e); + return false; } + return true; } @Override @@ -558,4 +591,11 @@ public class BufferWriteProcessor extends Processor { public boolean isClosed() { return isClosed; } + + public TsFileResource getCurrentTsFileResource() { + return currentTsFileResource; + } + public void setCurrentTsFileResource(TsFileResource resource) { + this.currentTsFileResource = resource; + } } diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/CopyOnWriteLinkedList.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/CopyOnWriteLinkedList.java new file mode 100644 index 0000000..5021336 --- /dev/null +++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/CopyOnWriteLinkedList.java @@ -0,0 +1,57 @@ +/** + * 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.engine.filenode; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * this class can just guarantee some behavior in a concurrent thread safety mode: + * + * @param <T> + */ +public class CopyOnWriteLinkedList<T> { + LinkedList<T> data = new LinkedList<>(); + List<T> readCopy; + + public synchronized void add(T d) { + data.add(d); + readCopy = new ArrayList<>(data); + } + + public synchronized Iterator<T> iterator() { + readCopy = new ArrayList<>(data); + return data.iterator(); + } + + public synchronized void reset() { + readCopy = null; + } + + public synchronized List<T> read() { + if (readCopy == null) { + readCopy = new ArrayList<>(data); + } + return readCopy; + } +} diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java index 656ece6..96aced1 100644 --- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java +++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java @@ -28,10 +28,14 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.io.FileUtils; +import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBConstant; import org.apache.iotdb.db.conf.IoTDBDescriptor; @@ -95,6 +99,9 @@ public class FileNodeManager implements IStatistic, IService { // There is no need to add concurrently private HashMap<String, AtomicLong> statParamsHashMap; + ScheduledExecutorService closedProcessorCleaner = IoTDBThreadPoolFactory.newScheduledThreadPool(1, + "Closed FileNodeProcessors Cleaner"); + private FileNodeManager(String baseDir) { processorMap = new ConcurrentHashMap<>(); statParamsHashMap = new HashMap<>(); @@ -119,6 +126,28 @@ public class FileNodeManager implements IStatistic, IService { registerStatMetadata(); statMonitor.registerStatistics(MonitorConstants.STAT_STORAGE_DELTA_NAME, this); } + + closedProcessorCleaner.scheduleWithFixedDelay(()->{ + for (FileNodeProcessor fileNodeProcessor : processorMap.values()) { + Iterator<BufferWriteProcessor> iterator = + fileNodeProcessor.getClosingBufferWriteProcessor().iterator(); + while (iterator.hasNext()) { + BufferWriteProcessor processor = iterator.next(); + try { + if (processor.getFlushFuture().get(10, TimeUnit.MILLISECONDS)) { + //if finished, we can remove it. + iterator.remove(); + } + } catch (InterruptedException | ExecutionException e) { + LOGGER.error("Close bufferwrite processor {} failed.", processor.getProcessorName(), e); + } catch (TimeoutException e) { + //do nothing. + } + } + fileNodeProcessor.getClosingBufferWriteProcessor().reset(); + } + }, 0, 3000, TimeUnit.MILLISECONDS); + } public static FileNodeManager getInstance() { @@ -413,7 +442,8 @@ public class FileNodeManager implements IStatistic, IService { String bufferwriteBaseDir = bufferWriteProcessor.getBaseDir(); String bufferwriteRelativePath = bufferWriteProcessor.getFileRelativePath(); try { - fileNodeProcessor.addIntervalFileNode(new File(new File(bufferwriteBaseDir), bufferwriteRelativePath)); + bufferWriteProcessor.setCurrentTsFileResource(new TsFileResource(new File(new File(bufferwriteBaseDir), bufferwriteRelativePath), false)); + fileNodeProcessor.addIntervalFileNode(bufferWriteProcessor.getCurrentTsFileResource()); } catch (Exception e) { if (!isMonitor) { updateStatHashMapWhenFail(tsRecord); @@ -1176,6 +1206,7 @@ public class FileNodeManager implements IStatistic, IService { } catch (FileNodeManagerException e) { LOGGER.error("Failed to close file node manager because .", e); } + closedProcessorCleaner.shutdownNow(); } @Override @@ -1225,4 +1256,7 @@ public class FileNodeManager implements IStatistic, IService { TsFileDBConf.getFileNodeDir()); } -} \ No newline at end of file +} + + + diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java index 209c1f1..16e591b 100644 --- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java +++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java @@ -33,17 +33,25 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; +import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Objects; import java.util.Set; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Consumer; +import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBConstant; import org.apache.iotdb.db.conf.IoTDBDescriptor; @@ -81,6 +89,7 @@ import org.apache.iotdb.db.query.control.FileReaderManager; import org.apache.iotdb.db.query.factory.SeriesReaderFactory; import org.apache.iotdb.db.query.reader.IReader; import org.apache.iotdb.db.sync.conf.Constans; +import org.apache.iotdb.db.utils.ImmediateFuture; import org.apache.iotdb.db.utils.MemUtils; import org.apache.iotdb.db.utils.QueryUtils; import org.apache.iotdb.db.utils.TimeValuePair; @@ -126,7 +135,7 @@ public class FileNodeProcessor extends Processor implements IStatistic { private Map<String, Long> flushLastUpdateTimeMap; private Map<String, List<TsFileResource>> invertedIndexOfFiles; private TsFileResource emptyTsFileResource; - private TsFileResource currentTsFileResource; +// private TsFileResource currentTsFileResource; private List<TsFileResource> newFileNodes; private FileNodeProcessorStatus isMerging; @@ -143,6 +152,12 @@ public class FileNodeProcessor extends Processor implements IStatistic { */ private long lastMergeTime = -1; private BufferWriteProcessor bufferWriteProcessor = null; + + //the bufferwrite Processors that are closing. (Because they are not closed well, + // their memtable are not released and we have to query data from them. + //private ConcurrentSkipListSet<BufferWriteProcessor> closingBufferWriteProcessor = new ConcurrentSkipListSet<>(); + private CopyOnWriteLinkedList<BufferWriteProcessor> closingBufferWriteProcessor = new CopyOnWriteLinkedList<>(); + private OverflowProcessor overflowProcessor = null; private Set<Integer> oldMultiPassTokenSet = null; private Set<Integer> newMultiPassTokenSet = new HashSet<>(); @@ -188,30 +203,50 @@ public class FileNodeProcessor extends Processor implements IStatistic { } }; - private Action bufferwriteCloseAction = new Action() { - - @Override - public void act() { - synchronized (fileNodeProcessorStore) { - fileNodeProcessorStore.setLastUpdateTimeMap(lastUpdateTimeMap); - addLastTimeToIntervalFile(); - fileNodeProcessorStore.setNewFileNodes(newFileNodes); - } - } - - private void addLastTimeToIntervalFile() { +// private Action bufferwriteCloseAction = new Action() { +// +// @Override +// public void act() { +// synchronized (fileNodeProcessorStore) { +// fileNodeProcessorStore.setLastUpdateTimeMap(lastUpdateTimeMap); +// addLastTimeToIntervalFile(); +// fileNodeProcessorStore.setNewFileNodes(newFileNodes); +// } +// } +// +// private void addLastTimeToIntervalFile() { +// +// if (!newFileNodes.isEmpty()) { +// // end time with one start time +// Map<String, Long> endTimeMap = new HashMap<>(); +// for (Entry<String, Long> startTime : currentTsFileResource.getStartTimeMap().entrySet()) { +// String deviceId = startTime.getKey(); +// endTimeMap.put(deviceId, lastUpdateTimeMap.get(deviceId)); +// } +// currentTsFileResource.setEndTimeMap(endTimeMap); +// } +// } +// }; + + private Consumer<BufferWriteProcessor> bufferwriteCloseConsumer = (bwProcessor) -> { + synchronized (fileNodeProcessorStore) { + fileNodeProcessorStore.setLastUpdateTimeMap(lastUpdateTimeMap); if (!newFileNodes.isEmpty()) { // end time with one start time Map<String, Long> endTimeMap = new HashMap<>(); - for (Entry<String, Long> startTime : currentTsFileResource.getStartTimeMap().entrySet()) { + TsFileResource resource = bwProcessor.getCurrentTsFileResource(); + for (Entry<String, Long> startTime : resource.getStartTimeMap().entrySet()) { String deviceId = startTime.getKey(); endTimeMap.put(deviceId, lastUpdateTimeMap.get(deviceId)); } - currentTsFileResource.setEndTimeMap(endTimeMap); + resource.setEndTimeMap(endTimeMap); } + fileNodeProcessorStore.setNewFileNodes(newFileNodes); } }; + + private Action overflowFlushAction = () -> { // update the new TsFileResource List and emptyIntervalFile. @@ -365,10 +400,7 @@ public class FileNodeProcessor extends Processor implements IStatistic { /** * add interval FileNode. */ - void addIntervalFileNode(File file) throws ActionException, IOException { - - TsFileResource tsFileResource = new TsFileResource(file, false); - this.currentTsFileResource = tsFileResource; + void addIntervalFileNode(TsFileResource tsFileResource) throws ActionException { newFileNodes.add(tsFileResource); fileNodeProcessorStore.setNewFileNodes(newFileNodes); flushFileNodeProcessorAction.act(); @@ -380,26 +412,27 @@ public class FileNodeProcessor extends Processor implements IStatistic { * @param deviceId device ID */ void setIntervalFileNodeStartTime(String deviceId) { - if (currentTsFileResource.getStartTime(deviceId) == -1) { - currentTsFileResource.setStartTime(deviceId, flushLastUpdateTimeMap.get(deviceId)); + if (getBufferWriteProcessor().getCurrentTsFileResource().getStartTime(deviceId) == -1) { + getBufferWriteProcessor().getCurrentTsFileResource().setStartTime(deviceId, + flushLastUpdateTimeMap.get(deviceId)); if (!invertedIndexOfFiles.containsKey(deviceId)) { invertedIndexOfFiles.put(deviceId, new ArrayList<>()); } - invertedIndexOfFiles.get(deviceId).add(currentTsFileResource); + invertedIndexOfFiles.get(deviceId).add(getBufferWriteProcessor().getCurrentTsFileResource()); } } void setIntervalFileNodeStartTime(String deviceId, long time) { if (time != -1) { - currentTsFileResource.setStartTime(deviceId, time); + getBufferWriteProcessor().getCurrentTsFileResource().setStartTime(deviceId, time); } else { - currentTsFileResource.removeTime(deviceId); - invertedIndexOfFiles.get(deviceId).remove(currentTsFileResource); + getBufferWriteProcessor().getCurrentTsFileResource().removeTime(deviceId); + invertedIndexOfFiles.get(deviceId).remove(getBufferWriteProcessor().getCurrentTsFileResource()); } } long getIntervalFileNodeStartTime(String deviceId) { - return currentTsFileResource.getStartTime(deviceId); + return getBufferWriteProcessor().getCurrentTsFileResource().getStartTime(deviceId); } private void addAllFileIntoIndex(List<TsFileResource> fileList) { @@ -446,40 +479,46 @@ public class FileNodeProcessor extends Processor implements IStatistic { */ public void fileNodeRecovery() throws FileNodeProcessorException { // restore bufferwrite - if (!newFileNodes.isEmpty() && !newFileNodes.get(newFileNodes.size() - 1).isClosed()) { - // - // add the current file - // - currentTsFileResource = newFileNodes.get(newFileNodes.size() - 1); - - // this bufferwrite file is not close by normal operation - String damagedFilePath = newFileNodes.get(newFileNodes.size() - 1).getFile().getAbsolutePath(); - String[] fileNames = damagedFilePath.split("\\" + File.separator); - // all information to recovery the damaged file. - // contains file seriesPath, action parameters and processorName - parameters.put(FileNodeConstants.BUFFERWRITE_FLUSH_ACTION, bufferwriteFlushAction); - parameters.put(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION, bufferwriteCloseAction); - parameters - .put(FileNodeConstants.FILENODE_PROCESSOR_FLUSH_ACTION, flushFileNodeProcessorAction); - String baseDir = directories - .getTsFileFolder(newFileNodes.get(newFileNodes.size() - 1).getBaseDirIndex()); - if (LOGGER.isInfoEnabled()) { - LOGGER.info( - "The filenode processor {} will recovery the bufferwrite processor, " - + "the bufferwrite file is {}", - getProcessorName(), fileNames[fileNames.length - 1]); - } + if (!newFileNodes.isEmpty()) { + int i = newFileNodes.size() -1; + //for (int i = newFileNodes.size() -1; i >= 0; i--) { + if (!newFileNodes.get(i).isClosed()) { + // + // add the current file + // + //currentTsFileResource = newFileNodes.get(newFileNodes.size() - 1); + + // this bufferwrite file is not close by normal operation + String damagedFilePath = newFileNodes.get(i).getFile().getAbsolutePath(); + String[] fileNames = damagedFilePath.split("\\" + File.separator); + // all information to recovery the damaged file. + // contains file seriesPath, action parameters and processorName + parameters.put(FileNodeConstants.BUFFERWRITE_FLUSH_ACTION, bufferwriteFlushAction); + //parameters.put(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION, bufferwriteCloseAction); + parameters + .put(FileNodeConstants.FILENODE_PROCESSOR_FLUSH_ACTION, flushFileNodeProcessorAction); + String baseDir = directories + .getTsFileFolder(newFileNodes.get(newFileNodes.size() - 1).getBaseDirIndex()); + if (LOGGER.isInfoEnabled()) { + LOGGER.info( + "The filenode processor {} will recovery the bufferwrite processor, " + + "the bufferwrite file is {}", + getProcessorName(), fileNames[fileNames.length - 1]); + } - try { - bufferWriteProcessor = new BufferWriteProcessor(baseDir, getProcessorName(), - fileNames[fileNames.length - 1], parameters, versionController, fileSchema); - } catch (BufferWriteProcessorException e) { - LOGGER.error( - "The filenode processor {} failed to recovery the bufferwrite processor, " - + "the last bufferwrite file is {}.", - getProcessorName(), fileNames[fileNames.length - 1]); - throw new FileNodeProcessorException(e); - } + try { + bufferWriteProcessor = new BufferWriteProcessor(baseDir, getProcessorName(), + fileNames[fileNames.length - 1], parameters, bufferwriteCloseConsumer, + versionController, fileSchema); + } catch (BufferWriteProcessorException e) { + LOGGER.error( + "The filenode processor {} failed to recovery the bufferwrite processor, " + + "the last bufferwrite file is {}.", + getProcessorName(), fileNames[fileNames.length - 1]); + throw new FileNodeProcessorException(e); + } + } + //} } // restore the overflow processor LOGGER.info("The filenode processor {} will recovery the overflow processor.", @@ -516,6 +555,12 @@ public class FileNodeProcessor extends Processor implements IStatistic { addAllFileIntoIndex(newFileNodes); } + //when calling this method, the bufferWriteProcessor must not be null + private BufferWriteProcessor getBufferWriteProcessor() { + + return bufferWriteProcessor; + } + /** * get buffer write processor by processor name and insert time. */ @@ -524,7 +569,7 @@ public class FileNodeProcessor extends Processor implements IStatistic { if (bufferWriteProcessor == null) { Map<String, Action> params = new HashMap<>(); params.put(FileNodeConstants.BUFFERWRITE_FLUSH_ACTION, bufferwriteFlushAction); - params.put(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION, bufferwriteCloseAction); + //params.put(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION, bufferwriteCloseAction); params .put(FileNodeConstants.FILENODE_PROCESSOR_FLUSH_ACTION, flushFileNodeProcessorAction); String baseDir = directories.getNextFolderForTsfile(); @@ -534,7 +579,7 @@ public class FileNodeProcessor extends Processor implements IStatistic { bufferWriteProcessor = new BufferWriteProcessor(baseDir, processorName, insertTime + FileNodeConstants.BUFFERWRITE_FILE_SEPARATOR + System.currentTimeMillis(), - params, versionController, fileSchema); + params, bufferwriteCloseConsumer, versionController, fileSchema); } catch (BufferWriteProcessorException e) { throw new FileNodeProcessorException(String .format("The filenode processor %s failed to get the bufferwrite processor.", @@ -821,7 +866,7 @@ public class FileNodeProcessor extends Processor implements IStatistic { try { List<Modification> pathModifications = context.getPathModifications( - currentTsFileResource.getModFile(), deviceId + bufferWriteProcessor.getCurrentTsFileResource().getModFile(), deviceId + IoTDBConstant.PATH_SEPARATOR + measurementId ); if (!pathModifications.isEmpty()) { @@ -1011,7 +1056,10 @@ public class FileNodeProcessor extends Processor implements IStatistic { try { LOGGER.info("The filenode processor {} prepares for merge, closes the bufferwrite processor", getProcessorName()); - closeBufferWrite(); + Future<Boolean> future = closeBufferWrite(); + future.get(); + LOGGER.info("The bufferwrite processor {} is closed successfully", + getProcessorName()); // try to get overflow processor getOverflowProcessor(getProcessorName()); // must close the overflow processor @@ -1021,7 +1069,7 @@ public class FileNodeProcessor extends Processor implements IStatistic { LOGGER.info("The filenode processor {} prepares for merge, closes the overflow processor", getProcessorName()); getOverflowProcessor().close(); - } catch (FileNodeProcessorException | OverflowProcessorException | IOException e) { + } catch (FileNodeProcessorException | OverflowProcessorException | IOException | InterruptedException | ExecutionException e) { LOGGER.error("The filenode processor {} prepares for merge error.", getProcessorName()); writeUnlock(); throw new ErrorDebugException(e); @@ -1736,21 +1784,26 @@ public class FileNodeProcessor extends Processor implements IStatistic { /** * Close the bufferwrite processor. */ - public void closeBufferWrite() throws FileNodeProcessorException { + public Future<Boolean> closeBufferWrite() throws FileNodeProcessorException { if (bufferWriteProcessor == null || bufferWriteProcessor.isClosed()) { - return; + return new ImmediateFuture<>(true); } try { while (!bufferWriteProcessor.canBeClosed()) { waitForBufferWriteClose(); } bufferWriteProcessor.close(); + Future<Boolean> result = bufferWriteProcessor.getFlushFuture(); + closingBufferWriteProcessor.add(bufferWriteProcessor); bufferWriteProcessor = null; + return result; } catch (BufferWriteProcessorException e) { throw new FileNodeProcessorException(e); } } + + private void waitForBufferWriteClose() { try { LOGGER.info("The bufferwrite {} can't be closed, wait 100ms", @@ -1792,7 +1845,12 @@ public class FileNodeProcessor extends Processor implements IStatistic { @Override public void close() throws FileNodeProcessorException { - closeBufferWrite(); + Future<Boolean> result = closeBufferWrite(); + try { + result.get(); + } catch (InterruptedException | ExecutionException e) { + throw new FileNodeProcessorException(e); + } closeOverflow(); for (TsFileResource fileNode : newFileNodes) { if (fileNode.getModFile() != null) { @@ -1924,12 +1982,19 @@ public class FileNodeProcessor extends Processor implements IStatistic { private void deleteBufferWriteFiles(String deviceId, Deletion deletion, List<ModificationFile> updatedModFiles) throws IOException { - if (currentTsFileResource != null && currentTsFileResource.containsDevice(deviceId)) { - currentTsFileResource.getModFile().write(deletion); - updatedModFiles.add(currentTsFileResource.getModFile()); + BufferWriteProcessor bufferWriteProcessor = getBufferWriteProcessor(); + TsFileResource resource = null; + if (bufferWriteProcessor != null) { + //bufferWriteProcessor == null means the bufferWriteProcessor is closed now. + resource = bufferWriteProcessor.getCurrentTsFileResource(); + if (resource != null && resource.containsDevice(deviceId)) { + resource.getModFile().write(deletion); + updatedModFiles.add(resource.getModFile()); + } } + for (TsFileResource fileNode : newFileNodes) { - if (fileNode != currentTsFileResource && fileNode.containsDevice(deviceId) + if (fileNode != resource && fileNode.containsDevice(deviceId) && fileNode.getStartTime(deviceId) <= deletion.getTimestamp()) { fileNode.getModFile().write(deletion); updatedModFiles.add(fileNode.getModFile()); @@ -1984,6 +2049,10 @@ public class FileNodeProcessor extends Processor implements IStatistic { } } + public CopyOnWriteLinkedList<BufferWriteProcessor> getClosingBufferWriteProcessor() { + return closingBufferWriteProcessor; + } + @Override public boolean equals(Object o) { if (this == o) { @@ -2007,7 +2076,6 @@ public class FileNodeProcessor extends Processor implements IStatistic { Objects.equals(flushLastUpdateTimeMap, that.flushLastUpdateTimeMap) && Objects.equals(invertedIndexOfFiles, that.invertedIndexOfFiles) && Objects.equals(emptyTsFileResource, that.emptyTsFileResource) && - Objects.equals(currentTsFileResource, that.currentTsFileResource) && Objects.equals(newFileNodes, that.newFileNodes) && isMerging == that.isMerging && Objects.equals(fileNodeProcessorStore, that.fileNodeProcessorStore) && @@ -2022,7 +2090,6 @@ public class FileNodeProcessor extends Processor implements IStatistic { Objects.equals(fileSchema, that.fileSchema) && Objects.equals(flushFileNodeProcessorAction, that.flushFileNodeProcessorAction) && Objects.equals(bufferwriteFlushAction, that.bufferwriteFlushAction) && - Objects.equals(bufferwriteCloseAction, that.bufferwriteCloseAction) && Objects.equals(overflowFlushAction, that.overflowFlushAction); } @@ -2030,12 +2097,12 @@ public class FileNodeProcessor extends Processor implements IStatistic { public int hashCode() { return Objects.hash(super.hashCode(), statStorageDeltaName, statParamsHashMap, isOverflowed, lastUpdateTimeMap, flushLastUpdateTimeMap, invertedIndexOfFiles, - emptyTsFileResource, currentTsFileResource, newFileNodes, isMerging, + emptyTsFileResource, newFileNodes, isMerging, numOfMergeFile, fileNodeProcessorStore, fileNodeRestoreFilePath, lastMergeTime, bufferWriteProcessor, overflowProcessor, oldMultiPassTokenSet, newMultiPassTokenSet, oldMultiPassCount, newMultiPassCount, shouldRecovery, parameters, fileSchema, flushFileNodeProcessorAction, bufferwriteFlushAction, - bufferwriteCloseAction, overflowFlushAction, multiPassLockToken); + overflowFlushAction, multiPassLockToken); } public class MergeRunnale implements Runnable { diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java index c031181..8b19e5c 100644 --- a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java +++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java @@ -483,7 +483,7 @@ public class OverflowProcessor extends Processor { return isFlush; } - private boolean flushTask(String displayMessage) { + private boolean flushTask(String displayMessage, long walTaskId) { boolean result; long flushStartTime = System.currentTimeMillis(); try { @@ -496,7 +496,7 @@ public class OverflowProcessor extends Processor { filenodeFlushAction.act(); // write-ahead log if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) { - logNode.notifyEndFlush(null); + logNode.notifyEndFlush(null, walTaskId); } result = true; } catch (IOException e) { @@ -556,22 +556,23 @@ public class OverflowProcessor extends Processor { LOGGER.error("Flush the overflow rowGroup to file faied, when overflowFlushAction act"); throw new IOException(e); } - + long taskId = 0; if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) { try { - logNode.notifyStartFlush(); + taskId = logNode.notifyStartFlush(); } catch (IOException e) { LOGGER.error("Overflow processor {} encountered an error when notifying log node, {}", getProcessorName(), e); } } + final long walTaskId = taskId; BasicMemController.getInstance().releaseUsage(this, memSize.get()); memSize.set(0); valueCount = 0; // switch from work to flush switchWorkToFlush(); flushFuture = FlushManager.getInstance().submit( () -> - flushTask("asynchronously")); + flushTask("asynchronously", walTaskId)); } else { flushFuture = new ImmediateFuture(true); } diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java index 1b2682f..3c7fb4d 100644 --- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java +++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java @@ -22,21 +22,23 @@ import java.io.File; import java.io.IOException; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; - +import java.util.regex.Matcher; +import java.util.regex.Pattern; import org.apache.commons.io.FileUtils; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.RecoverException; import org.apache.iotdb.db.qp.physical.PhysicalPlan; +import org.apache.iotdb.db.qp.physical.transfer.PhysicalPlanLogTransfer; import org.apache.iotdb.db.utils.MemUtils; import org.apache.iotdb.db.writelog.LogPosition; import org.apache.iotdb.db.writelog.io.ILogWriter; import org.apache.iotdb.db.writelog.io.LogWriter; import org.apache.iotdb.db.writelog.recover.ExclusiveLogRecoverPerformer; import org.apache.iotdb.db.writelog.recover.RecoverPerformer; -import org.apache.iotdb.db.qp.physical.transfer.PhysicalPlanLogTransfer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -67,6 +69,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive private ReadWriteLock forceLock = new ReentrantReadWriteLock(); + private AtomicLong taskId; /** * constructor of ExclusiveWriteLogNode. * @@ -79,10 +82,24 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive this.identifier = identifier; this.logDirectory = config.getWalFolder() + File.separator + this.identifier; new File(logDirectory).mkdirs(); + //read current wals to get the largest task id. + long task = 1; + Pattern pattern = Pattern.compile(OLD_SUFFIX+"(\\d+)"); + for (File file : new File(logDirectory).listFiles()) { + Matcher matcher = pattern.matcher(file.getName()); + if (matcher.find()) { + long id = Long.parseLong(matcher.group(1)); + if (id > task) { + task = id; + } + } + } + taskId = new AtomicLong(task); recoverPerformer = new ExclusiveLogRecoverPerformer(restoreFilePath, processorStoreFilePath, this); currentFileWriter = new LogWriter(logDirectory + File.separator + WAL_FILE_NAME); + } public void setRecoverPerformer(RecoverPerformer recoverPerformer) { @@ -144,27 +161,30 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive * Warning : caller must have lock. */ @Override - public void notifyStartFlush() { + public long notifyStartFlush() { close(); File oldLogFile = new File(logDirectory + File.separator + WAL_FILE_NAME); - File newLogFile = new File(logDirectory + File.separator + WAL_FILE_NAME + OLD_SUFFIX); if (!oldLogFile.exists()) { - return; + return 0; } + + long id = taskId.incrementAndGet(); + File newLogFile = new File(logDirectory + File.separator + WAL_FILE_NAME + OLD_SUFFIX + id); if (!oldLogFile.renameTo(newLogFile)) { logger.error("Log node {} renaming log file failed!", identifier); } else { logger.info("Log node {} renamed log file, file size is {}", identifier, MemUtils.bytesCntToStr(newLogFile.length())); } + return id; } /* * Warning : caller must have lock. */ @Override - public void notifyEndFlush(List<LogPosition> logPositions) { - discard(); + public void notifyEndFlush(List<LogPosition> logPositions, long taskId) { + discard(taskId); } @Override @@ -250,8 +270,8 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive } } - private void discard() { - File oldLogFile = new File(logDirectory + File.separator + WAL_FILE_NAME + OLD_SUFFIX); + private void discard(long id) { + File oldLogFile = new File(logDirectory + File.separator + WAL_FILE_NAME + OLD_SUFFIX + id); if (!oldLogFile.exists()) { logger.info("No old log to be deleted"); } else { diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java index fb53388..b2e7fc3 100644 --- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java +++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java @@ -58,14 +58,17 @@ public interface WriteLogNode { /** * When a FileNode attempts to start a flush, this method must be called to rename log file. + * @return the task id ( being used in the renamed log file) */ - void notifyStartFlush() throws IOException; + long notifyStartFlush() throws IOException; /** * When the flush of a FlieNode ends, this method must be called to check if log file needs * cleaning. + * @param logPositions (deprecated) + * @param taskId the task id that notifyStartFlush() returns. */ - void notifyEndFlush(List<LogPosition> logPositions); + void notifyEndFlush(List<LogPosition> logPositions, long taskId); /** * return identifier of the log node. diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java index 0b58283..9d4b958 100644 --- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java +++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java @@ -286,17 +286,19 @@ public class ExclusiveLogRecoverPerformer implements RecoverPerformer { return failedCnt; } - private void replayLog() throws RecoverException { + private void replayLog() throws RecoverException { int failedEntryCnt = 0; // if old log file exists, replay it first. - File oldLogFile = new File( - writeLogNode.getLogDirectory() + File.separator + ExclusiveWriteLogNode.WAL_FILE_NAME - + ExclusiveWriteLogNode.OLD_SUFFIX); - try { - failedEntryCnt += replayLogFile(oldLogFile); - } catch (IOException e) { - throw new RecoverException(e); + File logFolder = new File(writeLogNode.getLogDirectory()); + for (File file : logFolder.listFiles( name -> name.getName().startsWith(ExclusiveWriteLogNode.WAL_FILE_NAME + + ExclusiveWriteLogNode.OLD_SUFFIX))) { + try { + failedEntryCnt += replayLogFile(file); + } catch (IOException e) { + throw new RecoverException(e); + } } + // then replay new log File newLogFile = new File( writeLogNode.getLogDirectory() + File.separator + ExclusiveWriteLogNode.WAL_FILE_NAME); @@ -341,13 +343,15 @@ public class ExclusiveLogRecoverPerformer implements RecoverPerformer { failedFiles.add(recoverProcessorStoreFilePath); } // clean log file - File oldLogFile = new File( - writeLogNode.getLogDirectory() + File.separator + ExclusiveWriteLogNode.WAL_FILE_NAME - + ExclusiveWriteLogNode.OLD_SUFFIX); - if (oldLogFile.exists() && !oldLogFile.delete()) { + File logFolder = new File(writeLogNode.getLogDirectory()); + for (File file : logFolder.listFiles( name -> name.getName().startsWith(ExclusiveWriteLogNode.WAL_FILE_NAME + + ExclusiveWriteLogNode.OLD_SUFFIX))) { + if (file.exists() && !file.delete()) { logger.error("Log node {} cannot delete old log file", writeLogNode.getLogDirectory()); - failedFiles.add(oldLogFile.getPath()); + failedFiles.add(file.getPath()); + } } + File newLogFile = new File( writeLogNode.getLogDirectory() + File.separator + ExclusiveWriteLogNode.WAL_FILE_NAME); if (newLogFile.exists() && !newLogFile.delete()) { diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteBenchmark.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteBenchmark.java index 7c62ff7..5a013f7 100644 --- a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteBenchmark.java +++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteBenchmark.java @@ -81,12 +81,6 @@ public class BufferWriteBenchmark { System.out.println(FileNodeConstants.BUFFERWRITE_FLUSH_ACTION); } }); - parameters.put(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION, new Action() { - @Override - public void act() throws ActionException { - System.out.println(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION); - } - }); parameters.put(FileNodeConstants.FILENODE_PROCESSOR_FLUSH_ACTION, new Action() { @Override public void act() throws ActionException { @@ -96,7 +90,9 @@ public class BufferWriteBenchmark { BufferWriteProcessor bufferWriteProcessor = new BufferWriteProcessor("BufferBenchmark", "bench", "benchFile", - parameters, SysTimeVersionController.INSTANCE, fileSchema); + parameters, + x -> System.out.println(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION), + SysTimeVersionController.INSTANCE, fileSchema); long startTime = System.currentTimeMillis(); for (int i = 0; i < numOfPoint; i++) { diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorNewTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorNewTest.java index 8cbf640..1fb5e48 100644 --- a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorNewTest.java +++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorNewTest.java @@ -61,13 +61,6 @@ public class BufferWriteProcessorNewTest { } }; - Action bfcloseaction = new Action() { - - @Override - public void act() throws ActionException { - } - }; - Action fnflushaction = new Action() { @Override @@ -86,7 +79,6 @@ public class BufferWriteProcessorNewTest { @Before public void setUp() throws Exception { parameters.put(FileNodeConstants.BUFFERWRITE_FLUSH_ACTION, bfflushaction); - parameters.put(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION, bfcloseaction); parameters.put(FileNodeConstants.FILENODE_PROCESSOR_FLUSH_ACTION, fnflushaction); MetadataManagerHelper.initMetadata(); EnvironmentUtils.envSetUp(); @@ -103,7 +95,7 @@ public class BufferWriteProcessorNewTest { throws BufferWriteProcessorException, WriteProcessException, IOException, InterruptedException { bufferwrite = new BufferWriteProcessor(Directories.getInstance().getFolderForTest(), processorName, filename, - parameters, SysTimeVersionController.INSTANCE, + parameters, x->{}, SysTimeVersionController.INSTANCE, FileSchemaUtils.constructFileSchema(processorName)); assertEquals(processorName + File.separator + filename, bufferwrite.getFileRelativePath()); assertTrue(bufferwrite.isNewProcessor()); @@ -157,7 +149,7 @@ public class BufferWriteProcessorNewTest { // test recovery BufferWriteProcessor bufferWriteProcessor = new BufferWriteProcessor( - Directories.getInstance().getFolderForTest(), processorName, filename, parameters, + Directories.getInstance().getFolderForTest(), processorName, filename, parameters, x->{}, SysTimeVersionController.INSTANCE, FileSchemaUtils.constructFileSchema(processorName)); pair = bufferWriteProcessor.queryBufferWriteData(processorName, measurementId, dataType, props); diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorTest.java index f058562..ebaa612 100644 --- a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorTest.java +++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorTest.java @@ -23,16 +23,20 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.io.File; import java.io.FileOutputStream; import java.io.IOException; +import java.nio.Buffer; import java.util.Collections; import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import org.apache.iotdb.db.conf.directories.Directories; import org.apache.iotdb.db.engine.MetadataManagerHelper; import org.apache.iotdb.db.engine.PathUtils; @@ -75,6 +79,8 @@ public class BufferWriteProcessorTest { } }; + Consumer<BufferWriteProcessor> bfcloseConsumer = x -> {}; + Action fnflushaction = new Action() { @Override @@ -97,7 +103,6 @@ public class BufferWriteProcessorTest { @Before public void setUp() throws Exception { parameters.put(FileNodeConstants.BUFFERWRITE_FLUSH_ACTION, bfflushaction); - parameters.put(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION, bfcloseaction); parameters.put(FileNodeConstants.FILENODE_PROCESSOR_FLUSH_ACTION, fnflushaction); // origin value groupSizeInByte = TsFileConf.groupSizeInByte; @@ -121,7 +126,7 @@ public class BufferWriteProcessorTest { public void testWriteAndAbnormalRecover() throws WriteProcessException, InterruptedException, IOException, ProcessorException { bufferwrite = new BufferWriteProcessor(directories.getFolderForTest(), deviceId, insertPath, - parameters, SysTimeVersionController.INSTANCE, + parameters, bfcloseConsumer, SysTimeVersionController.INSTANCE, FileSchemaUtils.constructFileSchema(deviceId)); for (int i = 1; i < 100; i++) { bufferwrite.write(deviceId, measurementId, i, dataType, String.valueOf(i)); @@ -143,10 +148,15 @@ public class BufferWriteProcessorTest { File file = new File("temp"); restoreFile.renameTo(file); bufferwrite.close(); + try { + bufferwrite.getFlushFuture().get(); + } catch (ExecutionException e) { + fail(e.getMessage()); + } file.renameTo(restoreFile); BufferWriteProcessor bufferWriteProcessor = new BufferWriteProcessor( directories.getFolderForTest(), deviceId, - insertPath, parameters, SysTimeVersionController.INSTANCE, + insertPath, parameters, bfcloseConsumer, SysTimeVersionController.INSTANCE, FileSchemaUtils.constructFileSchema(deviceId)); assertTrue(insertFile.exists()); assertEquals(insertFileLength, insertFile.length()); @@ -158,6 +168,11 @@ public class BufferWriteProcessorTest { assertEquals(measurementId, chunkMetaData.getMeasurementUid()); assertEquals(dataType, chunkMetaData.getTsDataType()); bufferWriteProcessor.close(); + try { + bufferWriteProcessor.getFlushFuture().get(); + } catch (ExecutionException e) { + fail(e.getMessage()); + } assertFalse(restoreFile.exists()); } @@ -165,7 +180,7 @@ public class BufferWriteProcessorTest { public void testWriteAndNormalRecover() throws WriteProcessException, ProcessorException, InterruptedException { bufferwrite = new BufferWriteProcessor(directories.getFolderForTest(), deviceId, insertPath, - parameters, SysTimeVersionController.INSTANCE, + parameters, bfcloseConsumer, SysTimeVersionController.INSTANCE, FileSchemaUtils.constructFileSchema(deviceId)); for (int i = 1; i < 100; i++) { bufferwrite.write(deviceId, measurementId, i, dataType, String.valueOf(i)); @@ -179,7 +194,7 @@ public class BufferWriteProcessorTest { assertTrue(restoreFile.exists()); BufferWriteProcessor bufferWriteProcessor = new BufferWriteProcessor( directories.getFolderForTest(), deviceId, - insertPath, parameters, SysTimeVersionController.INSTANCE, + insertPath, parameters, bfcloseConsumer, SysTimeVersionController.INSTANCE, FileSchemaUtils.constructFileSchema(deviceId)); Pair<ReadOnlyMemChunk, List<ChunkMetaData>> pair = bufferWriteProcessor .queryBufferWriteData(deviceId, measurementId, dataType, Collections.emptyMap()); @@ -190,6 +205,11 @@ public class BufferWriteProcessorTest { assertEquals(dataType, chunkMetaData.getTsDataType()); bufferWriteProcessor.close(); bufferwrite.close(); + try { + bufferwrite.getFlushFuture().get(); + } catch (ExecutionException e) { + fail(e.getMessage()); + } assertFalse(restoreFile.exists()); } @@ -197,7 +217,7 @@ public class BufferWriteProcessorTest { public void testWriteAndQuery() throws WriteProcessException, InterruptedException, ProcessorException { bufferwrite = new BufferWriteProcessor(directories.getFolderForTest(), deviceId, insertPath, - parameters, SysTimeVersionController.INSTANCE, + parameters, bfcloseConsumer, SysTimeVersionController.INSTANCE, FileSchemaUtils.constructFileSchema(deviceId)); assertFalse(bufferwrite.isFlush()); assertTrue(bufferwrite.canBeClosed()); diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteFileSizeControlTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteFileSizeControlTest.java index 9a2393c..3655273 100644 --- a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteFileSizeControlTest.java +++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteFileSizeControlTest.java @@ -25,6 +25,7 @@ import static org.junit.Assert.fail; import java.io.File; import java.util.HashMap; import java.util.Map; +import java.util.function.Consumer; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.conf.directories.Directories; @@ -63,6 +64,7 @@ public class BufferwriteFileSizeControlTest { public void act() throws ActionException { } }; + Consumer<BufferWriteProcessor> bfcloseConsumer = bfProcessor -> {}; Action fnflushaction = new Action() { @@ -131,13 +133,14 @@ public class BufferwriteFileSizeControlTest { Map<String, Action> parameters = new HashMap<>(); parameters.put(FileNodeConstants.BUFFERWRITE_FLUSH_ACTION, bfflushaction); - parameters.put(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION, bfcloseaction); + //parameters.put(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION, bfcloseaction); parameters.put(FileNodeConstants.FILENODE_PROCESSOR_FLUSH_ACTION, fnflushaction); try { processor = new BufferWriteProcessor(Directories.getInstance().getFolderForTest(), nsp, filename, - parameters, SysTimeVersionController.INSTANCE, FileSchemaUtils.constructFileSchema(nsp)); + parameters, bfcloseConsumer, SysTimeVersionController.INSTANCE, + FileSchemaUtils.constructFileSchema(nsp)); } catch (BufferWriteProcessorException e) { e.printStackTrace(); fail(e.getMessage()); diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteMetaSizeControlTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteMetaSizeControlTest.java index 926f2cb..7f1e4b7 100644 --- a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteMetaSizeControlTest.java +++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteMetaSizeControlTest.java @@ -64,6 +64,7 @@ public class BufferwriteMetaSizeControlTest { } }; + Action fnflushaction = new Action() { @Override @@ -131,13 +132,12 @@ public class BufferwriteMetaSizeControlTest { Map<String, Action> parameters = new HashMap<>(); parameters.put(FileNodeConstants.BUFFERWRITE_FLUSH_ACTION, bfflushaction); - parameters.put(FileNodeConstants.BUFFERWRITE_CLOSE_ACTION, bfcloseaction); parameters.put(FileNodeConstants.FILENODE_PROCESSOR_FLUSH_ACTION, fnflushaction); try { processor = new BufferWriteProcessor(Directories.getInstance().getFolderForTest(), nsp, filename, - parameters, SysTimeVersionController.INSTANCE, FileSchemaUtils.constructFileSchema(nsp)); + parameters, x->{}, SysTimeVersionController.INSTANCE, FileSchemaUtils.constructFileSchema(nsp)); } catch (BufferWriteProcessorException e) { e.printStackTrace(); fail(e.getMessage()); diff --git a/iotdb/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java b/iotdb/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java index 8b1c241..321387f 100644 --- a/iotdb/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java +++ b/iotdb/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java @@ -164,13 +164,13 @@ public class WriteLogNodeTest { config.getWalFolder() + File.separator + "root.logTestDevice" + File.separator + "wal"); assertTrue(walFile.exists()); - logNode.notifyStartFlush(); + long taskId = logNode.notifyStartFlush(); File oldWalFile = new File( - config.getWalFolder() + File.separator + "root.logTestDevice" + File.separator + "wal-old"); + config.getWalFolder() + File.separator + "root.logTestDevice" + File.separator + "wal-old"+taskId); assertTrue(oldWalFile.exists()); assertTrue(oldWalFile.length() > 0); - logNode.notifyEndFlush(null); + logNode.notifyEndFlush(null, taskId); assertTrue(!oldWalFile.exists()); assertEquals(0, walFile.length());
