diff --git oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java index 307d2c6..67201ea 100644 --- oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java +++ oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopier.java @@ -120,7 +120,7 @@ public class IndexCopier implements CopyOnReadStatsMBean { public Directory wrapForRead(String indexPath, IndexDefinition definition, Directory remote) throws IOException { Directory local = createLocalDirForIndexReader(indexPath, definition); - return new CopyOnReadDirectory(remote, local); + return new CopyOnReadDirectory(remote, local, definition.isPrefetchEnabledForCopyOnRead()); } public Directory wrapForWrite(IndexDefinition definition, Directory remote, boolean reindexMode) throws IOException { @@ -249,11 +249,14 @@ public class IndexCopier implements CopyOnReadStatsMBean { */ private final Set localFileNames = Sets.newConcurrentHashSet(); - public CopyOnReadDirectory(Directory remote, Directory local) throws IOException { + public CopyOnReadDirectory(Directory remote, Directory local, boolean prefetch) throws IOException { super(remote); this.remote = remote; this.local = local; this.localFileNames.addAll(Arrays.asList(local.listAll())); + if (prefetch){ + prefetchIndexFiles(); + } } @Override @@ -302,55 +305,80 @@ public class IndexCopier implements CopyOnReadStatsMBean { executor.execute(new Runnable() { @Override public void run() { - String name = reference.name; - boolean success = false; - boolean copyAttempted = false; + scheduledForCopyCount.decrementAndGet(); + copyFilesToLocal(reference); + } + }); + } + + private void prefetchIndexFiles() throws IOException { + long start = PERF_LOGGER.start(); + long totalSize = 0; + int copyCount = 0; + for (String name : remote.listAll()) { + if (REMOTE_ONLY.contains(name)) { + continue; + } + CORFileReference fileRef = new CORFileReference(name); + files.putIfAbsent(name, fileRef); + long fileSize = copyFilesToLocal(fileRef); + if (fileSize > 0) { + copyCount++; + totalSize += fileSize; + } + } + PERF_LOGGER.end(start, -1, "Copied {} files totaling {}", copyCount, humanReadableByteCount(totalSize)); + } + + private long copyFilesToLocal(CORFileReference reference) { + String name = reference.name; + boolean success = false; + boolean copyAttempted = false; + long fileSize = 0; + try { + if (!local.fileExists(name)) { + fileSize = remote.fileLength(name); + LocalIndexFile file = new LocalIndexFile(local, name, fileSize, true); + long start = startCopy(file); + copyAttempted = true; + + remote.copy(local, name, name, IOContext.READ); + reference.markValid(); + + doneCopy(file, start); + } else { + long localLength = local.fileLength(name); + long remoteLength = remote.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) { + log.warn("Found local copy for {} in {} but size of local {} differs from remote {}. " + + "Content would be read from remote file only", + name, local, localLength, remoteLength); + invalidFileCount.incrementAndGet(); + } else { + reference.markValid(); + } + } + success = true; + } catch (IOException e) { + //TODO In case of exception there would not be any other attempt + //to download the file. Look into support for retry + log.warn("Error occurred while copying file [{}] " + + "from {} to {}", name, remote, local, e); + } finally { + if (copyAttempted && !success){ try { - scheduledForCopyCount.decrementAndGet(); - if (!local.fileExists(name)) { - long fileSize = remote.fileLength(name); - LocalIndexFile file = new LocalIndexFile(local, name, fileSize, true); - long start = startCopy(file); - copyAttempted = true; - - remote.copy(local, name, name, IOContext.READ); - reference.markValid(); - - doneCopy(file, start); - } else { - long localLength = local.fileLength(name); - long remoteLength = remote.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) { - log.warn("Found local copy for {} in {} but size of local {} differs from remote {}. " + - "Content would be read from remote file only", - name, local, localLength, remoteLength); - invalidFileCount.incrementAndGet(); - } else { - reference.markValid(); - } + if (local.fileExists(name)) { + local.deleteFile(name); } - success = true; } catch (IOException e) { - //TODO In case of exception there would not be any other attempt - //to download the file. Look into support for retry - log.warn("Error occurred while copying file [{}] " + - "from {} to {}", name, remote, local, e); - } finally { - if (copyAttempted && !success){ - try { - if (local.fileExists(name)) { - local.deleteFile(name); - } - } catch (IOException e) { - log.warn("Error occurred while deleting corrupted file [{}] from [{}]", name, local, e); - } - } + log.warn("Error occurred while deleting corrupted file [{}] from [{}]", name, local, e); } } - }); + } + return fileSize; } /** diff --git oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexDefinition.java oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexDefinition.java index 72a1b78..2f56a88 100644 --- oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexDefinition.java +++ oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexDefinition.java @@ -209,6 +209,8 @@ class IndexDefinition implements Aggregate.AggregateMapper{ private final boolean saveDirListing; + private final boolean prefetchEnabled; + public IndexDefinition(NodeState root, NodeState defn) { this(root, defn, null); } @@ -273,6 +275,7 @@ class IndexDefinition implements Aggregate.AggregateMapper{ this.pathFilter = PathFilter.from(new ReadOnlyBuilder(defn)); this.queryPaths = getQueryPaths(defn); this.saveDirListing = getOptionalValue(defn, LuceneIndexConstants.SAVE_DIR_LISTING, true); + this.prefetchEnabled = getOptionalValue(defn, LuceneIndexConstants.COR_PREFETCH_FILES, false); } public boolean isFullTextEnabled() { @@ -378,6 +381,10 @@ class IndexDefinition implements Aggregate.AggregateMapper{ return saveDirListing; } + public boolean isPrefetchEnabledForCopyOnRead() { + return prefetchEnabled; + } + public PathFilter getPathFilter() { return pathFilter; } diff --git oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexConstants.java oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexConstants.java index b60431d..19acff8 100644 --- oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexConstants.java +++ oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexConstants.java @@ -270,4 +270,10 @@ public interface LuceneIndexConstants { * existing index files */ String INDEX_PATH = "indexPath"; + + /** + * Allow pre fetching the index files upon opening the directory when CopyOnRead is + * enabled + */ + String COR_PREFETCH_FILES = "enablePrefetchIndexFiles"; } diff --git oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierTest.java oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierTest.java index 6bb26d3..609186d 100644 --- oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierTest.java +++ oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/IndexCopierTest.java @@ -111,6 +111,34 @@ public class IndexCopierTest { } @Test + public void basicTestWithPrefetch() throws Exception{ + Directory baseDir = new RAMDirectory(); + builder.setProperty(LuceneIndexConstants.COR_PREFETCH_FILES, true); + IndexDefinition defn = new IndexDefinition(root, builder.getNodeState()); + IndexCopier c1 = new RAMIndexCopier(baseDir, sameThreadExecutor(), getWorkDir()); + + Directory remote = new RAMDirectory(); + + byte[] t1 = writeFile(remote , "t1"); + byte[] t2 = writeFile(remote , "t2"); + + Directory wrapped = c1.wrapForRead("/foo", defn, remote); + assertEquals(2, wrapped.listAll().length); + + assertTrue(wrapped.fileExists("t1")); + assertTrue(wrapped.fileExists("t2")); + + assertTrue(baseDir.fileExists("t1")); + assertTrue(baseDir.fileExists("t2")); + + assertEquals(t1.length, wrapped.fileLength("t1")); + assertEquals(t2.length, wrapped.fileLength("t2")); + + readAndAssert(wrapped, "t1", t1); + + } + + @Test public void basicTestWithFS() throws Exception{ IndexDefinition defn = new IndexDefinition(root, builder.getNodeState()); IndexCopier c1 = new IndexCopier(sameThreadExecutor(), getWorkDir());