Skip to content

Commit a271983

Browse files
CaideyipiCaideyipi
committed
Fixed tsfile concurrency problem
1 parent a09eb7c commit a271983

8 files changed

Lines changed: 83 additions & 71 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeCompactedTsFileInsertionEvent.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ public PipeCompactedTsFileInsertionEvent(
4949
final boolean shouldReportProgress) {
5050
super(
5151
tsFileResource,
52+
null,
5253
bindIsWithMod(originalEvents),
5354
bindIsLoaded(originalEvents),
5455
bindIsGeneratedByHistoricalExtractor(originalEvents),

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java‎

Lines changed: 5 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -89,11 +89,13 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
8989

9090
public PipeTsFileInsertionEvent(final TsFileResource resource, final boolean isLoaded) {
9191
// The modFile must be copied before the event is assigned to the listening pipes
92-
this(resource, true, isLoaded, false, null, 0, null, null, Long.MIN_VALUE, Long.MAX_VALUE);
92+
this(
93+
resource, null, true, isLoaded, false, null, 0, null, null, Long.MIN_VALUE, Long.MAX_VALUE);
9394
}
9495

9596
public PipeTsFileInsertionEvent(
9697
final TsFileResource resource,
98+
final File tsFile,
9799
final boolean isWithMod,
98100
final boolean isLoaded,
99101
final boolean isGeneratedByHistoricalExtractor,
@@ -110,7 +112,7 @@ public PipeTsFileInsertionEvent(
110112
// For events created for source, the tsFile is inherited from the assigner, because the
111113
// original tsFile may be gone, and we need to get the assigner's hard-linked tsFile to
112114
// hard-link it to each pipe dir
113-
this.tsFile = resource.getTsFile();
115+
this.tsFile = Objects.isNull(tsFile) ? resource.getTsFile() : tsFile;
114116

115117
final ModificationFile modFile = resource.getModFile();
116118
this.isWithMod = isWithMod && modFile.exists();
@@ -244,10 +246,6 @@ public long getTimePartitionId() {
244246
public boolean internallyIncreaseResourceReferenceCount(final String holderMessage) {
245247
extractTime = System.nanoTime();
246248
try {
247-
if (Objects.isNull(pipeName)) {
248-
return true;
249-
}
250-
251249
tsFile = PipeDataNodeResourceManager.tsfile().increaseFileReference(tsFile, true, pipeName);
252250
if (isWithMod) {
253251
modFile =
@@ -272,9 +270,6 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa
272270
@Override
273271
public boolean internallyDecreaseResourceReferenceCount(final String holderMessage) {
274272
try {
275-
if (pipeName == null) {
276-
return true;
277-
}
278273
PipeDataNodeResourceManager.tsfile().decreaseFileReference(tsFile, pipeName);
279274
if (isWithMod) {
280275
PipeDataNodeResourceManager.tsfile().decreaseFileReference(modFile, pipeName);
@@ -345,6 +340,7 @@ public PipeTsFileInsertionEvent shallowCopySelfAndBindPipeTaskMetaForProgressRep
345340
final long endTime) {
346341
return new PipeTsFileInsertionEvent(
347342
resource,
343+
tsFile,
348344
isWithMod,
349345
isLoaded,
350346
isGeneratedByHistoricalExtractor,

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@
2929
import org.apache.iotdb.db.pipe.metric.overview.PipeTsFileToTabletsMetrics;
3030
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
3131
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
32-
import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileMemResource;
32+
import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFilePublicResource;
3333

3434
import org.apache.tsfile.file.metadata.IDeviceID;
3535
import org.apache.tsfile.file.metadata.PlainDeviceID;
@@ -81,7 +81,7 @@ public TsFileInsertionDataContainer provide() throws IOException {
8181
// Use scan container to save memory
8282
if ((double) PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes()
8383
/ PipeMemoryManager.getTotalNonFloatingMemorySizeInBytes()
84-
> PipeTsFileMemResource.MEMORY_SUFFICIENT_THRESHOLD) {
84+
> PipeTsFilePublicResource.MEMORY_SUFFICIENT_THRESHOLD) {
8585
return new TsFileInsertionScanDataContainer(
8686
pipeName, creationTime, tsFile, pattern, startTime, endTime, pipeTaskMeta, sourceEvent);
8787
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -562,6 +562,7 @@ public synchronized Event supply() {
562562
final PipeTsFileInsertionEvent event =
563563
new PipeTsFileInsertionEvent(
564564
resource,
565+
null,
565566
shouldTransferModFile,
566567
false,
567568
true,

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileMemResource.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -39,20 +39,22 @@
3939
import java.util.Map;
4040
import java.util.Objects;
4141

42-
public class PipeTsFileMemResource extends PipeTsFileResource {
43-
private static final Logger LOGGER = LoggerFactory.getLogger(PipeTsFileMemResource.class);
42+
public class PipeTsFilePublicResource extends PipeTsFileResource {
43+
private static final Logger LOGGER = LoggerFactory.getLogger(PipeTsFilePublicResource.class);
4444
public static final float MEMORY_SUFFICIENT_THRESHOLD = 0.7f;
4545
private PipeMemoryBlock allocatedMemoryBlock;
4646
private Map<IDeviceID, List<String>> deviceMeasurementsMap = null;
4747
private Map<IDeviceID, Boolean> deviceIsAlignedMap = null;
4848
private Map<String, TSDataType> measurementDataTypeMap = null;
4949

50-
public PipeTsFileMemResource() {
51-
super(null);
50+
public PipeTsFilePublicResource(File hardlinkOrCopiedFile) {
51+
super(hardlinkOrCopiedFile);
5252
}
5353

5454
@Override
5555
public void close() {
56+
super.close();
57+
5658
if (deviceMeasurementsMap != null) {
5759
deviceMeasurementsMap = null;
5860
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -81,8 +81,9 @@ public boolean decreaseReferenceCount() {
8181

8282
@Override
8383
public synchronized void close() {
84+
boolean successful = false;
8485
try {
85-
Files.deleteIfExists(hardlinkOrCopiedFile.toPath());
86+
successful = Files.deleteIfExists(hardlinkOrCopiedFile.toPath());
8687
} catch (final Exception e) {
8788
LOGGER.error(
8889
"PipeTsFileResource: Failed to delete tsfile {} when closing, because {}. Please MANUALLY delete it.",
@@ -91,6 +92,8 @@ public synchronized void close() {
9192
e);
9293
}
9394

94-
LOGGER.info("PipeTsFileResource: Closed tsfile {} and cleaned up.", hardlinkOrCopiedFile);
95+
if (successful) {
96+
LOGGER.info("PipeTsFileResource: Closed tsfile {} and cleaned up.", hardlinkOrCopiedFile);
97+
}
9598
}
9699
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java‎

Lines changed: 59 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -48,8 +48,8 @@ public class PipeTsFileResourceManager {
4848
// This is used to hold the assigner pinned tsFiles.
4949
// Also, it is used to provide metadata cache of the tsFile, and is shared by all the pipe's
5050
// tsFiles.
51-
private final Map<String, PipeTsFileMemResource> hardlinkOrCopiedFileToTsFileMemResourceMap =
52-
new ConcurrentHashMap<>();
51+
private final Map<String, PipeTsFilePublicResource>
52+
hardlinkOrCopiedFileToTsFilePublicResourceMap = new ConcurrentHashMap<>();
5353

5454
// PipeName -> TsFilePath -> PipeTsFileResource
5555
private final Map<String, Map<String, PipeTsFileResource>>
@@ -81,7 +81,7 @@ public File increaseFileReference(
8181
// just increase reference count and return it
8282
segmentLock.lock(file);
8383
try {
84-
if (increaseReferenceIfExists(file, isTsFile, pipeName)) {
84+
if (increaseReferenceIfExists(file, pipeName)) {
8585
return file;
8686
}
8787
} finally {
@@ -93,11 +93,8 @@ public File increaseFileReference(
9393
final File hardlinkOrCopiedFile = getHardlinkOrCopiedFileInPipeDir(file, pipeName);
9494
segmentLock.lock(hardlinkOrCopiedFile);
9595
try {
96-
if (increaseReferenceIfExists(hardlinkOrCopiedFile, isTsFile, pipeName)) {
97-
return hardlinkOrCopiedFileToPipeTsFileResourceMap
98-
.computeIfAbsent(pipeName, pipe -> new ConcurrentHashMap<>())
99-
.get(hardlinkOrCopiedFile.getPath())
100-
.getFile();
96+
if (increaseReferenceIfExists(hardlinkOrCopiedFile, pipeName)) {
97+
return getResourceMap(pipeName).get(hardlinkOrCopiedFile.getPath()).getFile();
10198
}
10299

103100
// If the file is a tsfile, create a hardlink in pipe dir and will return it.
@@ -110,53 +107,55 @@ public File increaseFileReference(
110107
// If the file is not a hardlink or copied file, and there is no related hardlink or copied
111108
// file in pipe dir, create a hardlink or copy it to pipe dir, maintain a reference count for
112109
// the hardlink or copied file, and return the hardlink or copied file.
113-
hardlinkOrCopiedFileToPipeTsFileResourceMap
114-
.computeIfAbsent(pipeName, pipe -> new ConcurrentHashMap<>())
115-
.put(resultFile.getPath(), new PipeTsFileResource(resultFile));
110+
if (Objects.nonNull(pipeName)) {
111+
hardlinkOrCopiedFileToPipeTsFileResourceMap
112+
.computeIfAbsent(pipeName, k -> new ConcurrentHashMap<>())
113+
.put(resultFile.getPath(), new PipeTsFileResource(resultFile));
114+
} else {
115+
hardlinkOrCopiedFileToTsFilePublicResourceMap.put(
116+
resultFile.getPath(), new PipeTsFilePublicResource(resultFile));
117+
}
116118

117-
increaseMemReference(resultFile, isTsFile);
119+
increasePublicReference(resultFile, pipeName);
118120

119121
return resultFile;
120122
} finally {
121123
segmentLock.unlock(hardlinkOrCopiedFile);
122124
}
123125
}
124126

125-
private boolean increaseReferenceIfExists(
126-
final File file, final boolean isTsFile, final @Nullable String pipeName) {
127+
private boolean increaseReferenceIfExists(final File file, final @Nullable String pipeName) {
127128
final String path = file.getPath();
128-
final PipeTsFileResource resource =
129-
hardlinkOrCopiedFileToPipeTsFileResourceMap
130-
.computeIfAbsent(pipeName, pipe -> new ConcurrentHashMap<>())
131-
.get(path);
129+
final PipeTsFileResource resource = getResourceMap(pipeName).get(path);
132130
if (resource != null) {
133131
resource.increaseReferenceCount();
134-
increaseMemReference(file, isTsFile);
132+
increasePublicReference(file, pipeName);
135133
return true;
136134
}
137135
return false;
138136
}
139137

140-
private void increaseMemReference(final File file, final boolean isTsFile) {
141-
if (!isTsFile) {
138+
private void increasePublicReference(final File file, final String pipeName) {
139+
if (Objects.isNull(pipeName)) {
142140
return;
143141
}
144142
// Increase the assigner's file to avoid hard-link or memory cache cleaning
145143
// Note that it does not exist for historical files
146-
hardlinkOrCopiedFileToTsFileMemResourceMap.compute(
147-
getCommonFilePath(file),
144+
final String path = getCommonFilePath(file);
145+
hardlinkOrCopiedFileToTsFilePublicResourceMap.compute(
146+
path,
148147
(k, v) -> {
149148
if (Objects.isNull(v)) {
150-
return new PipeTsFileMemResource();
149+
return new PipeTsFilePublicResource(new File(path));
151150
} else {
152151
v.increaseReferenceCount();
153152
return v;
154153
}
155154
});
156155
}
157156

158-
public static File getHardlinkOrCopiedFileInPipeDir(final File file, final String pipeName)
159-
throws IOException {
157+
public static File getHardlinkOrCopiedFileInPipeDir(
158+
final File file, final @Nullable String pipeName) throws IOException {
160159
try {
161160
return new File(getPipeTsFileDirPath(file, pipeName), getRelativeFilePath(file));
162161
} catch (final Exception e) {
@@ -169,7 +168,8 @@ public static File getHardlinkOrCopiedFileInPipeDir(final File file, final Strin
169168
}
170169
}
171170

172-
private static String getPipeTsFileDirPath(File file, final String pipeName) throws IOException {
171+
private static String getPipeTsFileDirPath(File file, final @Nullable String pipeName)
172+
throws IOException {
173173
while (!file.getName().equals(IoTDBConstant.SEQUENCE_FOLDER_NAME)
174174
&& !file.getName().equals(IoTDBConstant.UNSEQUENCE_FOLDER_NAME)
175175
&& !file.getName().equals(PipeConfig.getInstance().getPipeHardlinkBaseDirName())) {
@@ -206,33 +206,31 @@ private static String getRelativeFilePath(File file) {
206206
* @param hardlinkOrCopiedFile the copied or hard-linked file
207207
*/
208208
public void decreaseFileReference(
209-
final File hardlinkOrCopiedFile, final @Nonnull String pipeName) {
209+
final File hardlinkOrCopiedFile, final @Nullable String pipeName) {
210210
segmentLock.lock(hardlinkOrCopiedFile);
211211
try {
212212
final String filePath = hardlinkOrCopiedFile.getPath();
213-
final PipeTsFileResource resource =
214-
hardlinkOrCopiedFileToPipeTsFileResourceMap
215-
.computeIfAbsent(pipeName, pipe -> new ConcurrentHashMap<>())
216-
.get(filePath);
213+
final PipeTsFileResource resource = getResourceMap(pipeName).get(filePath);
217214
if (resource != null && resource.decreaseReferenceCount()) {
218-
hardlinkOrCopiedFileToPipeTsFileResourceMap
219-
.computeIfAbsent(pipeName, pipe -> new ConcurrentHashMap<>())
220-
.remove(filePath);
215+
getResourceMap(pipeName).remove(filePath);
221216
}
222217
// Decrease the assigner's file to clear hard-link and memory cache
223218
// Note that it does not exist for historical files
224-
decreaseMemReferenceIfExists(hardlinkOrCopiedFile);
219+
decreasePublicReferenceIfExists(hardlinkOrCopiedFile, pipeName);
225220
} finally {
226221
segmentLock.unlock(hardlinkOrCopiedFile);
227222
}
228223
}
229224

230-
private void decreaseMemReferenceIfExists(final File file) {
225+
private void decreasePublicReferenceIfExists(final File file, final @Nullable String pipeName) {
226+
if (Objects.isNull(pipeName)) {
227+
return;
228+
}
231229
// Increase the assigner's file to avoid hard-link or memory cache cleaning
232230
// Note that it does not exist for historical files
233231
final String commonFilePath = getCommonFilePath(file);
234-
if (hardlinkOrCopiedFileToTsFileMemResourceMap.containsKey(commonFilePath)
235-
&& hardlinkOrCopiedFileToTsFileMemResourceMap
232+
if (hardlinkOrCopiedFileToTsFilePublicResourceMap.containsKey(commonFilePath)
233+
&& hardlinkOrCopiedFileToTsFilePublicResourceMap
236234
.get(commonFilePath)
237235
.decreaseReferenceCount()) {
238236
hardlinkOrCopiedFileToPipeTsFileResourceMap.remove(commonFilePath);
@@ -265,7 +263,7 @@ public int getFileReferenceCount(
265263
? hardlinkOrCopiedFileToPipeTsFileResourceMap
266264
.computeIfAbsent(pipeName, pipe -> new ConcurrentHashMap<>())
267265
.get(hardlinkOrCopiedFile.getPath())
268-
: hardlinkOrCopiedFileToTsFileMemResourceMap.get(
266+
: hardlinkOrCopiedFileToTsFilePublicResourceMap.get(
269267
getCommonFilePath(hardlinkOrCopiedFile));
270268
return resource != null ? resource.getReferenceCount() : 0;
271269
} finally {
@@ -286,8 +284,9 @@ public boolean cacheObjectsIfAbsent(final File hardlinkOrCopiedTsFile) throws IO
286284
|| hardlinkOrCopiedTsFile.getParentFile().getParentFile() == null) {
287285
return false;
288286
}
289-
final PipeTsFileMemResource resource =
290-
hardlinkOrCopiedFileToTsFileMemResourceMap.get(getCommonFilePath(hardlinkOrCopiedTsFile));
287+
final PipeTsFilePublicResource resource =
288+
hardlinkOrCopiedFileToTsFilePublicResourceMap.get(
289+
getCommonFilePath(hardlinkOrCopiedTsFile));
291290
return resource != null && resource.cacheObjectsIfAbsent(hardlinkOrCopiedTsFile);
292291
} finally {
293292
segmentLock.unlock(hardlinkOrCopiedTsFile);
@@ -298,8 +297,9 @@ public Map<IDeviceID, List<String>> getDeviceMeasurementsMapFromCache(
298297
final File hardlinkOrCopiedTsFile) throws IOException {
299298
segmentLock.lock(hardlinkOrCopiedTsFile);
300299
try {
301-
final PipeTsFileMemResource resource =
302-
hardlinkOrCopiedFileToTsFileMemResourceMap.get(getCommonFilePath(hardlinkOrCopiedTsFile));
300+
final PipeTsFilePublicResource resource =
301+
hardlinkOrCopiedFileToTsFilePublicResourceMap.get(
302+
getCommonFilePath(hardlinkOrCopiedTsFile));
303303
return resource == null ? null : resource.tryGetDeviceMeasurementsMap(hardlinkOrCopiedTsFile);
304304
} finally {
305305
segmentLock.unlock(hardlinkOrCopiedTsFile);
@@ -310,8 +310,9 @@ public Map<IDeviceID, Boolean> getDeviceIsAlignedMapFromCache(
310310
final File hardlinkOrCopiedTsFile, final boolean cacheOtherMetadata) throws IOException {
311311
segmentLock.lock(hardlinkOrCopiedTsFile);
312312
try {
313-
final PipeTsFileMemResource resource =
314-
hardlinkOrCopiedFileToTsFileMemResourceMap.get(getCommonFilePath(hardlinkOrCopiedTsFile));
313+
final PipeTsFilePublicResource resource =
314+
hardlinkOrCopiedFileToTsFilePublicResourceMap.get(
315+
getCommonFilePath(hardlinkOrCopiedTsFile));
315316
return resource == null
316317
? null
317318
: resource.tryGetDeviceIsAlignedMap(cacheOtherMetadata, hardlinkOrCopiedTsFile);
@@ -324,8 +325,9 @@ public Map<String, TSDataType> getMeasurementDataTypeMapFromCache(
324325
final File hardlinkOrCopiedTsFile) throws IOException {
325326
segmentLock.lock(hardlinkOrCopiedTsFile);
326327
try {
327-
final PipeTsFileMemResource resource =
328-
hardlinkOrCopiedFileToTsFileMemResourceMap.get(getCommonFilePath(hardlinkOrCopiedTsFile));
328+
final PipeTsFilePublicResource resource =
329+
hardlinkOrCopiedFileToTsFilePublicResourceMap.get(
330+
getCommonFilePath(hardlinkOrCopiedTsFile));
329331
return resource == null
330332
? null
331333
: resource.tryGetMeasurementDataTypeMap(hardlinkOrCopiedTsFile);
@@ -334,16 +336,23 @@ public Map<String, TSDataType> getMeasurementDataTypeMapFromCache(
334336
}
335337
}
336338

339+
public Map<String, ? extends PipeTsFileResource> getResourceMap(final @Nullable String pipeName) {
340+
return Objects.nonNull(pipeName)
341+
? hardlinkOrCopiedFileToPipeTsFileResourceMap.computeIfAbsent(
342+
pipeName, k -> new ConcurrentHashMap<>())
343+
: hardlinkOrCopiedFileToTsFilePublicResourceMap;
344+
}
345+
337346
public void pinTsFileResource(
338-
final TsFileResource resource, final boolean withMods, final String pipeName)
347+
final TsFileResource resource, final boolean withMods, final @Nullable String pipeName)
339348
throws IOException {
340349
increaseFileReference(resource.getTsFile(), true, pipeName);
341350
if (withMods && resource.getModFile().exists()) {
342351
increaseFileReference(new File(resource.getModFile().getFilePath()), false, pipeName);
343352
}
344353
}
345354

346-
public void unpinTsFileResource(final TsFileResource resource, final String pipeName)
355+
public void unpinTsFileResource(final TsFileResource resource, final @Nullable String pipeName)
347356
throws IOException {
348357
final File pinnedFile = getHardlinkOrCopiedFileInPipeDir(resource.getTsFile(), pipeName);
349358
decreaseFileReference(pinnedFile, pipeName);

0 commit comments

Comments
 (0)