Author: frm
Date: Wed Nov 1 10:35:39 2017
New Revision: 1813959
URL: http://svn.apache.org/viewvc?rev=1813959&view=rev
Log:
OAK-6888 - Force data to be persisted to disk when flushing
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/Flusher.java
(with props)
Modified:
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/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=1813959&r1=1813958&r2=1813959&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
Wed Nov 1 10:35:39 2017
@@ -212,10 +212,9 @@ public class FileStore extends AbstractF
return;
}
try {
- flush();
+ maybeFlush();
} catch (IOException e) {
- log.warn("Failed to flush the TarMK at {}",
- directory, e);
+ log.warn("Failed to flush the TarMK at {}",
directory, e);
}
}
});
@@ -314,6 +313,24 @@ public class FileStore extends AbstractF
return stats;
}
+ private void doMaybeFlush() throws IOException {
+ if (revisions == null) {
+ log.debug("No TarRevisions available, skipping flush");
+ return;
+ }
+ revisions.maybeFlush(() -> {
+ segmentWriter.flush();
+ tarFiles.flush();
+ stats.flushed();
+ });
+ }
+
+ private void maybeFlush() throws IOException {
+ try (ShutDownCloser ignored = shutDown.keepAlive()) {
+ doMaybeFlush();
+ }
+ }
+
private void doFlush() throws IOException {
if (revisions == null) {
log.debug("No TarRevisions available, skipping flush");
@@ -323,7 +340,6 @@ public class FileStore extends AbstractF
segmentWriter.flush();
tarFiles.flush();
stats.flushed();
- return null;
});
}
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/Flusher.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/Flusher.java?rev=1813959&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/Flusher.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/Flusher.java
Wed Nov 1 10:35:39 2017
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.jackrabbit.oak.segment.file;
+
+import java.io.IOException;
+
+interface Flusher {
+
+ void flush() throws IOException;
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/Flusher.java
------------------------------------------------------------------------------
svn:eol-style = native
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=1813959&r1=1813958&r2=1813959&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
Wed Nov 1 10:35:39 2017
@@ -21,8 +21,6 @@ package org.apache.jackrabbit.oak.segmen
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
-import static com.google.common.base.Throwables.propagate;
-import static com.google.common.base.Throwables.propagateIfInstanceOf;
import static java.lang.Long.MAX_VALUE;
import static java.util.concurrent.TimeUnit.DAYS;
import static
org.apache.jackrabbit.oak.segment.file.FileStoreUtil.findPersistedRecordId;
@@ -54,7 +52,7 @@ import org.slf4j.LoggerFactory;
/**
* This implementation of {@code Revisions} is backed by a
* {@link #JOURNAL_FILE_NAME journal} file where the current head is persisted
- * by calling {@link #flush(Callable)}.
+ * by calling {@link #maybeFlush(Callable)}.
* <p>
* The {@link #setHead(Function, Option...)} method supports a timeout
* {@link Option}, which can be retrieved through factory methods of this
class.
@@ -181,26 +179,45 @@ public class TarRevisions implements Rev
}
/**
- * Flush the id of the current head to the journal after a call to
- * {@code persisted}. This method does nothing and returns immediately if
- * called concurrently and a call is already in progress.
- * @param persisted call back for upstream dependencies to ensure
- * the current head state is actually persisted before
- * its id is written to the head state.
- * @throws IOException
+ * Flush the id of the current head to the journal after a call to {@code
+ * persisted}. Differently from {@link #maybeFlush(Flusher)}, this method
+ * does not return early if a concurrent call is in progress. Instead, it
+ * blocks the caller until the requested flush operation is performed.
+ *
+ * @param flusher call back for upstream dependencies to ensure the current
+ * head state is actually persisted before its id is written
+ * to the head state.
*/
- public void flush(@Nonnull Callable<Void> persisted) throws IOException {
+ void flush(Flusher flusher) throws IOException {
+ if (head.get() == null) {
+ LOG.debug("No head available, skipping flush");
+ return;
+ }
+ journalFileLock.lock();
+ try {
+ doFlush(flusher);
+ } finally {
+ journalFileLock.unlock();
+ }
+ }
+
+ /**
+ * Flush the id of the current head to the journal after a call to {@code
+ * persisted}. This method does nothing and returns immediately if called
+ * concurrently and a call is already in progress.
+ *
+ * @param flusher call back for upstream dependencies to ensure the current
+ * head state is actually persisted before its id is written
+ * to the head state.
+ */
+ void maybeFlush(Flusher flusher) throws IOException {
if (head.get() == null) {
LOG.debug("No head available, skipping flush");
return;
}
if (journalFileLock.tryLock()) {
try {
- if (journalFile == null) {
- LOG.debug("No journal file available, skipping flush");
- return;
- }
- doFlush(persisted);
+ doFlush(flusher);
} finally {
journalFileLock.unlock();
}
@@ -209,23 +226,22 @@ public class TarRevisions implements Rev
}
}
- private void doFlush(Callable<Void> persisted) throws IOException {
- try {
- RecordId before = persistedHead.get();
- RecordId after = getHead();
- if (after.equals(before)) {
- LOG.debug("Head state did not change, skipping flush");
- } else {
- persisted.call();
- LOG.debug("TarMK journal update {} -> {}", before, after);
- journalFile.writeBytes(after.toString10() + " root " +
System.currentTimeMillis() + "\n");
- journalFile.getChannel().force(false);
- persistedHead.set(after);
- }
- } catch (Exception e) {
- propagateIfInstanceOf(e, IOException.class);
- propagate(e);
+ private void doFlush(Flusher flusher) throws IOException {
+ if (journalFile == null) {
+ LOG.debug("No journal file available, skipping flush");
+ return;
+ }
+ RecordId before = persistedHead.get();
+ RecordId after = getHead();
+ if (after.equals(before)) {
+ LOG.debug("Head state did not change, skipping flush");
+ return;
}
+ flusher.flush();
+ LOG.debug("TarMK journal update {} -> {}", before, after);
+ journalFile.writeBytes(after.toString10() + " root " +
System.currentTimeMillis() + "\n");
+ journalFile.getChannel().force(false);
+ persistedHead.set(after);
}
@Nonnull
