trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/directory/CopyOnReadDirectory.java - jackrabbit-oak - Git at Google

 /*
  * 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.plugins.index.lucene.directory;

 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;

 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import org.apache.jackrabbit.oak.commons.PerfLogger;
 import org.apache.jackrabbit.oak.plugins.index.lucene.IndexCopier;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FilterDirectory;
 import org.apache.lucene.store.IOContext;
 import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.IndexOutput;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import static com.google.common.collect.Maps.newConcurrentMap;
 import static java.util.Arrays.stream;
 import static org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCount;

 /**
  * Directory implementation which lazily copies the index files from a
  * remote directory in background.
  */
 public class CopyOnReadDirectory extends FilterDirectory {
     private static final Logger log = LoggerFactory.getLogger(CopyOnReadDirectory.class);
     private static final PerfLogger PERF_LOGGER = new PerfLogger(LoggerFactory.getLogger(log.getName() + ".perf"));

     public static final String DELETE_MARGIN_MILLIS_NAME = "oak.lucene.delete.margin";
     public final long DELETE_MARGIN_MILLIS = Long.getLong(DELETE_MARGIN_MILLIS_NAME, TimeUnit.MINUTES.toMillis(5));

     private final IndexCopier indexCopier;
     private final Directory remote;
     private final Directory local;
     private final String indexPath;
     private final Executor executor;
     private final AtomicBoolean closed = new AtomicBoolean();

     private final ConcurrentMap<String, CORFileReference> files = newConcurrentMap();

     public CopyOnReadDirectory(IndexCopier indexCopier, Directory remote, Directory local, boolean prefetch,
                                String indexPath, Executor executor) throws IOException {
         super(remote);
         this.indexCopier = indexCopier;
         this.executor = executor;
         this.remote = remote;
         this.local = local;
         this.indexPath = indexPath;

         if (prefetch) {
             prefetchIndexFiles();
         }
     }

     @Override
     public void deleteFile(String name) throws IOException {
         throw new UnsupportedOperationException("Cannot delete in a ReadOnly directory");
     }

     @Override
     public IndexOutput createOutput(String name, IOContext context) throws IOException {
         throw new UnsupportedOperationException("Cannot write in a ReadOnly directory");
     }

     @Override
     public IndexInput openInput(String name, IOContext context) throws IOException {
         if (IndexCopier.REMOTE_ONLY.contains(name)) {
             log.trace("[{}] opening remote only file {}", indexPath, name);
             return remote.openInput(name, context);
         }

         CORFileReference ref = files.get(name);
         if (ref != null) {
             if (ref.isLocalValid()) {
                 log.trace("[{}] opening existing local file {}", indexPath, name);
                 return files.get(name).openLocalInput(context);
             } else {
                 indexCopier.readFromRemote(true);
                 log.trace(
                         "[{}] opening existing remote file as local version is not valid {}",
                         indexPath, name);
                 return remote.openInput(name, context);
             }
         }

         //If file does not exist then just delegate to remote and not
         //schedule a copy task
         if (!remote.fileExists(name)){
             if (log.isDebugEnabled()) {
                 log.debug("[{}] Looking for non existent file {}. Current known files {}",
                         indexPath, name, Arrays.toString(remote.listAll()));
             }
             return remote.openInput(name, context);
         }

         CORFileReference toPut = new CORFileReference(name);
         CORFileReference old = files.putIfAbsent(name, toPut);
         if (old == null) {
             log.trace("[{}] scheduled local copy for {}", indexPath, name);
             copy(toPut);
         }

         //If immediate executor is used the result would be ready right away
         if (toPut.isLocalValid()) {
             log.trace("[{}] opening new local file {}", indexPath, name);
             return toPut.openLocalInput(context);
         }

         log.trace("[{}] opening new remote file {}", indexPath, name);
         indexCopier.readFromRemote(true);
         return remote.openInput(name, context);
     }

     public Directory getLocal() {
         return local;
     }

     private void copy(final CORFileReference reference) {
         indexCopier.scheduledForCopy();
         executor.execute(new Runnable() {
             @Override
             public void run() {
                 indexCopier.copyDone();
                 copyFilesToLocal(reference, true, true);
             }
         });
     }

     private void prefetchIndexFiles() throws IOException {
         long start = PERF_LOGGER.start();
         long totalSize = 0;
         int copyCount = 0;
         List<String> copiedFileNames = Lists.newArrayList();
         for (String name : remote.listAll()) {
             if (IndexCopier.REMOTE_ONLY.contains(name)) {
                 continue;
             }
             CORFileReference fileRef = new CORFileReference(name);
             files.putIfAbsent(name, fileRef);
             long fileSize = copyFilesToLocal(fileRef, false, false);
             if (fileSize > 0) {
                 copyCount++;
                 totalSize += fileSize;
                 copiedFileNames.add(name);
             }
         }

         local.sync(copiedFileNames);
         PERF_LOGGER.end(start, -1, "[{}] Copied {} files totaling {}", indexPath, copyCount, humanReadableByteCount(totalSize));
     }

     private long copyFilesToLocal(CORFileReference reference, boolean sync, boolean logDuration) {
         String name = reference.name;
         boolean success = false;
         boolean copyAttempted = false;
         long fileSize = 0;
         try {
             if (!local.fileExists(name)) {
                 long perfStart = -1;
                 if (logDuration) {
                     perfStart = PERF_LOGGER.start();
                 }

                 fileSize = remote.fileLength(name);
                 LocalIndexFile file = new LocalIndexFile(local, name, fileSize, true);
                 long start = indexCopier.startCopy(file);
                 copyAttempted = true;

                 remote.copy(local, name, name, IOContext.READ);
                 reference.markValid();

                 if (sync) {
                     local.sync(Collections.singleton(name));
                 }

                 indexCopier.doneCopy(file, start);
                 if (logDuration) {
                     PERF_LOGGER.end(perfStart, 0,
                             "[{}] Copied file {} of size {}", indexPath,
                             name, humanReadableByteCount(fileSize));
                 }
             } 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) {
                     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 {
                         log.trace("[{}] Found in progress copy of file {}. Would read from remote", indexPath, name);
                     }
                 } else {
                     reference.markValid();
                     log.trace("[{}] found local copy of file {}",
                             indexPath, 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 {}", indexPath, 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 [{}]", indexPath, name, local, e);
                 }
             }
         }
         return fileSize;
     }

     /**
      * On close file which are not present in remote are removed from local.
      * CopyOnReadDir is opened at different revisions of the index state
      *
      * CDir1 - V1
      * CDir2 - V2
      *
      * Its possible that two different IndexSearcher are opened at same local
      * directory but pinned to different revisions. So while removing it must
      * be ensured that any currently opened IndexSearcher does not get affected.
      * The way IndexSearchers get created in IndexTracker it ensures that new searcher
      * pinned to newer revision gets opened first and then existing ones are closed.
      *
      *
      */
     @Override
     public void close() throws IOException {
         if (!closed.compareAndSet(false, true)){
             return;
         }
         //Always remove old index file on close as it ensures that
         //no other IndexSearcher are opened with previous revision of Index due to
         //way IndexTracker closes LuceneIndexNode. At max there would be only two LuceneIndexNode
         //opened pinned to different revision of same Lucene index
         executor.execute(new Runnable() {
             @Override
             public void run() {
                 try{
                     removeDeletedFiles();
                 } catch (IOException e) {
                     log.warn(
                             "[{}] Error occurred while removing deleted files from Local {}, Remote {}",
                             indexPath, local, remote, e);
                 }

                 try {
                     //This would also remove old index files if current
                     //directory was based on newerRevision as local would
                     //be of type DeleteOldDirOnClose
                     local.close();
                     remote.close();
                 } catch (IOException e) {
                     log.warn(
                             "[{}] Error occurred while closing directory ",
                             indexPath, e);
                 }
             }
         });
     }

     @Override
     public String toString() {
         return String.format("[COR] Local %s, Remote %s", local, remote);
     }

     private void removeDeletedFiles() throws IOException {
         Set<String> remoteFiles = stream(remote.listAll())
                 .filter(name -> !IndexCopier.REMOTE_ONLY.contains(name))
                 .collect(Collectors.toSet());

         long maxTS = IndexCopier.getNewestLocalFSTimestampFor(remoteFiles, local);
         if (maxTS == -1) {
             log.warn("Couldn't compute safe timestamp to delete files from {}", local);
             return;
         }

         // subtract DELETE_MARGIN_MILLIS from maxTS for safety (you can never be too careful with time)
         final long deleteBeforeTS = maxTS - DELETE_MARGIN_MILLIS;

         Set<String> filesToBeDeleted =
                 // Files present locally
                 ImmutableSet.copyOf(local.listAll()).stream()
                 // but not in my view
                 .filter(name -> !remoteFiles.contains(name))
                 // and also older than a safe timestamp (deleteBeforeTS)
                 .filter(name -> IndexCopier.isFileModifiedBefore(name, local, deleteBeforeTS))
                 // can be deleted
                 .collect(Collectors.toSet())
         ;

         Set<String> failedToDelete = Sets.newHashSet();

         for (String fileName : filesToBeDeleted) {
             boolean deleted = indexCopier.deleteFile(local, fileName, true);
             if (!deleted){
                 failedToDelete.add(fileName);
             }
         }

         filesToBeDeleted = new HashSet<String>(filesToBeDeleted);
         filesToBeDeleted.removeAll(failedToDelete);
         if(!filesToBeDeleted.isEmpty()) {
             log.debug(
                     "[{}] Following files have been removed from Lucene index directory {}",
                     indexPath, filesToBeDeleted);
         }
     }

     private class CORFileReference {
         final String name;
         private volatile boolean valid;

         private CORFileReference(String name) {
             this.name = name;
         }

         boolean isLocalValid(){
             return valid;
         }

         IndexInput openLocalInput( IOContext context) throws IOException {
             indexCopier.readFromLocal(true);
             return local.openInput(name, context);
         }

         void markValid(){
             this.valid = true;
         }
     }
 }
	/*
	* 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.plugins.index.lucene.directory;

	import java.io.IOException;
	import java.util.Arrays;
	import java.util.Collections;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Set;
	import java.util.concurrent.ConcurrentMap;
	import java.util.concurrent.Executor;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.atomic.AtomicBoolean;
	import java.util.stream.Collectors;

	import com.google.common.collect.ImmutableSet;
	import com.google.common.collect.Lists;
	import com.google.common.collect.Sets;
	import org.apache.jackrabbit.oak.commons.PerfLogger;
	import org.apache.jackrabbit.oak.plugins.index.lucene.IndexCopier;
	import org.apache.lucene.store.Directory;
	import org.apache.lucene.store.FilterDirectory;
	import org.apache.lucene.store.IOContext;
	import org.apache.lucene.store.IndexInput;
	import org.apache.lucene.store.IndexOutput;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import static com.google.common.collect.Maps.newConcurrentMap;
	import static java.util.Arrays.stream;
	import static org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCount;

	/**
	* Directory implementation which lazily copies the index files from a
	* remote directory in background.
	*/
	public class CopyOnReadDirectory extends FilterDirectory {
	private static final Logger log = LoggerFactory.getLogger(CopyOnReadDirectory.class);
	private static final PerfLogger PERF_LOGGER = new PerfLogger(LoggerFactory.getLogger(log.getName() + ".perf"));

	public static final String DELETE_MARGIN_MILLIS_NAME = "oak.lucene.delete.margin";
	public final long DELETE_MARGIN_MILLIS = Long.getLong(DELETE_MARGIN_MILLIS_NAME, TimeUnit.MINUTES.toMillis(5));

	private final IndexCopier indexCopier;
	private final Directory remote;
	private final Directory local;
	private final String indexPath;
	private final Executor executor;
	private final AtomicBoolean closed = new AtomicBoolean();

	private final ConcurrentMap<String, CORFileReference> files = newConcurrentMap();

	public CopyOnReadDirectory(IndexCopier indexCopier, Directory remote, Directory local, boolean prefetch,
	String indexPath, Executor executor) throws IOException {
	super(remote);
	this.indexCopier = indexCopier;
	this.executor = executor;
	this.remote = remote;
	this.local = local;
	this.indexPath = indexPath;

	if (prefetch) {
	prefetchIndexFiles();
	}
	}

	@Override
	public void deleteFile(String name) throws IOException {
	throw new UnsupportedOperationException("Cannot delete in a ReadOnly directory");
	}

	@Override
	public IndexOutput createOutput(String name, IOContext context) throws IOException {
	throw new UnsupportedOperationException("Cannot write in a ReadOnly directory");
	}

	@Override
	public IndexInput openInput(String name, IOContext context) throws IOException {
	if (IndexCopier.REMOTE_ONLY.contains(name)) {
	log.trace("[{}] opening remote only file {}", indexPath, name);
	return remote.openInput(name, context);
	}

	CORFileReference ref = files.get(name);
	if (ref != null) {
	if (ref.isLocalValid()) {
	log.trace("[{}] opening existing local file {}", indexPath, name);
	return files.get(name).openLocalInput(context);
	} else {
	indexCopier.readFromRemote(true);
	log.trace(
	"[{}] opening existing remote file as local version is not valid {}",
	indexPath, name);
	return remote.openInput(name, context);
	}
	}

	//If file does not exist then just delegate to remote and not
	//schedule a copy task
	if (!remote.fileExists(name)){
	if (log.isDebugEnabled()) {
	log.debug("[{}] Looking for non existent file {}. Current known files {}",
	indexPath, name, Arrays.toString(remote.listAll()));
	}
	return remote.openInput(name, context);
	}

	CORFileReference toPut = new CORFileReference(name);
	CORFileReference old = files.putIfAbsent(name, toPut);
	if (old == null) {
	log.trace("[{}] scheduled local copy for {}", indexPath, name);
	copy(toPut);
	}

	//If immediate executor is used the result would be ready right away
	if (toPut.isLocalValid()) {
	log.trace("[{}] opening new local file {}", indexPath, name);
	return toPut.openLocalInput(context);
	}

	log.trace("[{}] opening new remote file {}", indexPath, name);
	indexCopier.readFromRemote(true);
	return remote.openInput(name, context);
	}

	public Directory getLocal() {
	return local;
	}

	private void copy(final CORFileReference reference) {
	indexCopier.scheduledForCopy();
	executor.execute(new Runnable() {
	@Override
	public void run() {
	indexCopier.copyDone();
	copyFilesToLocal(reference, true, true);
	}
	});
	}

	private void prefetchIndexFiles() throws IOException {
	long start = PERF_LOGGER.start();
	long totalSize = 0;
	int copyCount = 0;
	List<String> copiedFileNames = Lists.newArrayList();
	for (String name : remote.listAll()) {
	if (IndexCopier.REMOTE_ONLY.contains(name)) {
	continue;
	}
	CORFileReference fileRef = new CORFileReference(name);
	files.putIfAbsent(name, fileRef);
	long fileSize = copyFilesToLocal(fileRef, false, false);
	if (fileSize > 0) {
	copyCount++;
	totalSize += fileSize;
	copiedFileNames.add(name);
	}
	}

	local.sync(copiedFileNames);
	PERF_LOGGER.end(start, -1, "[{}] Copied {} files totaling {}", indexPath, copyCount, humanReadableByteCount(totalSize));
	}

	private long copyFilesToLocal(CORFileReference reference, boolean sync, boolean logDuration) {
	String name = reference.name;
	boolean success = false;
	boolean copyAttempted = false;
	long fileSize = 0;
	try {
	if (!local.fileExists(name)) {
	long perfStart = -1;
	if (logDuration) {
	perfStart = PERF_LOGGER.start();
	}

	fileSize = remote.fileLength(name);
	LocalIndexFile file = new LocalIndexFile(local, name, fileSize, true);
	long start = indexCopier.startCopy(file);
	copyAttempted = true;

	remote.copy(local, name, name, IOContext.READ);
	reference.markValid();

	if (sync) {
	local.sync(Collections.singleton(name));
	}

	indexCopier.doneCopy(file, start);
	if (logDuration) {
	PERF_LOGGER.end(perfStart, 0,
	"[{}] Copied file {} of size {}", indexPath,
	name, humanReadableByteCount(fileSize));
	}
	} 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) {
	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 {
	log.trace("[{}] Found in progress copy of file {}. Would read from remote", indexPath, name);
	}
	} else {
	reference.markValid();
	log.trace("[{}] found local copy of file {}",
	indexPath, 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 {}", indexPath, 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 [{}]", indexPath, name, local, e);
	}
	}
	}
	return fileSize;
	}

	/**
	* On close file which are not present in remote are removed from local.
	* CopyOnReadDir is opened at different revisions of the index state
	*
	* CDir1 - V1
	* CDir2 - V2
	*
	* Its possible that two different IndexSearcher are opened at same local
	* directory but pinned to different revisions. So while removing it must
	* be ensured that any currently opened IndexSearcher does not get affected.
	* The way IndexSearchers get created in IndexTracker it ensures that new searcher
	* pinned to newer revision gets opened first and then existing ones are closed.
	*
	*
	*/
	@Override
	public void close() throws IOException {
	if (!closed.compareAndSet(false, true)){
	return;
	}
	//Always remove old index file on close as it ensures that
	//no other IndexSearcher are opened with previous revision of Index due to
	//way IndexTracker closes LuceneIndexNode. At max there would be only two LuceneIndexNode
	//opened pinned to different revision of same Lucene index
	executor.execute(new Runnable() {
	@Override
	public void run() {
	try{
	removeDeletedFiles();
	} catch (IOException e) {
	log.warn(
	"[{}] Error occurred while removing deleted files from Local {}, Remote {}",
	indexPath, local, remote, e);
	}

	try {
	//This would also remove old index files if current
	//directory was based on newerRevision as local would
	//be of type DeleteOldDirOnClose
	local.close();
	remote.close();
	} catch (IOException e) {
	log.warn(
	"[{}] Error occurred while closing directory ",
	indexPath, e);
	}
	}
	});
	}

	@Override
	public String toString() {
	return String.format("[COR] Local %s, Remote %s", local, remote);
	}

	private void removeDeletedFiles() throws IOException {
	Set<String> remoteFiles = stream(remote.listAll())
	.filter(name -> !IndexCopier.REMOTE_ONLY.contains(name))
	.collect(Collectors.toSet());

	long maxTS = IndexCopier.getNewestLocalFSTimestampFor(remoteFiles, local);
	if (maxTS == -1) {
	log.warn("Couldn't compute safe timestamp to delete files from {}", local);
	return;
	}

	// subtract DELETE_MARGIN_MILLIS from maxTS for safety (you can never be too careful with time)
	final long deleteBeforeTS = maxTS - DELETE_MARGIN_MILLIS;

	Set<String> filesToBeDeleted =
	// Files present locally
	ImmutableSet.copyOf(local.listAll()).stream()
	// but not in my view
	.filter(name -> !remoteFiles.contains(name))
	// and also older than a safe timestamp (deleteBeforeTS)
	.filter(name -> IndexCopier.isFileModifiedBefore(name, local, deleteBeforeTS))
	// can be deleted
	.collect(Collectors.toSet())
	;

	Set<String> failedToDelete = Sets.newHashSet();

	for (String fileName : filesToBeDeleted) {
	boolean deleted = indexCopier.deleteFile(local, fileName, true);
	if (!deleted){
	failedToDelete.add(fileName);
	}
	}

	filesToBeDeleted = new HashSet<String>(filesToBeDeleted);
	filesToBeDeleted.removeAll(failedToDelete);
	if(!filesToBeDeleted.isEmpty()) {
	log.debug(
	"[{}] Following files have been removed from Lucene index directory {}",
	indexPath, filesToBeDeleted);
	}
	}

	private class CORFileReference {
	final String name;
	private volatile boolean valid;

	private CORFileReference(String name) {
	this.name = name;
	}

	boolean isLocalValid(){
	return valid;
	}

	IndexInput openLocalInput( IOContext context) throws IOException {
	indexCopier.readFromLocal(true);
	return local.openInput(name, context);
	}

	void markValid(){
	this.valid = true;
	}
	}
	}