Author: catholicon
Date: Fri Aug 2 05:24:31 2019
New Revision: 1864197
URL: http://svn.apache.org/viewvc?rev=1864197&view=rev
Log:
OAK-8513: Concurrent index access via CopyOnRead directory can lead to reading
directly off of remote
Modified:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/CopyOnReadDirectory.java
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/LocalIndexFile.java
jackrabbit/oak/trunk/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ConcurrentCopyOnReadDirectoryTest.java
Modified:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java?rev=1864197&r1=1864196&r2=1864197&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java
(original)
+++
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java
Fri Aug 2 05:24:31 2019
@@ -28,6 +28,7 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -43,6 +44,7 @@ import javax.management.openmbean.Tabula
import com.google.common.base.Function;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.Monitor;
import org.apache.commons.io.FileUtils;
import
org.apache.jackrabbit.oak.plugins.index.lucene.directory.CopyOnReadDirectory;
import
org.apache.jackrabbit.oak.plugins.index.lucene.directory.CopyOnWriteDirectory;
@@ -98,6 +100,7 @@ public class IndexCopier implements Copy
private final AtomicLong downloadTime = new AtomicLong();
private final AtomicLong uploadTime = new AtomicLong();
+ private final Monitor copyCompletionMonitor = new Monitor();
private final Map<String, String> indexPathVersionMapping =
newConcurrentMap();
private final ConcurrentMap<String, LocalIndexFile> failedToDeleteFiles =
newConcurrentMap();
@@ -356,8 +359,61 @@ public class IndexCopier implements Copy
return copyInProgressFiles.contains(file);
}
+ /**
+ * Waits for maximum of {@code timeoutMillis} while checking if {@code
file} isn't being copied already.
+ * The method can return before {@code timeoutMillis} if it got
interrupted. So, if the method reports false,
+ * caller should apply its own logic to see if a wait should be repeated
or not.
+ * @param file
+ * @param timeoutMillis
+ */
+ public void waitForCopyCompletion(LocalIndexFile file, long timeoutMillis)
{
+ final Monitor.Guard notCopyingGuard = new
Monitor.Guard(copyCompletionMonitor) {
+ @Override
+ public boolean isSatisfied() {
+ return !isCopyInProgress(file);
+ }
+ };
+ long localLength = file.actualSize();
+ long lastLocalLength = localLength;
+
+ boolean notCopying = !isCopyInProgress(file);
+ while (!notCopying) {
+ try {
+ if (log.isDebugEnabled()) {
+ log.debug("Checking for copy completion of {} - {}",
file.getKey(), file.copyLog());
+ }
+ notCopying = copyCompletionMonitor.enterWhen(notCopyingGuard,
timeoutMillis, TimeUnit.MILLISECONDS);
+ if (notCopying) {
+ copyCompletionMonitor.leave();
+ }
+ } catch (InterruptedException e) {
+ // ignore
+ }
+
+ localLength = file.actualSize();
+
+ // Break out if local file length hasn't changed since last we
checked.
+ // Do note that our assumption is that our monitor would return
false only on timeout.
+ // BUT that's not true and the monitor could be interrupted as
well. We are ignoring that
+ // gotcha explicitly as we don't really want to over complicate
here for a rare race of
+ // concurrent copying and avoid reading from remote.
+ if (localLength <= lastLocalLength) {
+ log.warn("Breaking out of waiting for copy to finish as
current local length ({})" +
+ " hasn't increased from {}",
+ localLength, lastLocalLength);
+ break;
+ }
+ lastLocalLength = localLength;
+ }
+ }
+
public void doneCopy(LocalIndexFile file, long start) {
- copyInProgressFiles.remove(file);
+ copyCompletionMonitor.enter();
+ try {
+ copyInProgressFiles.remove(file);
+ } finally {
+ copyCompletionMonitor.leave();
+ }
copyInProgressCount.decrementAndGet();
copyInProgressSize.addAndGet(-file.getSize());
Modified:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/CopyOnReadDirectory.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/CopyOnReadDirectory.java?rev=1864197&r1=1864196&r2=1864197&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/CopyOnReadDirectory.java
(original)
+++
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/CopyOnReadDirectory.java
Fri Aug 2 05:24:31 2019
@@ -67,6 +67,11 @@ public class CopyOnReadDirectory extends
private final Executor executor;
private final AtomicBoolean closed = new AtomicBoolean();
+ // exported as package private to be useful in tests
+ static final String WAIT_OTHER_COPY_SYSPROP_NAME = "cor.waitCopyMillis";
+
+ long waitOtherCopyTimeoutMillis =
Long.getLong(WAIT_OTHER_COPY_SYSPROP_NAME, TimeUnit.SECONDS.toMillis(30));
+
private final ConcurrentMap<String, CORFileReference> files =
newConcurrentMap();
public CopyOnReadDirectory(IndexCopier indexCopier, Directory remote,
Directory local, boolean prefetch,
@@ -212,19 +217,24 @@ public class CopyOnReadDirectory extends
name, humanReadableByteCount(fileSize));
}
} else {
- long localLength = local.fileLength(name);
long remoteLength = remote.fileLength(name);
+ LocalIndexFile file = new LocalIndexFile(local, name,
remoteLength, true);
+ // as a local file exists, attempt a wait for completion of
any potential ongoing concurrent copy
+ indexCopier.waitForCopyCompletion(file,
waitOtherCopyTimeoutMillis);
+
+ long localLength = local.fileLength(name);
+
//Do a simple consistency check. Ideally Lucene index files
are never
//updated but still do a check if the copy is consistent
if (localLength != remoteLength) {
- LocalIndexFile file = new LocalIndexFile(local, name,
remoteLength, true);
if (!indexCopier.isCopyInProgress(file)) {
log.warn("[{}] Found local copy for {} in {} but size
of local {} differs from remote {}. " +
"Content would be read from remote
file only",
indexPath, name, local, localLength,
remoteLength);
indexCopier.foundInvalidFile();
} else {
+
logRemoteAccess("[{}] Found in progress copy of file
{}. Would read from remote", indexPath, name);
}
} else {
Modified:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/LocalIndexFile.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/LocalIndexFile.java?rev=1864197&r1=1864196&r2=1864197&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/LocalIndexFile.java
(original)
+++
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/LocalIndexFile.java
Fri Aug 2 05:24:31 2019
@@ -111,7 +111,7 @@ public final class LocalIndexFile {
return actualSize() * 1.0f / size * 100;
}
- private long actualSize(){
+ public long actualSize(){
return dir != null ? new File(dir, name).length() : 0;
}
Modified:
jackrabbit/oak/trunk/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ConcurrentCopyOnReadDirectoryTest.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ConcurrentCopyOnReadDirectoryTest.java?rev=1864197&r1=1864196&r2=1864197&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ConcurrentCopyOnReadDirectoryTest.java
(original)
+++
jackrabbit/oak/trunk/oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/ConcurrentCopyOnReadDirectoryTest.java
Fri Aug 2 05:24:31 2019
@@ -17,45 +17,79 @@
package org.apache.jackrabbit.oak.plugins.index.lucene.directory;
-import com.google.common.io.Closer;
+import com.google.common.collect.Iterables;
+import org.apache.commons.compress.utils.Lists;
import org.apache.jackrabbit.oak.InitialContentHelper;
import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.commons.junit.TemporarySystemProperty;
import org.apache.jackrabbit.oak.plugins.index.lucene.IndexCopier;
import org.apache.jackrabbit.oak.plugins.index.lucene.LuceneIndexDefinition;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.apache.lucene.store.*;
-import org.junit.*;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
-import java.util.concurrent.*;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import static
com.google.common.util.concurrent.MoreExecutors.sameThreadExecutor;
+import static
org.apache.jackrabbit.oak.plugins.index.lucene.directory.CopyOnReadDirectory.WAIT_OTHER_COPY_SYSPROP_NAME;
import static
org.apache.jackrabbit.oak.plugins.index.search.FulltextIndexConstants.INDEX_DATA_CHILD_NAME;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
public class ConcurrentCopyOnReadDirectoryTest {
@Rule
public TemporaryFolder temporaryFolder = new TemporaryFolder(new
File("target"));
- private final Closer closer = Closer.create();
- private ExecutorService executorService = Executors.newFixedThreadPool(2);
+ @Rule
+ public TemporarySystemProperty tempSysProp = new TemporarySystemProperty();
+
+ private ExecutorService executorService = null;
private Directory remote;
+ private IndexCopier copier;
+
+ private Directory firstCoR = null;
+ private List<Future<String>> leechingCoRFutures = Lists.newArrayList();
+ private List<Directory> leechingCoRs = Lists.newArrayList();
+
+ private CountDownLatch firstCoRBlocker;
+ private Future<String> firstCoRFutre;
+ private LuceneIndexDefinition defn;
+
+ private static final String REMOTE_INPUT_PREFIX = "Remote - ";
@Before
public void setup() throws Exception {
- executorService = Executors.newFixedThreadPool(2);
- closer.register(new ExecutorCloser(executorService, 1,
TimeUnit.SECONDS));
+ System.setProperty(WAIT_OTHER_COPY_SYSPROP_NAME,
String.valueOf(TimeUnit.MILLISECONDS.toMillis(30)));
+ executorService = Executors.newFixedThreadPool(5);
// normal remote directory
remote = new RAMDirectory() {
@Override
public IndexInput openInput(String name, IOContext context) throws
IOException {
- return new RemoteIndexInput(super.openInput(name, context));
+ IndexInput ret = spy(super.openInput(name, context));
+ when(ret.toString())
+ .thenAnswer(invocationOnMock -> REMOTE_INPUT_PREFIX +
invocationOnMock.callRealMethod());
+ return ret;
}
};
IndexOutput output = remote.createOutput("file", IOContext.DEFAULT);
@@ -64,86 +98,86 @@ public class ConcurrentCopyOnReadDirecto
IndexInput remoteInput = remote.openInput("file", IOContext.READ);
assertTrue(remoteInput.length() > 1);
+
+ copier = new IndexCopier(sameThreadExecutor(),
temporaryFolder.newFolder(), true);
+
+ NodeState root = InitialContentHelper.INITIAL_CONTENT;
+ defn = new LuceneIndexDefinition(root, root, "/foo");
}
@After
- public void tearDown() throws Exception {
- closer.close();
+ public void tearDown() {
+ // This is no-op usually but would save us in case first CoR is stuck
in wait
+ firstCoRBlocker.countDown();
+
+ if (executorService != null) {
+ new ExecutorCloser(executorService, 1, TimeUnit.SECONDS).close();
+ }
}
- @Ignore("OAK-8513")
@Test
public void concurrentPrefetch() throws Exception {
- // create filtering remote that would block on opening an input
- CountDownLatch copyWaiter = new CountDownLatch(1);
- CountDownLatch copyBlocker = new CountDownLatch(1);
- Directory blockingRemote = new BlockingInputDirectory(copyWaiter,
copyBlocker, remote);
-
- IndexCopier copier = new IndexCopier(sameThreadExecutor(),
temporaryFolder.newFolder(), true);
-
- NodeState root = InitialContentHelper.INITIAL_CONTENT;
- final LuceneIndexDefinition defn = new LuceneIndexDefinition(root,
root, "/foo");
-
- // create a CoR instance to start pre-fetching in a separate thread as
we'd be blocking it
- Future<String> concCoR = executorService.submit(() -> {
- try {
- openCoR(copier, blockingRemote, defn);
- return null;
- } catch (Throwable t) {
- return getThrowableAsString(t);
- }
- });
+ // setup one primary CoR and 2 subsequent ones to read. Each would run
concurrently.
+ setupCopiers(2);
+ // let of go of CoR1 to finish its work
+ firstCoRBlocker.countDown();
+
+ assertNull("First CoR must not throw exception", firstCoRFutre.get());
+
+ waitForLeechingCoRsToFinish();
+
+ for (Directory d : Iterables.concat(Collections.singleton(firstCoR),
leechingCoRs)) {
+ IndexInput input = d.openInput("file", IOContext.READ);
+ assertFalse(d + " must not be reading from remote",
+ input.toString().startsWith(REMOTE_INPUT_PREFIX));
+ }
+ }
- try {
- // wait for CoR to start fetching which we're blocking its
completion via copyBlocker latch
- copyWaiter.await();
+ @Test
+ public void concurrentPrefetchWithTimeout() throws Exception {
+ // setup one primary CoR and 2 subsequent ones to read. Each would run
concurrently.
+ setupCopiers(2);
- // get another directory instance with normal remote while the
previous is blocked by us
- Directory dir = openCoR(copier, remote, defn);
- IndexInput input = dir.openInput("file", IOContext.READ);
+ // don't unblock firstCor so that leeching CoRs time out
+ waitForLeechingCoRsToFinish();
- copyBlocker.countDown();
+ // let it go now as leeching CoRs have finished
+ firstCoRBlocker.countDown();
- String futureException = concCoR.get();
+ assertNull("First CoR must not throw exception", firstCoRFutre.get());
- assertNull("Concurrent CoR must not throw exception",
futureException);
+ IndexInput input = firstCoR.openInput("file", IOContext.READ);
+ assertFalse(firstCoR + " must not be reading from remote",
+ input.toString().startsWith(REMOTE_INPUT_PREFIX));
- assertFalse("Must not be reading from remote", input instanceof
RemoteIndexInput);
- } finally {
- copyBlocker.countDown();
+ for (Directory d : leechingCoRs) {
+ input = d.openInput("file", IOContext.READ);
+ assertTrue(d + " must be reading from remote",
+ input.toString().startsWith(REMOTE_INPUT_PREFIX));
}
}
- private static Directory openCoR(IndexCopier copier, Directory remote,
LuceneIndexDefinition defn) throws IOException {
- return copier.wrapForRead("/oak:index/foo", defn, remote,
INDEX_DATA_CHILD_NAME);
- }
+ private void setupCopiers(int numLeechers) throws Exception {
+ // 1 thread each for leeching copier and another one for the first one
+ executorService = Executors.newFixedThreadPool(numLeechers + 1);
- private static String getThrowableAsString(Throwable t) {
- StringBuilder sb = new StringBuilder(t.getMessage() + "\n");
- StringWriter sw = new StringWriter();
- t.printStackTrace(new PrintWriter(sw));
- sb.append(sw.getBuffer());
- return sb.toString();
+ setupFirstCoR();
+ setupLeechingCoRs(numLeechers);
}
- static class BlockingInputDirectory extends FilterDirectory {
- private final CountDownLatch copyWaiter;
- private final CountDownLatch copyBlocker;
-
- BlockingInputDirectory(CountDownLatch copyWaiter, CountDownLatch
copyBlocker, Directory in) {
- super(in);
- this.copyWaiter = copyWaiter;
- this.copyBlocker = copyBlocker;
- }
+ private void setupFirstCoR() throws Exception {
+ firstCoRBlocker = new CountDownLatch(1);
+ CountDownLatch firstCoRWaiter = new CountDownLatch(1);
- @Override
- public IndexInput openInput(String name, IOContext context) throws
IOException {
+ // Create a blocking remote for CoR1 to signal how open input
progresses
+ Directory blockingRemote = spy(remote);
+ doAnswer(invocationOnMock -> {
IndexInput input;
try {
- input = super.openInput(name, context);
+ input = (IndexInput) invocationOnMock.callRealMethod();
} finally {
// signal that input has been opened
- copyWaiter.countDown();
+ firstCoRWaiter.countDown();
}
// wait while we're signalled that we can be done with opening
input
@@ -151,7 +185,7 @@ public class ConcurrentCopyOnReadDirecto
while (wait) {
try {
// block until we are signalled to call super
- copyBlocker.await();
+ firstCoRBlocker.await();
wait = false;
} catch (InterruptedException e) {
// ignore
@@ -159,45 +193,75 @@ public class ConcurrentCopyOnReadDirecto
}
return input;
- }
+ }).when(blockingRemote).openInput(any(), any());
+
+ // create CoR instance to start pre-fetching in a separate thread as
we want to block it mid-way
+ firstCoRFutre = executorService.submit(() -> {
+ try {
+ String description = "firstCoR";
+ Thread.currentThread().setName(description);
+ firstCoR = openCoR(copier, blockingRemote, defn, description);
+ return null;
+ } catch (Throwable t) {
+ return getThrowableAsString(t);
+ }
+ });
+
+ // wait for CoR to start fetching which we're blocking its completion
via cor1Blocker latch
+ firstCoRWaiter.await();
}
- static class RemoteIndexInput extends IndexInput {
- private final IndexInput delegate;
+ private void setupLeechingCoRs(int numLeechers) throws Exception {
+ CountDownLatch leechingCoRsWaiter = new CountDownLatch(numLeechers);
+ // Create a blocking copier for leeching CoRs to signal when it starts
to wait for it to wait for copy completion
+ IndexCopier blockingCopier = spy(copier);
+ doAnswer(invocationOnMock -> {
+ leechingCoRsWaiter.countDown();
+ return invocationOnMock.callRealMethod();
+ }).when(blockingCopier).isCopyInProgress(any());
- RemoteIndexInput(IndexInput delegate) {
- super("Remote " + delegate.toString());
- this.delegate = delegate;
+ for (int i = 0; i < numLeechers; i++) {
+ final String leecherName = "CoR-" + (i + 1);
+ leechingCoRFutures.add(executorService.submit(() ->
createLeechingCoR(blockingCopier, defn, leecherName)));
}
- @Override
- public void close() throws IOException {
- delegate.close();
- }
+ // wait for leeching CoRs to start
+ leechingCoRsWaiter.await();
+ }
- @Override
- public void seek(long pos) throws IOException {
- delegate.seek(pos);
- }
+ private String createLeechingCoR(IndexCopier blockingCopier,
LuceneIndexDefinition defn, String threadName) {
+ Thread.currentThread().setName(threadName);
- @Override
- public long length() {
- return delegate.length();
- }
+ // get another directory instance with normal remote while the
previous is blocked by us
+ try {
+ CopyOnReadDirectory dir = (CopyOnReadDirectory)
openCoR(blockingCopier, remote, defn, threadName);
+ leechingCoRs.add(dir);
- @Override
- public long getFilePointer() {
- return delegate.getFilePointer();
+ return null;
+ } catch (Throwable t) {
+ return getThrowableAsString(t);
}
+ }
- @Override
- public byte readByte() throws IOException {
- return delegate.readByte();
+ private void waitForLeechingCoRsToFinish() throws Exception {
+ for (Future<String> corFuture : leechingCoRFutures) {
+ assertNull("Leeching CoR must not throw exception",
corFuture.get());
}
+ }
- @Override
- public void readBytes(byte[] b, int offset, int len) throws
IOException {
- delegate.readBytes(b, offset, len);
- }
+ private static Directory openCoR(IndexCopier copier, Directory remote,
LuceneIndexDefinition defn,
+ String description) throws IOException {
+ Directory d = spy(copier.wrapForRead("/oak:index/foo", defn, remote,
INDEX_DATA_CHILD_NAME));
+ when(d.toString())
+ .thenAnswer(invocationOnMock -> description);
+ return d;
+ }
+
+ private static String getThrowableAsString(Throwable t) {
+ StringBuilder sb = new StringBuilder(t.getMessage() + "\n");
+ StringWriter sw = new StringWriter();
+ t.printStackTrace(new PrintWriter(sw));
+ sb.append(sw.getBuffer());
+ return sb.toString();
}
}
\ No newline at end of file
