Author: mduerig
Date: Thu Oct 6 09:03:32 2016
New Revision: 1763533
URL: http://svn.apache.org/viewvc?rev=1763533&view=rev
Log:
OAK-4138: Decouple revision cleanup from the flush thread
Schedule cleanup instead of executing it on the flush thread
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java?rev=1763533&r1=1763532&r2=1763533&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
Thu Oct 6 09:03:32 2016
@@ -202,11 +202,6 @@ public class FileStore implements Segmen
private final GCJournal gcJournal;
/**
- * Flag to request revision cleanup during the next flush.
- */
- private final AtomicBoolean cleanupNeeded = new AtomicBoolean(false);
-
- /**
* List of old tar file generations that are waiting to be removed. They
can
* not be removed immediately, because they first need to be closed, and
the
* JVM needs to release the memory mapped file references.
@@ -524,9 +519,7 @@ public class FileStore implements Segmen
if (!gcOptions.isPaused()) {
logAndClear(segmentWriter.getNodeWriteTimeStats(),
segmentWriter.getNodeCompactTimeStats());
log(segmentWriter.getNodeCacheOccupancyInfo());
- if (compact()) {
- cleanupNeeded.set(cleanup);
- }
+ compact();
logAndClear(segmentWriter.getNodeWriteTimeStats(),
segmentWriter.getNodeCompactTimeStats());
log(segmentWriter.getNodeCacheOccupancyInfo());
} else {
@@ -736,11 +729,6 @@ public class FileStore implements Segmen
}
});
- if (cleanupNeeded.getAndSet(false)) {
- // FIXME OAK-4138: Decouple revision cleanup from the flush thread
- fileReaper.add(cleanup());
- }
-
fileReaper.reap();
}
@@ -1024,25 +1012,43 @@ public class FileStore implements Segmen
}
if (success) {
+ gcScheduler.execute(format("TarMK cleanup [%s]", directory),
new Runnable() {
+ @Override
+ public void run() {
+ try {
+ fileReaper.add(cleanup());
+ } catch (IOException e) {
+ gcListener.error("TarMK GC #" + GC_COUNT + ":
cleanup failed", e);
+ }
+ }
+ });
+
gcListener.compacted(SUCCESS, newGeneration);
gcListener.info("TarMK GC #{}: compaction succeeded in {} ({}
ms), after {} cycles",
GC_COUNT, watch, watch.elapsed(MILLISECONDS), cycles);
return true;
} else {
- gcListener.info("TarMK GC #{}: cleaning up after failed
compaction", GC_COUNT);
-
- Predicate<Integer> cleanupPredicate = new Predicate<Integer>()
{
+ gcScheduler.execute(format("TarMK cleanup [%s]", directory),
new Runnable() {
@Override
- public boolean apply(Integer generation) {
- return generation == newGeneration;
+ public void run() {
+ try {
+ gcListener.info("TarMK GC #{}: cleaning up after
failed compaction", GC_COUNT);
+ Predicate<Integer> cleanupPredicate = new
Predicate<Integer>() {
+ @Override
+ public boolean apply(Integer generation) {
+ return generation == newGeneration;
+ }
+ };
+ fileReaper.add(cleanup(cleanupPredicate,
+ "gc-count=" + GC_COUNT +
+ ",gc-status=failed" +
+ ",store-generation=" + (newGeneration - 1) +
+ ",reclaim-predicate=(generation==" +
newGeneration + ")"));
+ } catch (IOException e) {
+ gcListener.error("TarMK GC #" + GC_COUNT + ":
cleanup failed", e);
+ }
}
- };
- fileReaper.add(cleanup(cleanupPredicate,
- "gc-count=" + GC_COUNT +
- ",gc-status=failed" +
- ",store-generation=" + (newGeneration - 1) +
- ",reclaim-predicate=(generation==" + newGeneration + ")"));
-
+ });
gcListener.compacted(FAILURE, newGeneration);
gcListener.info("TarMK GC #{}: compaction failed after {} ({}
ms), and {} cycles",
GC_COUNT, watch, watch.elapsed(MILLISECONDS), cycles);
@@ -1442,7 +1448,7 @@ public class FileStore implements Segmen
* Trigger a garbage collection cycle
*/
public void gc() {
- gcScheduler.execute(format("TarMK compaction thread [%s]", directory),
new Runnable() {
+ gcScheduler.execute(format("TarMK compaction [%s]", directory), new
Runnable() {
@Override
public void run() {
try {
