Author: mduerig
Date: Thu Sep 8 12:11:40 2016
New Revision: 1759793
URL: http://svn.apache.org/viewvc?rev=1759793&view=rev
Log:
OAK-4015: Expedite commits from the compactor
Use a fair lock and add an option to expedite set head calls via using the
write lock instead of the read lock.
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentNodeStore.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/TarRevisions.java
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentNodeStore.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentNodeStore.java?rev=1759793&r1=1759792&r2=1759793&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentNodeStore.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentNodeStore.java
Thu Sep 8 12:11:40 2016
@@ -204,7 +204,6 @@ public class SegmentNodeStore implements
* of {@code c.call()} otherwise.
* @throws Exception
*/
- // FIXME OAK-4015: Expedite commits from the compactor
// FIXME OAK-4122: Replace the commit semaphore in the segment node store
with a scheduler
// Replace by usage of expeditable lock or commit scheduler
boolean locked(Callable<Boolean> c, long timeout, TimeUnit unit) throws
Exception {
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=1759793&r1=1759792&r2=1759793&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 Sep 8 12:11:40 2016
@@ -40,6 +40,7 @@ import static org.apache.jackrabbit.oak.
import static
org.apache.jackrabbit.oak.segment.SegmentWriterBuilder.segmentWriterBuilder;
import static org.apache.jackrabbit.oak.segment.file.GCListener.Status.FAILURE;
import static org.apache.jackrabbit.oak.segment.file.GCListener.Status.SUCCESS;
+import static
org.apache.jackrabbit.oak.segment.file.TarRevisions.EXPEDITE_OPTION;
import static org.apache.jackrabbit.oak.segment.file.TarRevisions.timeout;
import java.io.Closeable;
@@ -1011,7 +1012,7 @@ public class FileStore implements Segmen
int cycles = 0;
boolean success = false;
while (cycles < gcOptions.getRetryCount() &&
- !(success = revisions.setHead(before.getRecordId(),
after.getRecordId()))) {
+ !(success = revisions.setHead(before.getRecordId(),
after.getRecordId(), EXPEDITE_OPTION))) {
// Some other concurrent changes have been made.
// Rebase (and compact) those changes on top of the
// compacted state before retrying to set the head.
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/TarRevisions.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/TarRevisions.java?rev=1759793&r1=1759792&r2=1759793&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/TarRevisions.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/TarRevisions.java
Thu Sep 8 12:11:40 2016
@@ -82,10 +82,8 @@ public class TarRevisions implements Rev
@Nonnull
private final AtomicReference<RecordId> persistedHead;
- // FIXME OAK-4015: Expedite commits from the compactor
- // use a lock that can expedite important commits like compaction and
checkpoints.
@Nonnull
- private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock(true);
private static class TimeOutOption implements Option {
private final long time;
@@ -109,6 +107,17 @@ public class TarRevisions implements Rev
}
/**
+ * Option to cause set head calls to be expedited. That is, cause them to
skip the queue
+ * of any other callers waiting to complete that don't have this option
specified.
+ */
+ public static final Option EXPEDITE_OPTION = new Option() {
+ @Override
+ public String toString() {
+ return "Expedite Option";
+ }
+ };
+
+ /**
* Timeout option approximating no time out ({@code Long.MAX_VALUE} days).
*/
public static final Option INFINITY = new TimeOutOption(MAX_VALUE, DAYS);
@@ -225,8 +234,10 @@ public class TarRevisions implements Rev
* This implementation blocks if a concurrent call to
* {@link #setHead(Function, Option...)} is already in
* progress.
-
- * @param options none
+ *
+ * @param options zero or one expedite option for expediting this call
+ * @throws IllegalArgumentException on any non recognised {@code option}.
+ * @see #EXPEDITE_OPTION
*/
@Override
public boolean setHead(
@@ -234,12 +245,19 @@ public class TarRevisions implements Rev
@Nonnull RecordId head,
@Nonnull Option... options) {
checkBound();
- rwLock.readLock().lock();
+
+ // If the expedite option was specified we acquire the write lock
instead of the read lock.
+ // This will cause this thread to get the lock before all threads
currently waiting to
+ // enter the read lock. See also the class comment of ReadWriteLock.
+ Lock lock = isExpedited(options)
+ ? rwLock.writeLock()
+ : rwLock.readLock();
+ lock.lock();
try {
RecordId id = this.head.get();
return id.equals(expected) && this.head.compareAndSet(id, head);
} finally {
- rwLock.readLock().unlock();
+ lock.unlock();
}
}
@@ -280,6 +298,16 @@ public class TarRevisions implements Rev
}
}
+ private static boolean isExpedited(Option[] options) {
+ if (options.length == 0) {
+ return false;
+ } else if (options.length == 1) {
+ return options[0] == EXPEDITE_OPTION;
+ } else {
+ throw new IllegalArgumentException("Expected zero or one options,
got " + options.length);
+ }
+ }
+
@Nonnull
private static TimeOutOption getTimeout(@Nonnull Option[] options) {
if (options.length == 0) {
