Skip to content

Commit a1286fc

Browse files
CaideyipiJackieTien97
Caideyipi
authored and
JackieTien97
committed
Pipe: avoid blocking startup on hardlink dir cleanup (#17932)
1 parent f8b8f82 commit a1286fc

4 files changed

Lines changed: 269 additions & 11 deletions

File tree

‎iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1245,6 +1245,18 @@ public final class DataNodePipeMessages {
12451245
"PipeTsFileResource's reference count is decreased to below 0.";
12461246
public static final String PIPE_HARDLINK_DIR_FOUND_DELETING_IT_RESULT =
12471247
"Pipe hardlink dir found, deleting it: {}, result: {}";
1248+
public static final String PIPE_HARDLINK_DIR_FOUND_MOVED_TO_PERIODICAL_DELETE =
1249+
"Pipe hardlink dir found, moved it from {} to {} for throttled periodical deletion.";
1250+
public static final String PIPE_STALE_HARDLINK_DIR_FOUND_REGISTERING_PERIODICAL_DELETE =
1251+
"Stale pipe hardlink dir found, registering it for throttled periodical deletion: {}";
1252+
public static final String PIPE_HARDLINK_DIR_PERIODICAL_DELETE_FINISHED =
1253+
"Finished deleting stale pipe hardlink dir {} by periodical job, result: {}";
1254+
public static final String PIPE_HARDLINK_DIR_PERIODICAL_DELETE_PROGRESS =
1255+
"Periodically deleted {} paths from stale pipe hardlink dirs, current dir: {}, current round result: {}";
1256+
public static final String PIPE_HARDLINK_DIR_PERIODICAL_DELETE_ALL_FINISHED =
1257+
"Finished deleting all stale pipe hardlink dirs by periodical job.";
1258+
public static final String PIPE_HARDLINK_DIR_MOVE_FAILED_DELETING_SYNC =
1259+
"Failed to move pipe hardlink dir {} for periodical deletion, deleting it synchronously.";
12481260
public static final String PIPE_SNAPSHOT_DIR_FOUND_DELETING_IT =
12491261
"Pipe snapshot dir found, deleting it: {},";
12501262
public static final String SHRINK_CALLBACK_IS_NOT_SUPPORTED_IN_PIPEFIXEDMEMORYBLOCK =

‎iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1201,6 +1201,18 @@ public final class DataNodePipeMessages {
12011201
"PipeTsFileResource's reference count is decreased to below 0.";
12021202
public static final String PIPE_HARDLINK_DIR_FOUND_DELETING_IT_RESULT =
12031203
"Pipe hardlink dir found, deleting it: {}, result: {}";
1204+
public static final String PIPE_HARDLINK_DIR_FOUND_MOVED_TO_PERIODICAL_DELETE =
1205+
"Pipe hardlink dir found, moved it from {} to {} for throttled periodical deletion.";
1206+
public static final String PIPE_STALE_HARDLINK_DIR_FOUND_REGISTERING_PERIODICAL_DELETE =
1207+
"Stale pipe hardlink dir found, registering it for throttled periodical deletion: {}";
1208+
public static final String PIPE_HARDLINK_DIR_PERIODICAL_DELETE_FINISHED =
1209+
"Finished deleting stale pipe hardlink dir {} by periodical job, result: {}";
1210+
public static final String PIPE_HARDLINK_DIR_PERIODICAL_DELETE_PROGRESS =
1211+
"Periodically deleted {} paths from stale pipe hardlink dirs, current dir: {}, current round result: {}";
1212+
public static final String PIPE_HARDLINK_DIR_PERIODICAL_DELETE_ALL_FINISHED =
1213+
"Finished deleting all stale pipe hardlink dirs by periodical job.";
1214+
public static final String PIPE_HARDLINK_DIR_MOVE_FAILED_DELETING_SYNC =
1215+
"Failed to move pipe hardlink dir {} for periodical deletion, deleting it synchronously.";
12041216
public static final String PIPE_SNAPSHOT_DIR_FOUND_DELETING_IT =
12051217
"Pipe snapshot dir found, deleting it: {},";
12061218
public static final String SHRINK_CALLBACK_IS_NOT_SUPPORTED_IN_PIPEFIXEDMEMORYBLOCK =

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@ public class PipeDataNodeRuntimeAgent implements IService {
8181
public synchronized void preparePipeResources(
8282
final ResourcesInformationHolder resourcesInformationHolder) throws StartupException {
8383
// Clean sender (connector) hardlink file dir and snapshot dir
84-
PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.clean();
84+
PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.clean(this::registerPeriodicalJob);
8585

8686
// Clean receiver file dir
8787
PipeDataNodeAgent.receiver().cleanPipeReceiverDirs();

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

Lines changed: 244 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -29,33 +29,267 @@
2929
import org.slf4j.LoggerFactory;
3030

3131
import java.io.File;
32+
import java.io.IOException;
33+
import java.nio.file.FileVisitResult;
34+
import java.nio.file.Files;
35+
import java.nio.file.Path;
36+
import java.nio.file.SimpleFileVisitor;
37+
import java.nio.file.attribute.BasicFileAttributes;
38+
import java.util.ArrayList;
39+
import java.util.List;
40+
import java.util.concurrent.atomic.AtomicBoolean;
41+
import java.util.concurrent.atomic.AtomicLong;
3242

3343
public class PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner {
3444

3545
private static final Logger LOGGER =
3646
LoggerFactory.getLogger(PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.class);
47+
private static final String STALE_PIPE_DIR_SUFFIX = ".startup-cleaning-";
48+
private static final String PERIODICAL_CLEANUP_JOB_ID =
49+
"PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner#cleanTsFileDir()";
50+
private static final long DELETE_MAX_PATH_COUNT_PER_ROUND = 100_000L;
51+
private static final long DELETE_MAX_TIME_PER_ROUND_MS = 1_000L;
3752

3853
/**
3954
* Delete the data directory and all of its subdirectories that contain the
4055
* PipeConfig.PIPE_TSFILE_DIR_NAME directory.
4156
*/
42-
public static void clean() {
43-
cleanTsFileDir();
57+
public static void clean(final PeriodicalJobRegistrar periodicalJobRegistrar) {
58+
cleanTsFileDir(periodicalJobRegistrar);
4459
cleanSnapshotDir();
4560
}
4661

47-
private static void cleanTsFileDir() {
62+
private static void cleanTsFileDir(final PeriodicalJobRegistrar periodicalJobRegistrar) {
63+
final String pipeHardlinkBaseDirName = PipeConfig.getInstance().getPipeHardlinkBaseDirName();
64+
final List<File> stalePipeDirs = new ArrayList<>();
4865
for (final String dataDir : IoTDBDescriptor.getInstance().getConfig().getDataDirs()) {
49-
final File pipeHardLinkDir =
50-
new File(
51-
dataDir + File.separator + PipeConfig.getInstance().getPipeHardlinkBaseDirName());
66+
final File localDataDir = new File(dataDir);
67+
collectInterruptedStalePipeDirs(localDataDir, pipeHardlinkBaseDirName, stalePipeDirs);
68+
69+
final File pipeHardLinkDir = new File(localDataDir, pipeHardlinkBaseDirName);
5270
if (pipeHardLinkDir.isDirectory()) {
53-
LOGGER.info(
54-
DataNodePipeMessages.PIPE_HARDLINK_DIR_FOUND_DELETING_IT_RESULT,
55-
pipeHardLinkDir,
56-
FileUtils.deleteQuietly(pipeHardLinkDir));
71+
moveAsideAndCollect(pipeHardLinkDir, pipeHardlinkBaseDirName, stalePipeDirs);
5772
}
5873
}
74+
registerPeriodicalCleanupJob(periodicalJobRegistrar, stalePipeDirs);
75+
}
76+
77+
private static void collectInterruptedStalePipeDirs(
78+
final File localDataDir,
79+
final String pipeHardlinkBaseDirName,
80+
final List<File> stalePipeDirs) {
81+
final File[] stalePipeDirFiles =
82+
localDataDir.listFiles(
83+
file ->
84+
file.isDirectory()
85+
&& file.getName().startsWith(pipeHardlinkBaseDirName + STALE_PIPE_DIR_SUFFIX));
86+
if (stalePipeDirFiles == null) {
87+
return;
88+
}
89+
90+
for (final File stalePipeDir : stalePipeDirFiles) {
91+
LOGGER.info(
92+
DataNodePipeMessages.PIPE_STALE_HARDLINK_DIR_FOUND_REGISTERING_PERIODICAL_DELETE,
93+
stalePipeDir);
94+
stalePipeDirs.add(stalePipeDir);
95+
}
96+
}
97+
98+
private static void moveAsideAndCollect(
99+
final File pipeHardLinkDir,
100+
final String pipeHardlinkBaseDirName,
101+
final List<File> stalePipeDirs) {
102+
try {
103+
final File stalePipeDir = moveAside(pipeHardLinkDir, pipeHardlinkBaseDirName);
104+
LOGGER.info(
105+
DataNodePipeMessages.PIPE_HARDLINK_DIR_FOUND_MOVED_TO_PERIODICAL_DELETE,
106+
pipeHardLinkDir,
107+
stalePipeDir);
108+
stalePipeDirs.add(stalePipeDir);
109+
} catch (final IOException e) {
110+
LOGGER.warn(
111+
DataNodePipeMessages.PIPE_HARDLINK_DIR_MOVE_FAILED_DELETING_SYNC, pipeHardLinkDir, e);
112+
LOGGER.info(
113+
DataNodePipeMessages.PIPE_HARDLINK_DIR_FOUND_DELETING_IT_RESULT,
114+
pipeHardLinkDir,
115+
FileUtils.deleteQuietly(pipeHardLinkDir));
116+
}
117+
}
118+
119+
private static File moveAside(final File pipeHardLinkDir, final String pipeHardlinkBaseDirName)
120+
throws IOException {
121+
final File parentDir = pipeHardLinkDir.getParentFile();
122+
if (parentDir == null) {
123+
throw new IOException("Failed to get parent dir of " + pipeHardLinkDir);
124+
}
125+
126+
final long timestamp = System.currentTimeMillis();
127+
for (int i = 0; ; ++i) {
128+
final File stalePipeDir =
129+
new File(
130+
parentDir, pipeHardlinkBaseDirName + STALE_PIPE_DIR_SUFFIX + timestamp + "-" + i);
131+
if (!stalePipeDir.exists()) {
132+
Files.move(pipeHardLinkDir.toPath(), stalePipeDir.toPath());
133+
return stalePipeDir;
134+
}
135+
}
136+
}
137+
138+
private static void registerPeriodicalCleanupJob(
139+
final PeriodicalJobRegistrar periodicalJobRegistrar, final List<File> stalePipeDirs) {
140+
if (stalePipeDirs.isEmpty()) {
141+
return;
142+
}
143+
144+
periodicalJobRegistrar.register(
145+
PERIODICAL_CLEANUP_JOB_ID,
146+
new PeriodicalStalePipeDirCleaner(stalePipeDirs)::cleanOneRound,
147+
PipeConfig.getInstance().getPipeSubtaskExecutorCronHeartbeatEventIntervalSeconds());
148+
}
149+
150+
private static CleanupRoundResult deleteQuietlyWithThrottle(final File stalePipeDir) {
151+
if (!stalePipeDir.exists()) {
152+
return CleanupRoundResult.finished();
153+
}
154+
155+
final AtomicBoolean deleteResult = new AtomicBoolean(true);
156+
final AtomicLong deletedPathCount = new AtomicLong(0);
157+
final long deadlineNanos = System.nanoTime() + DELETE_MAX_TIME_PER_ROUND_MS * 1_000_000L;
158+
try {
159+
Files.walkFileTree(
160+
stalePipeDir.toPath(),
161+
new SimpleFileVisitor<Path>() {
162+
@Override
163+
public FileVisitResult preVisitDirectory(
164+
final Path dir, final BasicFileAttributes attrs) {
165+
return shouldStop(deletedPathCount, deadlineNanos)
166+
? FileVisitResult.TERMINATE
167+
: FileVisitResult.CONTINUE;
168+
}
169+
170+
@Override
171+
public FileVisitResult visitFile(final Path file, final BasicFileAttributes attrs) {
172+
return deletePath(file, deleteResult, deletedPathCount, deadlineNanos);
173+
}
174+
175+
@Override
176+
public FileVisitResult visitFileFailed(final Path file, final IOException exc) {
177+
deleteResult.set(false);
178+
return deletePath(file, deleteResult, deletedPathCount, deadlineNanos);
179+
}
180+
181+
@Override
182+
public FileVisitResult postVisitDirectory(final Path dir, final IOException exc) {
183+
if (exc != null) {
184+
deleteResult.set(false);
185+
}
186+
return deletePath(dir, deleteResult, deletedPathCount, deadlineNanos);
187+
}
188+
});
189+
} catch (final IOException e) {
190+
deleteResult.set(false);
191+
}
192+
193+
return new CleanupRoundResult(
194+
deletedPathCount.get(), deleteResult.get() && !stalePipeDir.exists(), deleteResult.get());
195+
}
196+
197+
private static FileVisitResult deletePath(
198+
final Path path,
199+
final AtomicBoolean deleteResult,
200+
final AtomicLong deletedPathCount,
201+
final long deadlineNanos) {
202+
if (shouldStop(deletedPathCount, deadlineNanos)) {
203+
return FileVisitResult.TERMINATE;
204+
}
205+
206+
try {
207+
if (Files.deleteIfExists(path)) {
208+
deletedPathCount.incrementAndGet();
209+
}
210+
} catch (final IOException e) {
211+
deleteResult.set(false);
212+
}
213+
return shouldStop(deletedPathCount, deadlineNanos)
214+
? FileVisitResult.TERMINATE
215+
: FileVisitResult.CONTINUE;
216+
}
217+
218+
private static boolean shouldStop(final AtomicLong deletedPathCount, final long deadlineNanos) {
219+
return Thread.currentThread().isInterrupted()
220+
|| deletedPathCount.get() >= DELETE_MAX_PATH_COUNT_PER_ROUND
221+
|| System.nanoTime() >= deadlineNanos;
222+
}
223+
224+
@FunctionalInterface
225+
public interface PeriodicalJobRegistrar {
226+
227+
void register(String id, Runnable periodicalJob, long intervalInSeconds);
228+
}
229+
230+
private static class PeriodicalStalePipeDirCleaner {
231+
232+
private final List<File> stalePipeDirs;
233+
private int currentDirIndex;
234+
private boolean finished;
235+
236+
private PeriodicalStalePipeDirCleaner(final List<File> stalePipeDirs) {
237+
this.stalePipeDirs = stalePipeDirs;
238+
currentDirIndex = 0;
239+
finished = false;
240+
}
241+
242+
private void cleanOneRound() {
243+
if (finished) {
244+
return;
245+
}
246+
247+
long deletedPathCount = 0;
248+
while (currentDirIndex < stalePipeDirs.size()) {
249+
final File stalePipeDir = stalePipeDirs.get(currentDirIndex);
250+
final CleanupRoundResult result = deleteQuietlyWithThrottle(stalePipeDir);
251+
deletedPathCount += result.deletedPathCount;
252+
253+
if (result.finished) {
254+
LOGGER.info(
255+
DataNodePipeMessages.PIPE_HARDLINK_DIR_PERIODICAL_DELETE_FINISHED,
256+
stalePipeDir,
257+
result.success);
258+
++currentDirIndex;
259+
continue;
260+
}
261+
262+
if (deletedPathCount > 0 || !result.success) {
263+
LOGGER.info(
264+
DataNodePipeMessages.PIPE_HARDLINK_DIR_PERIODICAL_DELETE_PROGRESS,
265+
deletedPathCount,
266+
stalePipeDir,
267+
result.success);
268+
}
269+
return;
270+
}
271+
272+
finished = true;
273+
LOGGER.info(DataNodePipeMessages.PIPE_HARDLINK_DIR_PERIODICAL_DELETE_ALL_FINISHED);
274+
}
275+
}
276+
277+
private static class CleanupRoundResult {
278+
279+
private final long deletedPathCount;
280+
private final boolean finished;
281+
private final boolean success;
282+
283+
private CleanupRoundResult(
284+
final long deletedPathCount, final boolean finished, final boolean success) {
285+
this.deletedPathCount = deletedPathCount;
286+
this.finished = finished;
287+
this.success = success;
288+
}
289+
290+
private static CleanupRoundResult finished() {
291+
return new CleanupRoundResult(0, true, true);
292+
}
59293
}
60294

61295
private static void cleanSnapshotDir() {

0 commit comments

Comments
 (0)