You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by "dlmarion (via GitHub)" <gi...@apache.org> on 2025/03/11 21:08:09 UTC
[PR] Make garbage collecting wal and recovery files faster [accumulo]
dlmarion opened a new pull request, #5399:
URL: https://github.com/apache/accumulo/pull/5399
Uses the same property and technique used for deleting RFiles to delete wal and recovery files.
Closes #5397
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992332064
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +274,97 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private Future<?> removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ return deleteThreadPool.submit(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+
+ final Map<Path,Future<?>> futures = new HashMap<>(collection.size());
+ final AtomicLong counter = new AtomicLong();
+
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- log.debug("Removing {} WAL {}", stateFile.getFirst(), path);
- status.currentLog.deleted += removeFile(path);
+ futures.put(path, removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path));
+ }
+
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting wal file" + f.getKey(), e);
+ }
+ }
+ }
}
- return status.currentLog.deleted;
+ deleteThreadPool.shutdown();
+ try {
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
+ }
+ } catch (InterruptedException e1) {
+ log.error("{}", e1.getMessage(), e1);
+ }
+ status.currentLog.deleted += counter.get();
+ return counter.get();
}
private long removeFiles(Collection<Path> values) {
- long count = 0;
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+ final Map<Path,Future<?>> futures = new HashMap<>(values.size());
+ final AtomicLong counter = new AtomicLong();
+
for (Path path : values) {
- log.debug("Removing recovery log {}", path);
- count += removeFile(path);
+ futures.put(path,
+ removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path));
+ }
+
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting recovery log file" + f.getKey(),
+ e);
+ }
+ }
+ }
+ }
+ deleteThreadPool.shutdown();
Review Comment:
If an exception is throw the pool would not be shut down.
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +274,97 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private Future<?> removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ return deleteThreadPool.submit(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+
+ final Map<Path,Future<?>> futures = new HashMap<>(collection.size());
+ final AtomicLong counter = new AtomicLong();
+
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- log.debug("Removing {} WAL {}", stateFile.getFirst(), path);
- status.currentLog.deleted += removeFile(path);
+ futures.put(path, removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path));
+ }
+
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting wal file" + f.getKey(), e);
+ }
+ }
+ }
}
- return status.currentLog.deleted;
+ deleteThreadPool.shutdown();
+ try {
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
+ }
+ } catch (InterruptedException e1) {
+ log.error("{}", e1.getMessage(), e1);
+ }
+ status.currentLog.deleted += counter.get();
+ return counter.get();
}
private long removeFiles(Collection<Path> values) {
- long count = 0;
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+ final Map<Path,Future<?>> futures = new HashMap<>(values.size());
+ final AtomicLong counter = new AtomicLong();
+
for (Path path : values) {
- log.debug("Removing recovery log {}", path);
- count += removeFile(path);
+ futures.put(path,
+ removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path));
+ }
+
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting recovery log file" + f.getKey(),
+ e);
+ }
+ }
+ }
+ }
+ deleteThreadPool.shutdown();
+ try {
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
Review Comment:
Could remove this code now that futures are being waited on.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992334192
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -283,42 +286,77 @@ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong
} catch (FileNotFoundException ex) {
// ignored
} catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
+ log.error("Unable to delete {}", path, ex);
}
});
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
- .createExecutorService(context.getConfiguration(), Property.GC_DELETE_THREADS);
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+
+ final Map<Path,Future<?>> futures = new HashMap<>(collection.size());
final AtomicLong counter = new AtomicLong();
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- removeFile(deleteThreadPool, path, counter,
- "Removing " + stateFile.getFirst() + " WAL " + path);
+ futures.put(path, removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path));
}
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting wal file" + f.getKey(), e);
+ }
+ }
+ }
+ }
deleteThreadPool.shutdown();
try {
while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
}
} catch (InterruptedException e1) {
log.error("{}", e1.getMessage(), e1);
}
+ status.currentLog.deleted += counter.get();
return counter.get();
}
private long removeFiles(Collection<Path> values) {
+
final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
- .createExecutorService(context.getConfiguration(), Property.GC_DELETE_THREADS);
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+ final Map<Path,Future<?>> futures = new HashMap<>(values.size());
final AtomicLong counter = new AtomicLong();
for (Path path : values) {
- removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path);
+ futures.put(path,
+ removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path));
}
+ while (!futures.isEmpty()) {
Review Comment:
I was thinking the faster I process the ones that are complete and remove from the map, the faster I'm giving back memory to the VM. I probably need to put a wait at the bottom of the loop though. If the first one the iterator returns is the last one submitted, then I would be waiting until they are all done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion merged PR #5399:
URL: https://github.com/apache/accumulo/pull/5399
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992167152
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +271,62 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ deleteThreadPool.execute(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
Review Comment:
I was wondering if we need to handle any other exception types here as an uncaught exception can bubble up and kill the thead pool. It's pretty common to catch all exceptions as a catch all.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992184773
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +271,62 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ deleteThreadPool.execute(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
Review Comment:
If you used futures that might be better, you could wait for them to finish instead of relying on the thread pool to shutdown for completion. Either way could work, I just figured I'd mention it as I'm not too familiar with this code but I noticed we were only catching those specific exceptions so was curious if a random runtime exception would cause issues with the new thread pool.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#issuecomment-2719097515
>
> The only suggestion I have is using a separate `GC_WAL_DELETE_THREADS` property.
>
Added in 7c54495
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992165306
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +271,62 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ deleteThreadPool.execute(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete wal {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_THREADS);
+ final AtomicLong counter = new AtomicLong();
+
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- log.debug("Removing {} WAL {}", stateFile.getFirst(), path);
- status.currentLog.deleted += removeFile(path);
+ removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path);
}
- return status.currentLog.deleted;
+
+ deleteThreadPool.shutdown();
+ try {
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
+ }
+ } catch (InterruptedException e1) {
+ log.error("{}", e1.getMessage(), e1);
+ }
+ return counter.get();
Review Comment:
Yep, I forgot to increment the counter. I created it though, so I was half way there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1993355777
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -283,42 +286,77 @@ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong
} catch (FileNotFoundException ex) {
// ignored
} catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
+ log.error("Unable to delete {}", path, ex);
}
});
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
- .createExecutorService(context.getConfiguration(), Property.GC_DELETE_THREADS);
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+
+ final Map<Path,Future<?>> futures = new HashMap<>(collection.size());
final AtomicLong counter = new AtomicLong();
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- removeFile(deleteThreadPool, path, counter,
- "Removing " + stateFile.getFirst() + " WAL " + path);
+ futures.put(path, removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path));
}
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting wal file" + f.getKey(), e);
+ }
+ }
+ }
+ }
deleteThreadPool.shutdown();
try {
while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
}
} catch (InterruptedException e1) {
log.error("{}", e1.getMessage(), e1);
}
+ status.currentLog.deleted += counter.get();
return counter.get();
}
private long removeFiles(Collection<Path> values) {
+
final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
- .createExecutorService(context.getConfiguration(), Property.GC_DELETE_THREADS);
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+ final Map<Path,Future<?>> futures = new HashMap<>(values.size());
final AtomicLong counter = new AtomicLong();
for (Path path : values) {
- removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path);
+ futures.put(path,
+ removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path));
}
+ while (!futures.isEmpty()) {
Review Comment:
Added a sleep at the bottom of the while loop in 93f5c6d.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992102978
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +271,62 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ deleteThreadPool.execute(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete wal {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_THREADS);
+ final AtomicLong counter = new AtomicLong();
+
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- log.debug("Removing {} WAL {}", stateFile.getFirst(), path);
- status.currentLog.deleted += removeFile(path);
+ removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path);
}
- return status.currentLog.deleted;
+
+ deleteThreadPool.shutdown();
+ try {
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
+ }
+ } catch (InterruptedException e1) {
+ log.error("{}", e1.getMessage(), e1);
+ }
+ return counter.get();
Review Comment:
The code was incrementing this deleted counter, should it still do that?
```suggestion
status.currentLog.deleted += counter.get();
return counter.get();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992271319
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +271,62 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ deleteThreadPool.execute(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete wal {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_THREADS);
+ final AtomicLong counter = new AtomicLong();
+
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- log.debug("Removing {} WAL {}", stateFile.getFirst(), path);
- status.currentLog.deleted += removeFile(path);
+ removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path);
}
- return status.currentLog.deleted;
+
+ deleteThreadPool.shutdown();
+ try {
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
+ }
+ } catch (InterruptedException e1) {
+ log.error("{}", e1.getMessage(), e1);
+ }
+ return counter.get();
Review Comment:
Updated in 7c54495
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "ddanielr (via GitHub)" <gi...@apache.org>.
ddanielr commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1991729384
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +271,62 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ deleteThreadPool.execute(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete wal {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
Review Comment:
When deleting rfiles, the threadpool is created and shutdown for each delete pass to ensure that all rfiles are deleted before we remove the gcCandidates from their respective metadata location.
For WALs, the GC looks at each wal directory location to find files, then compares the file against the live tservers set, recovery operations, or dead tserver list to see if the file can be removed.
Since we aren't processing gcCandidates, do we need to create and delete the threadpool on each removeFiles run?
If we made a longer lived pool then we could enable metrics on the threadpool and get more details about how many threads might be utilized.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#issuecomment-2717674014
Full IT build successful
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "ddanielr (via GitHub)" <gi...@apache.org>.
ddanielr commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1991777365
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +271,62 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ deleteThreadPool.execute(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete wal {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
Review Comment:
Started looking at the WALStateManager code and realized my understanding of the WAL log process was incorrect. While the GC does look at recovery directory locations, the rest of the WAL information is stored in ZK.
After the WALs are processed, the WALStateManager is used to update ZK references to the WAL files.
So the existing threadpool start/shutdown pattern makes sense here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992168943
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +271,62 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ deleteThreadPool.execute(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete wal {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_THREADS);
+ final AtomicLong counter = new AtomicLong();
+
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- log.debug("Removing {} WAL {}", stateFile.getFirst(), path);
- status.currentLog.deleted += removeFile(path);
+ removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path);
}
- return status.currentLog.deleted;
+
+ deleteThreadPool.shutdown();
+ try {
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
+ }
+ } catch (InterruptedException e1) {
+ log.error("{}", e1.getMessage(), e1);
+ }
+ return counter.get();
Review Comment:
On second look, counter is being passed to removeFile, and incremented there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992238093
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +271,62 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ deleteThreadPool.execute(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete wal {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_THREADS);
+ final AtomicLong counter = new AtomicLong();
+
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- log.debug("Removing {} WAL {}", stateFile.getFirst(), path);
- status.currentLog.deleted += removeFile(path);
+ removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path);
}
- return status.currentLog.deleted;
+
+ deleteThreadPool.shutdown();
+ try {
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
+ }
+ } catch (InterruptedException e1) {
+ log.error("{}", e1.getMessage(), e1);
+ }
+ return counter.get();
Review Comment:
yeah counter is being incrementing removeFile, but the old code used to increment `status.currentLog.deleted` which may be used for display on the monitor and logging. Seems like that is no longer incremented.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1993355074
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +274,97 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private Future<?> removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ return deleteThreadPool.submit(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+
+ final Map<Path,Future<?>> futures = new HashMap<>(collection.size());
+ final AtomicLong counter = new AtomicLong();
+
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- log.debug("Removing {} WAL {}", stateFile.getFirst(), path);
- status.currentLog.deleted += removeFile(path);
+ futures.put(path, removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path));
+ }
+
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting wal file" + f.getKey(), e);
+ }
+ }
+ }
}
- return status.currentLog.deleted;
+ deleteThreadPool.shutdown();
+ try {
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
+ }
+ } catch (InterruptedException e1) {
+ log.error("{}", e1.getMessage(), e1);
+ }
+ status.currentLog.deleted += counter.get();
+ return counter.get();
}
private long removeFiles(Collection<Path> values) {
- long count = 0;
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+ final Map<Path,Future<?>> futures = new HashMap<>(values.size());
+ final AtomicLong counter = new AtomicLong();
+
for (Path path : values) {
- log.debug("Removing recovery log {}", path);
- count += removeFile(path);
+ futures.put(path,
+ removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path));
+ }
+
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting recovery log file" + f.getKey(),
+ e);
+ }
+ }
+ }
+ }
+ deleteThreadPool.shutdown();
Review Comment:
Fixed this in 93f5c6d
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992176891
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +271,62 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ deleteThreadPool.execute(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
Review Comment:
The previous code just bubbled the RuntimeExceptions up the call stack. I could call `submit` instead, test the futures, and return any errors up the stack.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1994235750
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +273,102 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private Future<?> removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ return deleteThreadPool.submit(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
- for (Pair<WalState,Path> stateFile : collection) {
- Path path = stateFile.getSecond();
- log.debug("Removing {} WAL {}", stateFile.getFirst(), path);
- status.currentLog.deleted += removeFile(path);
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+ final Map<Path,Future<?>> futures = new HashMap<>(collection.size());
+ final AtomicLong counter = new AtomicLong();
+
+ try {
+ for (Pair<WalState,Path> stateFile : collection) {
+ Path path = stateFile.getSecond();
+ futures.put(path, removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path));
+ }
+
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting wal file" + f.getKey(), e);
+ }
+ }
+ }
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while sleeping", e);
+ }
+ }
+ } finally {
+ deleteThreadPool.shutdownNow();
}
- return status.currentLog.deleted;
+ status.currentLog.deleted += counter.get();
+ return counter.get();
}
private long removeFiles(Collection<Path> values) {
- long count = 0;
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+ final Map<Path,Future<?>> futures = new HashMap<>(values.size());
+ final AtomicLong counter = new AtomicLong();
+
for (Path path : values) {
- log.debug("Removing recovery log {}", path);
- count += removeFile(path);
+ futures.put(path,
+ removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path));
+ }
+
+ try {
Review Comment:
smh, I did that on the other method. I'll fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1994190621
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +273,102 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private Future<?> removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ return deleteThreadPool.submit(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
- for (Pair<WalState,Path> stateFile : collection) {
- Path path = stateFile.getSecond();
- log.debug("Removing {} WAL {}", stateFile.getFirst(), path);
- status.currentLog.deleted += removeFile(path);
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+ final Map<Path,Future<?>> futures = new HashMap<>(collection.size());
+ final AtomicLong counter = new AtomicLong();
+
+ try {
+ for (Pair<WalState,Path> stateFile : collection) {
+ Path path = stateFile.getSecond();
+ futures.put(path, removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path));
+ }
+
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting wal file" + f.getKey(), e);
+ }
+ }
+ }
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while sleeping", e);
+ }
+ }
+ } finally {
+ deleteThreadPool.shutdownNow();
}
- return status.currentLog.deleted;
+ status.currentLog.deleted += counter.get();
+ return counter.get();
}
private long removeFiles(Collection<Path> values) {
- long count = 0;
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+ final Map<Path,Future<?>> futures = new HashMap<>(values.size());
+ final AtomicLong counter = new AtomicLong();
+
for (Path path : values) {
- log.debug("Removing recovery log {}", path);
- count += removeFile(path);
+ futures.put(path,
+ removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path));
+ }
+
+ try {
Review Comment:
could move this try up to be immediately after the creation of deleteThreadPool
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992271077
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +271,62 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ deleteThreadPool.execute(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
Review Comment:
Updated to use futures in 7c54495
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1993354696
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -266,37 +274,97 @@ private long removeTabletServerMarkers(Map<UUID,TServerInstance> uidMap,
return result;
}
- private long removeFile(Path path) {
- try {
- if (!useTrash || !fs.moveToTrash(path)) {
- fs.deleteRecursively(path);
+ private Future<?> removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong counter,
+ String msg) {
+ return deleteThreadPool.submit(() -> {
+ try {
+ log.debug(msg);
+ if (!useTrash || !fs.moveToTrash(path)) {
+ fs.deleteRecursively(path);
+ }
+ counter.incrementAndGet();
+ } catch (FileNotFoundException ex) {
+ // ignored
+ } catch (IOException ex) {
+ log.error("Unable to delete {}", path, ex);
}
- return 1;
- } catch (FileNotFoundException ex) {
- // ignored
- } catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
- }
-
- return 0;
+ });
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+
+ final Map<Path,Future<?>> futures = new HashMap<>(collection.size());
+ final AtomicLong counter = new AtomicLong();
+
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- log.debug("Removing {} WAL {}", stateFile.getFirst(), path);
- status.currentLog.deleted += removeFile(path);
+ futures.put(path, removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path));
+ }
+
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting wal file" + f.getKey(), e);
+ }
+ }
+ }
}
- return status.currentLog.deleted;
+ deleteThreadPool.shutdown();
+ try {
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
+ }
+ } catch (InterruptedException e1) {
+ log.error("{}", e1.getMessage(), e1);
+ }
+ status.currentLog.deleted += counter.get();
+ return counter.get();
}
private long removeFiles(Collection<Path> values) {
- long count = 0;
+
+ final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+ final Map<Path,Future<?>> futures = new HashMap<>(values.size());
+ final AtomicLong counter = new AtomicLong();
+
for (Path path : values) {
- log.debug("Removing recovery log {}", path);
- count += removeFile(path);
+ futures.put(path,
+ removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path));
+ }
+
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting recovery log file" + f.getKey(),
+ e);
+ }
+ }
+ }
+ }
+ deleteThreadPool.shutdown();
+ try {
+ while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
Review Comment:
Fixed in 93f5c6d
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#discussion_r1992328131
##########
server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java:
##########
@@ -283,42 +286,77 @@ private void removeFile(ExecutorService deleteThreadPool, Path path, AtomicLong
} catch (FileNotFoundException ex) {
// ignored
} catch (IOException ex) {
- log.error("Unable to delete wal {}", path, ex);
+ log.error("Unable to delete {}", path, ex);
}
});
}
private long removeFiles(Collection<Pair<WalState,Path>> collection, final GCStatus status) {
final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
- .createExecutorService(context.getConfiguration(), Property.GC_DELETE_THREADS);
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+
+ final Map<Path,Future<?>> futures = new HashMap<>(collection.size());
final AtomicLong counter = new AtomicLong();
for (Pair<WalState,Path> stateFile : collection) {
Path path = stateFile.getSecond();
- removeFile(deleteThreadPool, path, counter,
- "Removing " + stateFile.getFirst() + " WAL " + path);
+ futures.put(path, removeFile(deleteThreadPool, path, counter,
+ "Removing " + stateFile.getFirst() + " WAL " + path));
}
+ while (!futures.isEmpty()) {
+ Iterator<Entry<Path,Future<?>>> iter = futures.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<Path,Future<?>> f = iter.next();
+ if (f.getValue().isDone()) {
+ try {
+ iter.remove();
+ f.getValue().get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException("Uncaught exception deleting wal file" + f.getKey(), e);
+ }
+ }
+ }
+ }
deleteThreadPool.shutdown();
try {
while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
}
} catch (InterruptedException e1) {
log.error("{}", e1.getMessage(), e1);
}
+ status.currentLog.deleted += counter.get();
return counter.get();
}
private long removeFiles(Collection<Path> values) {
+
final ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
- .createExecutorService(context.getConfiguration(), Property.GC_DELETE_THREADS);
+ .createExecutorService(context.getConfiguration(), Property.GC_DELETE_WAL_THREADS);
+ final Map<Path,Future<?>> futures = new HashMap<>(values.size());
final AtomicLong counter = new AtomicLong();
for (Path path : values) {
- removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path);
+ futures.put(path,
+ removeFile(deleteThreadPool, path, counter, "Removing recovery log " + path));
}
+ while (!futures.isEmpty()) {
Review Comment:
The get() method in a future waits until its done or failed, so should be able to loop through all futures once calling get.
```java
futures.forEach((path,future)->{
try {
future.get();
} catch (InterruptedException | ExecutionException e) {
throw new RuntimeException("Uncaught exception deleting recovery log file" + path,
e);
}
});
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make garbage collecting wal and recovery files faster [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on PR #5399:
URL: https://github.com/apache/accumulo/pull/5399#issuecomment-2715741430
Kicked off IT build
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org