| package org.apache.lucene.replicator; |
| |
| /* |
| * 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. |
| */ |
| |
| import java.io.Closeable; |
| import java.io.File; |
| import java.io.FileNotFoundException; |
| import java.io.IOException; |
| import java.nio.file.NoSuchFileException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Queue; |
| import java.util.Set; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentLinkedQueue; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.locks.Condition; |
| import java.util.concurrent.locks.Lock; |
| import java.util.concurrent.locks.Lock; |
| import java.util.concurrent.locks.ReentrantLock; |
| import java.util.zip.Adler32; |
| import java.util.zip.Checksum; |
| |
| import org.apache.lucene.analysis.MockAnalyzer; |
| import org.apache.lucene.codecs.CodecUtil; |
| import org.apache.lucene.document.Document; |
| import org.apache.lucene.document.Field; |
| import org.apache.lucene.index.AtomicReader; |
| import org.apache.lucene.index.AtomicReaderContext; |
| import org.apache.lucene.index.ConcurrentMergeScheduler; |
| import org.apache.lucene.index.CorruptIndexException; |
| import org.apache.lucene.index.DirectoryReader; |
| import org.apache.lucene.index.IndexFileNames; |
| import org.apache.lucene.index.IndexNotFoundException; |
| import org.apache.lucene.index.IndexReader; |
| import org.apache.lucene.index.IndexWriter.IndexReaderWarmer; |
| import org.apache.lucene.index.IndexWriter; |
| import org.apache.lucene.index.IndexWriterConfig; |
| import org.apache.lucene.index.KeepOnlyLastCommitDeletionPolicy; |
| import org.apache.lucene.index.MergePolicy; |
| import org.apache.lucene.index.RandomIndexWriter; |
| import org.apache.lucene.index.SegmentCommitInfo; |
| import org.apache.lucene.index.SegmentInfos; |
| import org.apache.lucene.index.SegmentReader; |
| import org.apache.lucene.index.StandardDirectoryReader; |
| import org.apache.lucene.index.Term; |
| import org.apache.lucene.search.IndexSearcher; |
| import org.apache.lucene.search.MatchAllDocsQuery; |
| import org.apache.lucene.search.ReferenceManager; |
| import org.apache.lucene.search.TopDocs; |
| import org.apache.lucene.store.AlreadyClosedException; |
| import org.apache.lucene.store.BaseDirectoryWrapper; |
| import org.apache.lucene.store.ByteArrayDataInput; |
| import org.apache.lucene.store.ChecksumIndexOutput; |
| import org.apache.lucene.store.DataInput; |
| import org.apache.lucene.store.Directory; |
| import org.apache.lucene.store.FlushInfo; |
| import org.apache.lucene.store.IOContext; |
| import org.apache.lucene.store.IndexInput; |
| import org.apache.lucene.store.IndexOutput; |
| import org.apache.lucene.store.MergeInfo; |
| import org.apache.lucene.store.MockDirectoryWrapper; |
| import org.apache.lucene.store.NRTCachingDirectory; |
| import org.apache.lucene.store.RAMDirectory; |
| import org.apache.lucene.store.RAMOutputStream; |
| import org.apache.lucene.store.RateLimitedIndexOutput; |
| import org.apache.lucene.store.RateLimiter; |
| import org.apache.lucene.util.IOUtils; |
| import org.apache.lucene.util.InfoStream; |
| import org.apache.lucene.util.LineFileDocs; |
| import org.apache.lucene.util.LuceneTestCase.SuppressCodecs; |
| import org.apache.lucene.util.LuceneTestCase; |
| import org.apache.lucene.util.PrintStreamInfoStream; |
| import org.apache.lucene.util.TreeLogger; |
| import org.apache.lucene.util._TestUtil; |
| import org.junit.AfterClass; |
| |
| // nocommit add SLM |
| |
| // nocommit also allow downgrade of Master to Replica, |
| // instead of Master.close then Replica init |
| |
| // nocommit make sure we are not over-IncRef'ing infos on master |
| |
| // nocommit test replicas that are slow to copy |
| |
| // nocommit add fang: master crashes |
| |
| // nocommit provoke more "going backwards", e.g. randomly |
| // sometimes shutdown whole cluster and pick random node to |
| // be the new master |
| |
| // nocommit should we have support for "flush as frequently |
| // as you can"? or at least, "do not flush so frequently |
| // that replicas can't finish copying before next flush"? |
| |
| // nocommit what about network partitioning |
| |
| // nocommit make MDW throw exceptions sometimes |
| |
| // nocommit test warming merge w/ O_DIRECT via NativeUnixDir |
| |
| // nocommit test rare bit errors during copy |
| |
| // nocommit test slow replica |
| |
| // nocommit test replica that "falls out" because it's too |
| // slow and then tries to join back again w/o having |
| // "properly" restarted |
| |
| // nocommit rewrite the test so each node has its own |
| // threads, to be closer to the concurrency we'd "really" |
| // see across N machines |
| |
| // nocommit also allow replicas pulling files from replicas; |
| // they need not always come from master |
| |
| @SuppressCodecs({ "SimpleText", "Memory", "Direct" }) |
| public class TestNRTReplication extends LuceneTestCase { |
| |
| static volatile Master master; |
| static final AtomicInteger masterCount = new AtomicInteger(); |
| static final Lock masterLock = new ReentrantLock(); |
| static Object[] nodes; |
| |
| @AfterClass |
| public static void afterClass() { |
| System.out.println("TEST: now afterClass"); |
| master = null; |
| nodes = null; |
| } |
| |
| public void test() throws Throwable { |
| try { |
| _test(); |
| } catch (Throwable t) { |
| System.out.println("FAILED:"); |
| t.printStackTrace(System.out); |
| throw t; |
| } |
| } |
| |
| private void _test() throws Exception { |
| |
| Thread.currentThread().setName("main"); |
| TreeLogger.setLogger(new TreeLogger("main")); |
| |
| // Maps all segmentInfos.getVersion() we've seen, to the |
| // expected doc count. On each replica when we do a |
| // *:* search we verify the totalHits is correct: |
| final Map<Long,Integer> versionDocCounts = new ConcurrentHashMap<Long,Integer>(); |
| |
| int numDirs = 1+_TestUtil.nextInt(random(), 2, 6); |
| // nocommit |
| //int numDirs = 2; |
| |
| final File[] dirs = new File[numDirs]; |
| |
| // One Master (initially node 0) and N-1 Replica: |
| nodes = new Object[numDirs]; |
| System.out.println("TEST: " + nodes.length + " nodes"); |
| for(int i=0;i<numDirs;i++) { |
| dirs[i] = _TestUtil.getTempDir("NRTReplication." + i + "_"); |
| if (i > 0) { |
| // Some replicas don't start on init: |
| if (random().nextInt(10) < 7) { |
| nodes[i] = new Replica(dirs[i], i, versionDocCounts, null); |
| } else { |
| System.out.println("TEST: skip replica " + i + " startup"); |
| } |
| } |
| } |
| |
| nodes[0] = master = new Master(dirs[0], 0, versionDocCounts); |
| |
| // nocommit test graceful full shutdown / restart |
| |
| // nocommit test all nodes hard shutdown |
| |
| long endTime = System.currentTimeMillis() + 60000; |
| //long endTime = System.currentTimeMillis() + 10000; |
| |
| while (System.currentTimeMillis() < endTime) { |
| Thread.sleep(_TestUtil.nextInt(random(), 2, 50)); |
| |
| assert master != null; |
| |
| TreeLogger.log("\nTEST: now replicate master id=" + master.id); |
| TreeLogger.start("replicate"); |
| |
| SegmentInfos infos; |
| boolean closeMaster = random().nextInt(100) == 57; |
| if (closeMaster) { |
| TreeLogger.log("top: id=" + master.id + " now move master"); |
| // Commits & closes current master and pull the |
| // infos of the final commit: |
| master.close(random().nextBoolean()); |
| TreeLogger.log("top: done shutdown master"); |
| } else { |
| |
| // Have writer do a full flush, and return the |
| // resulting segments, protected from deletion |
| // (incRef'd) just while we copy the files out to |
| // the replica (s). This is just like pulling an |
| // NRT reader, except we don't actually open the |
| // readers on the newly flushed segments: |
| TreeLogger.log("flush current master"); |
| master.flush(); |
| TreeLogger.log("done flush current master"); |
| } |
| |
| CopyState copyState = master.getCopyState(); |
| |
| // nocommit also allow downgrade Master -> Replica, |
| // NOT a full close |
| |
| int count = docCount(copyState.infos); |
| TreeLogger.log("record version=" + copyState.version + " count=" + count + " segs=" + copyState.infos.toString(copyState.dir)); |
| Integer oldCount = versionDocCounts.put(copyState.version, count); |
| |
| // Refresh the local searcher on master: |
| if (closeMaster == false) { |
| master.mgr.setCurrentInfos(copyState.infos); |
| master.mgr.maybeRefresh(); |
| } |
| |
| // nocommit break this into separate tests, so we can |
| // test the "clean" case where versions are "correct": |
| |
| // nocommit cannot do this: versions can go backwards |
| //if (oldCount != null) { |
| //assertEquals("version=" + infos.version + " oldCount=" + oldCount + " newCount=" + count, oldCount.intValue(), count); |
| //} |
| |
| // nocommit can we have commit commit the "older" SIS? |
| // they will commit quickly since the OS will have |
| // already moved those bytes to disk... |
| |
| String extra = " master.sizeInBytes=" + ((NRTCachingDirectory) master.dir.getDelegate()).sizeInBytes(); |
| |
| TreeLogger.log("replicate docCount=" + count + " version=" + copyState.version + extra + " segments=" + copyState.infos.toString(copyState.dir)); |
| |
| // nocommit test master crash (promoting replica to master) |
| |
| // nocommit simulate super-slow replica: it should not |
| // hold up the copying of other replicas, nor new |
| // flushing; the copy of a given SIS to a given |
| // replica should be fully concurrent/bg |
| |
| // Notify all running replicas that they should now |
| // pull the new flush over: |
| int upto = 0; |
| for(Object n : nodes) { |
| if (n != null && n instanceof Replica) { |
| Replica r = (Replica) n; |
| // nocommit can we "broadcast" the new files |
| // instead of each replica pulling its own copy |
| // ... |
| TreeLogger.log("id=" + upto + ": signal new flush"); |
| r.newFlush(); |
| } else if (n == null) { |
| TreeLogger.log("id=" + upto + " skip down replica"); |
| } |
| upto++; |
| } |
| |
| master.releaseCopyState(copyState); |
| |
| if (closeMaster) { |
| if (random().nextBoolean()) { |
| TreeLogger.log("top: id=" + master.id + " now waitIdle"); |
| master.waitIdle(); |
| TreeLogger.log("top: id=" + master.id + " done waitIdle"); |
| } else { |
| TreeLogger.log("top: id=" + master.id + " skip waitIdle"); |
| Thread.sleep(random().nextInt(5)); |
| } |
| |
| TreeLogger.log("top: id=" + master.id + " close old master dir dir.listAll()=" + Arrays.toString(master.dir.listAll())); |
| master.dir.close(); |
| |
| masterLock.lock(); |
| try { |
| masterCount.incrementAndGet(); |
| } finally { |
| masterLock.unlock(); |
| } |
| |
| nodes[master.id] = null; |
| |
| // nocommit make sure we test race here, where |
| // replica is coming up just as we are electing a |
| // new master |
| |
| // Must pick newest replica to promote, else we |
| // can't overwrite open files when trying to copy |
| // to the newer replicas: |
| int bestIDX = -1; |
| long highestVersion = -1; |
| for (int idx=0;idx<nodes.length;idx++) { |
| if (nodes[idx] instanceof Replica) { |
| Replica r = (Replica) nodes[idx]; |
| long version = r.mgr.getCurrentInfos().version; |
| TreeLogger.log("top: id=" + r.id + " check version=" + version); |
| if (version > highestVersion) { |
| bestIDX = idx; |
| highestVersion = version; |
| TreeLogger.log("top: id=" + r.id + " check version=" + version + " max so far"); |
| } |
| } |
| } |
| |
| int idx; |
| if (bestIDX != -1) { |
| idx = bestIDX; |
| } else { |
| // All replicas are down; it doesn't matter |
| // which one we pick |
| idx = random().nextInt(nodes.length); |
| } |
| |
| if (nodes[idx] == null) { |
| // Start up Master from scratch: |
| TreeLogger.log("top: id=" + idx + " promote down node to master"); |
| nodes[idx] = master = new Master(dirs[idx], idx, versionDocCounts); |
| } else { |
| // Promote a running replica to Master: |
| assert nodes[idx] instanceof Replica; |
| TreeLogger.log("top: id=" + idx + " promote replica to master"); |
| master = new Master((Replica) nodes[idx]); |
| nodes[idx] = master; |
| } |
| } else { |
| if (random().nextInt(100) == 17) { |
| TreeLogger.log("top: id=" + master.id + " commit master"); |
| master.commit(); |
| } |
| } |
| |
| // Maybe restart a down replica, or commit / shutdown |
| // / crash one: |
| for(int i=0;i<nodes.length;i++) { |
| if (nodes[i] == null && random().nextInt(20) == 17) { |
| // Restart this replica |
| try { |
| nodes[i] = new Replica(dirs[i], i, versionDocCounts, null); |
| } catch (Throwable t) { |
| TreeLogger.log("top: id=" + i + " FAIL startup", t); |
| throw t; |
| } |
| } else { |
| if (nodes[i] instanceof Replica) { |
| Replica r = (Replica) nodes[i]; |
| |
| if (random().nextInt(100) == 17) { |
| TreeLogger.log("top: id=" + i + " commit replica"); |
| r.commit(false); |
| } |
| |
| if (random().nextInt(100) == 17) { |
| // Now shutdown this replica |
| TreeLogger.log("top: id=" + i + " shutdown replica"); |
| r.shutdown(); |
| nodes[i] = null; |
| break; |
| } else if (random().nextInt(100) == 17) { |
| // Now crash the replica |
| TreeLogger.log("top: id=" + i + " crash replica"); |
| r.crash(); |
| nodes[i] = null; |
| break; |
| } |
| } |
| } |
| } |
| |
| TreeLogger.end("replicate"); |
| } |
| |
| System.out.println("TEST: close replicas"); |
| for(Object n : nodes) { |
| if (n != null && n instanceof Replica) { |
| ((Replica) n).shutdown(); |
| } |
| } |
| |
| if (master != null) { |
| System.out.println("TEST: close master"); |
| master.close(random().nextBoolean()); |
| master.dir.close(); |
| } |
| } |
| |
| static class FileMetaData { |
| public final long length; |
| public final long checksum; |
| |
| public FileMetaData(long length, long checksum) { |
| this.length = length; |
| this.checksum = checksum; |
| } |
| } |
| |
| static Map<String,FileMetaData> getFilesMetaData(Master master, Collection<String> files) throws IOException { |
| Map<String,FileMetaData> filesMetaData = new HashMap<String,FileMetaData>(); |
| for(String file : files) { |
| filesMetaData.put(file, new FileMetaData(master.dir.fileLength(file), master.dir.getChecksum(file))); |
| } |
| |
| return filesMetaData; |
| } |
| |
| static Map<String,FileMetaData> copyFiles(SlowChecksumDirectory src, Replica dst, Map<String,FileMetaData> files, boolean lowPriority, long totBytes) throws IOException { |
| long t0 = System.currentTimeMillis(); |
| |
| // nocommit should we "organize" the files to be copied |
| // by segment name? so that NRTCachingDir can |
| // separately decide on a seg by seg basis whether to |
| // cache the seg's files? |
| IOContext ioContext; |
| if (lowPriority) { |
| // nocommit do we need the tot docs / merge num segs? |
| ioContext = new IOContext(new MergeInfo(0, totBytes, false, 0)); |
| } else { |
| // nocommit can we get numDocs? |
| ioContext = new IOContext(new FlushInfo(0, totBytes)); |
| } |
| |
| CopyResult copyResult = dst.fileCopier.add(src, files, lowPriority, ioContext); |
| |
| try { |
| copyResult.done.await(); |
| } catch (InterruptedException ie) { |
| // nocommit |
| Thread.currentThread().interrupt(); |
| throw new RuntimeException(ie); |
| } |
| |
| if (copyResult.failed.get()) { |
| TreeLogger.log("replica " + dst.id + ": " + (lowPriority ? "low-priority " : "") + " failed to copy some files"); |
| return null; |
| } |
| |
| long t1 = System.currentTimeMillis(); |
| TreeLogger.log("replica " + dst.id + ": " + (lowPriority ? "low-priority " : "") + "took " + (t1-t0) + " millis for " + totBytes + " bytes; " + files.size() + " files; nrtDir.sizeInBytes=" + ((NRTCachingDirectory) dst.dir.getDelegate()).sizeInBytes()); |
| |
| return files; |
| } |
| |
| /** Like SearcherManager, except it refreshes via an |
| * externally provided (NRT) SegmentInfos. */ |
| private static class InfosSearcherManager extends ReferenceManager<IndexSearcher> { |
| private volatile SegmentInfos currentInfos; |
| private final Directory dir; |
| |
| public InfosSearcherManager(Directory dir, int id, SegmentInfos infosIn) throws IOException { |
| this.dir = dir; |
| if (infosIn != null) { |
| currentInfos = infosIn; |
| TreeLogger.log("InfosSearcherManager.init id=" + id + ": use incoming infos=" + infosIn.toString(dir)); |
| } else { |
| currentInfos = new SegmentInfos(); |
| String fileName = SegmentInfos.getLastCommitSegmentsFileName(dir); |
| if (fileName != null) { |
| // Load last commit: |
| currentInfos.read(dir, fileName); |
| TreeLogger.log("InfosSearcherManager.init id=" + id + ": load initial segments file " + fileName + ": loaded version=" + currentInfos.version + " docCount=" + docCount(currentInfos)); |
| } |
| } |
| current = new IndexSearcher(StandardDirectoryReader.open(dir, currentInfos, null)); |
| } |
| |
| @Override |
| protected int getRefCount(IndexSearcher s) { |
| return s.getIndexReader().getRefCount(); |
| } |
| |
| @Override |
| protected boolean tryIncRef(IndexSearcher s) { |
| return s.getIndexReader().tryIncRef(); |
| } |
| |
| @Override |
| protected void decRef(IndexSearcher s) throws IOException { |
| s.getIndexReader().decRef(); |
| } |
| |
| public SegmentInfos getCurrentInfos() { |
| return currentInfos; |
| } |
| |
| public void setCurrentInfos(SegmentInfos infos) { |
| if (currentInfos != null) { |
| // So that if we commit, we will go to the next |
| // (unwritten so far) generation: |
| infos.updateGeneration(currentInfos); |
| TreeLogger.log("mgr.setCurrentInfos: carry over infos gen=" + infos.getSegmentsFileName()); |
| } |
| currentInfos = infos; |
| } |
| |
| // nocommit who manages the "ref" for currentInfos? |
| |
| @Override |
| protected IndexSearcher refreshIfNeeded(IndexSearcher old) throws IOException { |
| List<AtomicReader> subs; |
| if (old == null) { |
| subs = null; |
| } else { |
| subs = new ArrayList<AtomicReader>(); |
| for(AtomicReaderContext ctx : old.getIndexReader().leaves()) { |
| subs.add(ctx.reader()); |
| } |
| } |
| |
| return new IndexSearcher(StandardDirectoryReader.open(dir, currentInfos, subs)); |
| } |
| } |
| |
| private static class CopyState { |
| public final SlowChecksumDirectory dir; |
| public final Map<String,FileMetaData> files; |
| public final long version; |
| public final byte[] infosBytes; |
| public final SegmentInfos infos; |
| |
| public CopyState(SlowChecksumDirectory dir, Map<String,FileMetaData> files, long version, byte[] infosBytes, SegmentInfos infos) { |
| this.dir = dir; |
| this.files = files; |
| this.version = version; |
| this.infosBytes = infosBytes; |
| this.infos = infos; |
| } |
| } |
| |
| // nocommit should extend replica and just add writer |
| private static class Master { |
| |
| final Set<String> finishedMergedSegments = Collections.newSetFromMap(new ConcurrentHashMap<String,Boolean>()); |
| |
| final SlowChecksumDirectory dir; |
| |
| final RandomIndexWriter writer; |
| final InfosSearcherManager mgr; |
| final SearchThread searchThread; |
| final IndexThread indexThread; |
| final int id; |
| private volatile boolean isClosed; |
| private AtomicInteger infosRefCount = new AtomicInteger(); |
| |
| SegmentInfos lastInfos; |
| |
| /** Start up a master from scratch. */ |
| public Master(File path, |
| int id, Map<Long,Integer> versionDocCounts) throws IOException { |
| final MockDirectoryWrapper dirOrig = newMockFSDirectory(path); |
| |
| // In some legitimate cases we will double-write: |
| dirOrig.setPreventDoubleWrite(false); |
| |
| this.id = id; |
| |
| // nocommit put back |
| dirOrig.setCheckIndexOnClose(false); |
| |
| // Master may legitimately close while replicas are |
| // still copying from it: |
| dirOrig.setAllowCloseWithOpenFiles(true); |
| |
| dir = new SlowChecksumDirectory(id, new NRTCachingDirectory(dirOrig, 1.0, 10.0)); |
| //((NRTCachingDirectory) master).VERBOSE = true; |
| SegmentInfos infos = new SegmentInfos(); |
| try { |
| infos.read(dir); |
| } catch (IndexNotFoundException infe) { |
| } |
| |
| mgr = new InfosSearcherManager(dir, id, infos); |
| searchThread = new SearchThread("master", mgr, versionDocCounts); |
| searchThread.start(); |
| |
| SegmentInfos curInfos = mgr.getCurrentInfos().clone(); |
| |
| IndexWriterConfig iwc = getIWC(); |
| iwc.setOpenMode(IndexWriterConfig.OpenMode.APPEND); |
| |
| // nocommit also test periodically committing, and |
| // preserving multiple commit points; verify these |
| // "survive" over to the replica |
| writer = new RandomIndexWriter(random(), dir, curInfos, iwc); |
| _TestUtil.reduceOpenFiles(writer.w); |
| System.out.println("after reduce: " + writer.w.getConfig()); |
| |
| lastInfos = mgr.getCurrentInfos(); |
| // nocommit thread hazard here? IW could have already |
| // nuked some segments...? |
| writer.w.incRefDeleter(lastInfos); |
| setCopyState(); |
| |
| indexThread = new IndexThread(this); |
| indexThread.start(); |
| } |
| |
| public void commit() throws IOException { |
| writer.w.prepareCommit(); |
| // It's harmless if we crash here, because the |
| // global state has already been updated |
| writer.w.commit(); |
| } |
| |
| /** Promotes an existing Replica to Master, re-using the |
| * open NRTCachingDir, the SearcherManager, the search |
| * thread, etc. */ |
| public Master(Replica replica) throws IOException { |
| this.id = replica.id; |
| this.dir = replica.dir; |
| this.mgr = replica.mgr; |
| this.searchThread = replica.searchThread; |
| |
| // Master may legitimately close while replicas are |
| // still copying from it: |
| ((MockDirectoryWrapper) ((NRTCachingDirectory) dir.getDelegate()).getDelegate()).setAllowCloseWithOpenFiles(true); |
| |
| // Do not copy from ourself: |
| try { |
| replica.copyThread.finish(); |
| } catch (InterruptedException ie) { |
| Thread.currentThread().interrupt(); |
| throw new RuntimeException(ie); |
| } |
| replica.fileCopier.close(); |
| |
| // nocommit must somehow "stop" this replica? e.g. we |
| // don't want it doing any more deleting? |
| |
| IndexWriterConfig iwc = getIWC(); |
| iwc.setOpenMode(IndexWriterConfig.OpenMode.APPEND); |
| |
| // nocommit also test periodically committing, and |
| // preserving multiple commit points; verify these |
| // "survive" over to the replica |
| SegmentInfos infos = replica.mgr.getCurrentInfos().clone(); |
| TreeLogger.log("top: id=" + replica.id + " sis version=" + infos.version); |
| |
| writer = new RandomIndexWriter(random(), dir, replica.mgr.getCurrentInfos().clone(), iwc); |
| _TestUtil.reduceOpenFiles(writer.w); |
| System.out.println("after reduce: " + writer.w.getConfig()); |
| |
| lastInfos = mgr.getCurrentInfos(); |
| |
| writer.w.incRefDeleter(lastInfos); |
| |
| setCopyState(); |
| |
| // nocommit thread hazard here? IW could have already |
| // nuked some segments...? |
| writer.w.incRefDeleter(lastInfos); |
| |
| indexThread = new IndexThread(this); |
| indexThread.start(); |
| } |
| |
| private IndexWriterConfig getIWC() { |
| IndexWriterConfig iwc = newIndexWriterConfig(TEST_VERSION_CURRENT, new MockAnalyzer(random())); |
| |
| // Install a merged segment warmer that pre-copies (low |
| // priority) merged segment files out to replica(s) |
| // before the merge is swapped into the live segments; |
| // this way a large merge won't interfere with |
| // NRT turnaround time: |
| iwc.setMergedSegmentWarmer(new IndexReaderWarmer() { |
| @Override |
| public void warm(AtomicReader reader) throws IOException { |
| SegmentCommitInfo info = ((SegmentReader) reader).getSegmentInfo(); |
| if (TreeLogger.getLogger() == null) { |
| TreeLogger.setLogger(new TreeLogger("merge")); |
| } |
| //System.out.println("TEST: warm merged segment files " + info); |
| Map<String,FileMetaData> toCopy = getFilesMetaData(Master.this, info.files()); |
| long totBytes = 0; |
| for(FileMetaData metaData : toCopy.values()) { |
| totBytes += metaData.length; |
| } |
| TreeLogger.log("warm files=" + toCopy.keySet() + " totBytes=" + totBytes); |
| TreeLogger.start("warmMerge"); |
| |
| IOContext context = new IOContext(new MergeInfo(0, totBytes, false, 0)); |
| |
| List<CopyResult> copyResults = new ArrayList<>(); |
| for(Object n : nodes) { |
| if (n != null && n instanceof Replica) { |
| Replica r = (Replica) n; |
| |
| try { |
| |
| // nocommit we could also have replica pre-warm a SegmentReader |
| // here, and add it onto subReader list |
| // for next reopen ...? |
| |
| // Must call filter, in case we are |
| // overwriting files and must invalidate |
| // the last commit: |
| int sizeBefore = toCopy.size(); |
| Map<String,FileMetaData> toCopy2 = r.filterFilesToCopy(toCopy); |
| |
| // Since this is a newly merged segment, |
| // all files should be new and need |
| // copying: |
| assert toCopy.size() == sizeBefore: "before=" + toCopy.keySet() + " after=" + toCopy2.keySet(); |
| |
| TreeLogger.log("copy to replica " + r.id); |
| copyResults.add(r.copyMergedFiles(Master.this.dir, toCopy2, context)); |
| } catch (AlreadyClosedException ace) { |
| // Ignore this: it means the replica shut down |
| // while we were trying to copy files. This |
| // "approximates" an exception the master would |
| // see trying to push file bytes to a replica |
| // that was just taken offline. |
| } |
| } |
| } |
| |
| TreeLogger.log("now wait for " + copyResults.size() + " replicas to finish copying"); |
| for(CopyResult result : copyResults) { |
| try { |
| result.done.await(); |
| } catch (InterruptedException ie) { |
| // nocommit |
| } |
| |
| // nocommit if there's an error ... what to |
| // do? |
| |
| // nocommit we should check merge.abort |
| // somehow in here, so if the master is in |
| // a hurry to shutdown, we respect that |
| } |
| TreeLogger.log("done warm merge for " + copyResults.size() + " replicas"); |
| TreeLogger.end("warmMerge"); |
| } |
| }); |
| |
| return iwc; |
| } |
| |
| /** Gracefully shuts down the master */ |
| public void close(boolean waitForMerges) throws IOException { |
| TreeLogger.log("id=" + id + " close master waitForMerges=" + waitForMerges); |
| |
| try { |
| searchThread.finish(); |
| } catch (InterruptedException ie) { |
| Thread.currentThread().interrupt(); |
| throw new RuntimeException(ie); |
| } |
| |
| if (waitForMerges) { |
| // Do it here, instead of on close, so we can |
| // continue indexing while waiting for merges: |
| writer.w.waitForMerges(); |
| TreeLogger.log("waitForMerges done"); |
| } |
| |
| mgr.close(); |
| |
| try { |
| indexThread.finish(); |
| } catch (InterruptedException ie) { |
| Thread.currentThread().interrupt(); |
| throw new RuntimeException(ie); |
| } |
| |
| TreeLogger.log("close writer"); |
| |
| // nocommit can we optionally NOT commit here? it |
| // should be decoupled from master migration? |
| |
| commit(); |
| |
| synchronized(this) { |
| isClosed = true; |
| if (lastInfos != null) { |
| writer.w.decRefDeleter(lastInfos); |
| lastInfos = null; |
| copyState = null; |
| } |
| |
| // nocommit what about exc from writer.close... |
| // Don't wait for merges now; we already did above: |
| writer.close(false); |
| TreeLogger.log("done close writer"); |
| |
| SegmentInfos infos = new SegmentInfos(); |
| infos.read(dir); |
| |
| TreeLogger.log("final infos=" + infos.toString(master.dir)); |
| |
| // nocommit caller must close |
| // dir.close(); |
| |
| lastInfos = infos; |
| setCopyState(); |
| } |
| } |
| |
| public synchronized boolean isClosed() { |
| return isClosed; |
| } |
| |
| private CopyState copyState; |
| |
| private synchronized void setCopyState() throws IOException { |
| RAMOutputStream out = new RAMOutputStream(); |
| lastInfos.write(out); |
| byte[] infosBytes = new byte[(int) out.getFilePointer()]; |
| out.writeTo(infosBytes, 0); |
| copyState = new CopyState(dir, |
| getFilesMetaData(this, lastInfos.files(dir, false)), |
| lastInfos.version, infosBytes, lastInfos); |
| } |
| |
| public synchronized CopyState getCopyState() throws IOException { |
| //if (isClosed && lastInfos == null) { |
| //return null; |
| //} |
| TreeLogger.log("master.getCopyState version=" + lastInfos.version + " files=" + lastInfos.files(dir, false) + " writer=" + writer); |
| if (isClosed == false) { |
| writer.w.incRefDeleter(copyState.infos); |
| } |
| int count = infosRefCount.incrementAndGet(); |
| assert count >= 1; |
| return copyState; |
| } |
| |
| /** Flushes, returns a ref with the resulting infos. */ |
| public void flush() throws IOException { |
| SegmentInfos newInfos = master.writer.w.flushAndIncRef(); |
| |
| synchronized(this) { |
| writer.w.decRefDeleter(lastInfos); |
| // Steals the reference returned by IW: |
| lastInfos = newInfos; |
| setCopyState(); |
| } |
| } |
| |
| public synchronized void releaseCopyState(CopyState copyState) throws IOException { |
| TreeLogger.log("master.releaseCopyState version=" + copyState.version + " files=" + copyState.files.keySet()); |
| // Must check because by the time the replica releases |
| // it's possible it's a different master: |
| if (copyState.dir == dir) { |
| if (isClosed == false) { |
| writer.w.decRefDeleter(copyState.infos); |
| } |
| int count = infosRefCount.decrementAndGet(); |
| assert count >= 0; |
| TreeLogger.log(" infosRefCount=" + infosRefCount); |
| if (count == 0) { |
| notify(); |
| } |
| } else { |
| TreeLogger.log(" skip: wrong master"); |
| } |
| } |
| |
| /** Waits until all outstanding infos refs are dropped. */ |
| public synchronized void waitIdle() throws InterruptedException { |
| while (true) { |
| if (infosRefCount.get() == 0) { |
| return; |
| } |
| wait(); |
| } |
| } |
| } |
| |
| static class IndexThread extends Thread { |
| |
| final Master master; |
| volatile boolean stop; |
| |
| public IndexThread(Master master) { |
| this.master = master; |
| setName("IndexThread id=" + master.id); |
| } |
| |
| @Override |
| public void run() { |
| try { |
| int docID = 0; |
| LineFileDocs docs = new LineFileDocs(random()); |
| while (stop == false) { |
| Document doc = docs.nextDoc(); |
| Field idField = doc.getField("docid"); |
| if (random().nextInt(10) == 9 && docID > 0) { |
| int randomDocID = random().nextInt(docID); |
| idField.setStringValue(""+randomDocID); |
| master.writer.updateDocument(new Term("docid", ""+randomDocID), doc); |
| } else { |
| idField.setStringValue(""+docID); |
| master.writer.addDocument(doc); |
| docID++; |
| } |
| // ~500 docs/sec |
| Thread.sleep(2); |
| } |
| } catch (Exception e) { |
| throw new RuntimeException("IndexThread on id=" + master.id + ": " + e, e); |
| } |
| } |
| |
| public void finish() throws InterruptedException { |
| stop = true; |
| join(); |
| } |
| } |
| |
| private static class Replica { |
| final int id; |
| final SlowChecksumDirectory dir; |
| final InfosRefCounts deleter; |
| |
| private final InfosSearcherManager mgr; |
| private volatile boolean stop; |
| final SearchThread searchThread; |
| final SimpleFileCopier fileCopier; |
| CopyThread copyThread; |
| |
| private final Collection<String> lastCommitFiles; |
| private final Collection<String> lastNRTFiles; |
| |
| // nocommit unused so far: |
| private final IndexReaderWarmer mergedSegmentWarmer; |
| |
| public Replica(File path, int id, Map<Long,Integer> versionDocCounts, IndexReaderWarmer mergedSegmentWarmer) throws IOException { |
| TreeLogger.log("top: id=" + id + " replica startup path=" + path); |
| TreeLogger.start("startup"); |
| |
| this.id = id; |
| this.mergedSegmentWarmer = mergedSegmentWarmer; |
| |
| Directory fsDir = newMockFSDirectory(path); |
| |
| // In some legitimate cases we will double-write: |
| ((MockDirectoryWrapper) fsDir).setPreventDoubleWrite(false); |
| |
| // nocommit put back |
| ((BaseDirectoryWrapper) fsDir).setCheckIndexOnClose(false); |
| |
| dir = new SlowChecksumDirectory(id, new NRTCachingDirectory(fsDir, 1.0, 10.0)); |
| TreeLogger.log("id=" + id + " created dirs, checksums; dir.listAll=" + Arrays.toString(dir.listAll())); |
| |
| fileCopier = new SimpleFileCopier(dir, id); |
| fileCopier.start(); |
| |
| lastCommitFiles = new HashSet<String>(); |
| lastNRTFiles = new HashSet<String>(); |
| |
| deleter = new InfosRefCounts(id, dir); |
| |
| String segmentsFileName = SegmentInfos.getLastCommitSegmentsFileName(dir); |
| SegmentInfos lastCommit = null; |
| if (segmentsFileName != null) { |
| lastCommit = new SegmentInfos(); |
| TreeLogger.log("id=" + id + " " + segmentsFileName + " now load"); |
| lastCommit.read(dir, segmentsFileName); |
| lastCommitFiles.addAll(lastCommit.files(dir, true)); |
| TreeLogger.log("id=" + id + " incRef lastCommitFiles"); |
| deleter.incRef(lastCommitFiles); |
| TreeLogger.log("id=" + id + " loaded version=" + lastCommit.version + " lastCommitFiles = " + lastCommitFiles); |
| } |
| |
| // Must sync latest index from master before starting |
| // up mgr, so that we don't hold open any files that |
| // need to be overwritten when the master is against |
| // an older index than our copy, and so we rollback |
| // our version if we had been at a higher version but |
| // were down when master moved: |
| SegmentInfos infos = null; |
| |
| // Startup sync to pull latest index over: |
| Map<String,FileMetaData> copiedFiles = null; |
| |
| if (master != null) { |
| |
| // nocommit this logic isn't right; else, on a full |
| // restart how will a master be found: |
| |
| // Repeat until we find a working master; this is to |
| // handle the case when a replica starts up but no |
| // new master has yet been selected when moving |
| // master: |
| while (true) { |
| Master curMaster = master; |
| CopyState copyState = curMaster.getCopyState(); |
| try { |
| // nocommit factor this out & share w/ CopyThread: |
| |
| deleter.incRef(copyState.files.keySet()); |
| |
| Map<String,FileMetaData> toCopy = filterFilesToCopy(copyState.files); |
| long totBytes = 0; |
| for(FileMetaData metaData : toCopy.values()) { |
| totBytes += metaData.length; |
| } |
| |
| // Copy files over to replica: |
| copiedFiles = copyFiles(copyState.dir, this, toCopy, false, totBytes); |
| if (copiedFiles == null) { |
| TreeLogger.log("id=" + id + " startup copyFiles failed"); |
| deleter.decRef(copyState.files.keySet()); |
| try { |
| Thread.sleep(5); |
| } catch (InterruptedException ie) { |
| Thread.currentThread().interrupt(); |
| } |
| continue; |
| } |
| |
| // Turn byte[] back to SegmentInfos: |
| infos = new SegmentInfos(); |
| infos.read(dir, new ByteArrayDataInput(copyState.infosBytes)); |
| lastNRTFiles.addAll(copyState.files.keySet()); |
| } finally { |
| curMaster.releaseCopyState(copyState); |
| } |
| break; |
| } |
| } else { |
| TreeLogger.log("id=" + id + " no master on startup; fallback to last commit"); |
| } |
| |
| if (lastCommit != null) { |
| infos.updateGeneration(lastCommit); |
| } |
| |
| mgr = new InfosSearcherManager(dir, id, infos); |
| |
| deleter.deleteUnknownFiles(); |
| |
| searchThread = new SearchThread(""+id, mgr, versionDocCounts); |
| searchThread.start(); |
| |
| copyThread = new CopyThread(this); |
| copyThread.start(); |
| |
| TreeLogger.log("done startup"); |
| TreeLogger.end("startup"); |
| } |
| |
| /** From the incoming files, determines which ones need |
| * copying on this replica, because they weren't yet |
| * copied, or they were previously copied but the |
| * checksum or file length is different. */ |
| public Map<String,FileMetaData> filterFilesToCopy(Map<String,FileMetaData> files) throws IOException { |
| Map<String,FileMetaData> toCopy = new HashMap<>(); |
| |
| for(Map.Entry<String,FileMetaData> ent : files.entrySet()) { |
| String fileName = ent.getKey(); |
| FileMetaData metaData = ent.getValue(); |
| long srcChecksum = ent.getValue().checksum; |
| |
| Long checksum = dir.getChecksum(fileName); |
| |
| long curLength; |
| try { |
| curLength = dir.fileLength(fileName); |
| } catch (FileNotFoundException | NoSuchFileException e) { |
| curLength = -1; |
| } |
| if (curLength != metaData.length) { |
| TreeLogger.log("id=" + id + " " + fileName + " will copy [wrong file length old=" + curLength + " new=" + metaData.length + "]"); |
| toCopy.put(fileName, metaData); |
| } else if (checksum == null) { |
| TreeLogger.log("id=" + id + " " + fileName + " will copy [no checksum] length=" + metaData.length + " srcChecksum=" + srcChecksum); |
| toCopy.put(fileName, metaData); |
| } else if (checksum.longValue() != srcChecksum) { |
| TreeLogger.log("id=" + id + " " + fileName + " will copy [wrong checksum: cur=" + checksum.longValue() + " new=" + srcChecksum + "] length=" + metaData.length); |
| toCopy.put(fileName, metaData); |
| } else { |
| TreeLogger.log("id=" + id + " " + fileName + " skip copy checksum=" + srcChecksum + " file.length=" + metaData.length); |
| } |
| } |
| |
| // nocommit what about multiple commit points? |
| |
| // Invalidate the current commit if we will overwrite |
| // any files from it: |
| |
| // nocommit do we need sync'd here? |
| for(String fileName : toCopy.keySet()) { |
| if (lastCommitFiles.contains(fileName)) { |
| TreeLogger.log("id=" + id + " " + fileName + " is being copied but is referenced by last commit; now drop last commit; lastCommitFiles=" + lastCommitFiles); |
| deleter.decRef(lastCommitFiles); |
| dir.deleteFile("segments.gen"); |
| lastCommitFiles.clear(); |
| break; |
| } |
| } |
| |
| return toCopy; |
| } |
| |
| public void newFlush() { |
| copyThread.lock.lock(); |
| try { |
| copyThread.cond.signal(); |
| } finally { |
| copyThread.lock.unlock(); |
| } |
| } |
| |
| public CopyResult copyMergedFiles(SlowChecksumDirectory src, Map<String,FileMetaData> files, IOContext context) throws IOException { |
| if (stop) { |
| throw new AlreadyClosedException("replica closed"); |
| } |
| return fileCopier.add(src, files, true, context); |
| } |
| |
| void sync(int curMasterMoveCount, SlowChecksumDirectory master, Map<String,FileMetaData> filesMetaData, byte[] infosBytes, |
| long infosVersion) throws IOException { |
| |
| SegmentInfos currentInfos = mgr.getCurrentInfos(); |
| TreeLogger.log("id=" + id + " sync version=" + infosVersion + " vs current version=" + currentInfos.getVersion()); |
| TreeLogger.start("sync"); |
| |
| // nocommit make overall test "modal", ie up front |
| // decide whether any docs are allowed to be lost |
| // (version goes backwards on replicas) or not |
| |
| /* |
| if (currentInfos != null && currentInfos.getVersion() >= infosVersion) { |
| System.out.println(" replica id=" + id + ": skip sync current version=" + currentInfos.getVersion() + " vs new version=" + infosVersion); |
| return; |
| } |
| */ |
| |
| // Must incRef before filter in case filter decRefs |
| // the last commit: |
| deleter.incRef(filesMetaData.keySet()); |
| |
| Map<String,FileMetaData> toCopy = filterFilesToCopy(filesMetaData); |
| long totBytes = 0; |
| for(FileMetaData metaData : toCopy.values()) { |
| totBytes += metaData.length; |
| } |
| |
| // Copy files over to replica: |
| Map<String,FileMetaData> copiedFiles = copyFiles(master, this, toCopy, false, totBytes); |
| |
| if (copiedFiles == null) { |
| // At least one file failed to copy; skip cutover |
| TreeLogger.log("top: id=" + id + " replica sync failed; abort"); |
| deleter.decRef(filesMetaData.keySet()); |
| TreeLogger.end("sync"); |
| return; |
| } |
| |
| TreeLogger.log("top: id=" + id + " replica sync done file copy"); |
| |
| // OK all files copied successfully, now rebuild the |
| // infos and cutover searcher mgr |
| |
| // Turn byte[] back to SegmentInfos: |
| SegmentInfos infos = new SegmentInfos(); |
| infos.read(dir, new ByteArrayDataInput(infosBytes)); |
| TreeLogger.log("id=" + id + " replica sync version=" + infos.version + " segments=" + infos.toString(dir)); |
| |
| masterLock.lock(); |
| try { |
| if (curMasterMoveCount != masterCount.get()) { |
| // At least one file failed to copy; skip cutover |
| TreeLogger.log("top: id=" + id + " master moved during sync; abort"); |
| deleter.decRef(filesMetaData.keySet()); |
| TreeLogger.end("sync"); |
| return; |
| } |
| |
| synchronized (this) { |
| mgr.setCurrentInfos(infos); |
| deleter.decRef(lastNRTFiles); |
| lastNRTFiles.clear(); |
| lastNRTFiles.addAll(filesMetaData.keySet()); |
| } |
| } finally { |
| masterLock.unlock(); |
| } |
| |
| // Cutover to new searcher |
| mgr.maybeRefresh(); |
| |
| TreeLogger.log("top: id=" + id + " done mgr refresh"); |
| |
| TreeLogger.end("sync"); |
| } |
| |
| /** Gracefully close & shutdown this replica. */ |
| public void shutdown() throws IOException, InterruptedException { |
| if (stop) { |
| return; |
| } |
| stop = true; |
| |
| copyThread.finish(); |
| fileCopier.close(); |
| |
| // Sometimes shutdown w/o commiting |
| TreeLogger.log("id=" + id + " replica shutdown"); |
| TreeLogger.start("shutdown"); |
| if (random().nextBoolean()) { |
| TreeLogger.log("id=" + id + " commit before shutdown"); |
| commit(true); |
| } |
| |
| searchThread.finish(); |
| mgr.close(); |
| |
| // Delete now un-referenced files: |
| TreeLogger.log("id=" + id + " now deletePending during shutdown"); |
| deleter.decRef(lastNRTFiles); |
| lastNRTFiles.clear(); |
| deleter.deletePending(); |
| TreeLogger.log("id=" + id + " at shutdown dir.listAll()=" + Arrays.toString(dir.listAll())); |
| dir.close(); |
| TreeLogger.end("shutdown"); |
| } |
| |
| /** Crashes the underlying directory, corrupting any |
| * un-sync'd files. */ |
| public void crash() throws IOException, InterruptedException { |
| if (stop) { |
| return; |
| } |
| stop = true; |
| TreeLogger.log("id=" + id + " replica crash; dir.listAll()=" + Arrays.toString(dir.listAll())); |
| TreeLogger.log("id=" + id + " replica crash; fsdir.listAll()=" + Arrays.toString(((NRTCachingDirectory) dir.getDelegate()).getDelegate().listAll())); |
| copyThread.finish(); |
| fileCopier.close(); |
| searchThread.finish(); |
| mgr.close(); |
| ((MockDirectoryWrapper) ((NRTCachingDirectory) dir.getDelegate()).getDelegate()).crash(); |
| ((NRTCachingDirectory) dir.getDelegate()).getDelegate().close(); |
| } |
| |
| /** Commit latest SegmentInfos (fsync'ing all referenced |
| * files). */ |
| public void commit(boolean deleteAll) throws IOException { |
| SegmentInfos infos; |
| Collection<String> newFiles; |
| |
| synchronized (this) { |
| infos = mgr.getCurrentInfos(); |
| if (infos != null) { |
| newFiles = infos.files(dir, false); |
| deleter.incRef(newFiles); |
| } else { |
| // nocommit is infos ever null? |
| newFiles = null; |
| } |
| } |
| |
| if (infos != null) { |
| TreeLogger.log("top: id=" + id + " commit deleteAll=" + deleteAll + "; infos.version=" + infos.getVersion() + " infos.files=" + newFiles + " segs=" + infos.toString(dir)); |
| TreeLogger.start("commit"); |
| dir.sync(newFiles); |
| infos.commit(dir); |
| |
| String segmentsFileName = infos.getSegmentsFileName(); |
| deleter.incRef(Collections.singletonList(segmentsFileName)); |
| |
| synchronized (this) { |
| // If a sync happened while we were committing, we |
| // must carry over the commit gen: |
| SegmentInfos curInfos = mgr.getCurrentInfos(); |
| if (curInfos != infos) { |
| curInfos.updateGeneration(infos); |
| } |
| } |
| |
| TreeLogger.log("top: id=" + id + " " + segmentsFileName + " committed; now decRef lastCommitFiles=" + lastCommitFiles); |
| |
| deleter.decRef(lastCommitFiles); |
| lastCommitFiles.clear(); |
| lastCommitFiles.addAll(newFiles); |
| lastCommitFiles.add(segmentsFileName); |
| TreeLogger.log("id=" + id + " " + infos.getSegmentsFileName() + " lastCommitFiles=" + lastCommitFiles); |
| |
| if (deleteAll) { |
| // nocommit this is messy: we may delete a merge's files |
| // that just copied over as we closed the writer: |
| TreeLogger.log("id=" + id + " now deleteUnknownFiles during commit"); |
| deleter.deleteUnknownFiles(); |
| TreeLogger.log("id=" + id + " done deleteUnknownFiles during commit"); |
| } |
| |
| TreeLogger.end("commit"); |
| } |
| } |
| } |
| |
| static class CopyFileJob { |
| public final CopyResult result; |
| public final Directory src; |
| public final FileMetaData metaData; |
| public final String fileName; |
| public final IOContext context; |
| |
| public CopyFileJob(CopyResult result, IOContext context, Directory src, String fileName, FileMetaData metaData) { |
| this.result = result; |
| this.context = context; |
| this.src = src; |
| this.fileName = fileName; |
| this.metaData = metaData; |
| } |
| } |
| |
| static class CopyOneFile implements Closeable { |
| private long bytesLeft; |
| |
| // TODO: reuse...? |
| private final byte[] buffer = new byte[65536]; |
| |
| private final IndexInput in; |
| private final IndexOutput out; |
| private final SlowChecksumDirectory dest; |
| |
| public final CopyFileJob job; |
| public boolean failed; |
| |
| public CopyOneFile(CopyFileJob job, SlowChecksumDirectory dest, RateLimiter rateLimiter) throws IOException { |
| this.job = job; |
| this.dest = dest; |
| in = job.src.openInput(job.fileName, job.context); |
| boolean success = false; |
| try { |
| IndexOutput out0 = dest.createOutput(job.fileName, job.context); |
| if (rateLimiter == null) { |
| // No IO rate limiting |
| out = out0; |
| } else { |
| out = new RateLimitedIndexOutput(rateLimiter, out0); |
| } |
| success = true; |
| } finally { |
| if (success == false) { |
| IOUtils.closeWhileHandlingException(in); |
| } |
| } |
| |
| bytesLeft = job.metaData.length; |
| } |
| |
| public boolean visit() throws IOException { |
| int chunk = bytesLeft > buffer.length ? buffer.length : (int) bytesLeft; |
| try { |
| in.readBytes(buffer, 0, chunk); |
| out.writeBytes(buffer, 0, chunk); |
| } catch (Exception e) { |
| TreeLogger.log("failed to copy " + job.fileName, e); |
| failed = true; |
| } |
| bytesLeft -= chunk; |
| return bytesLeft != 0; |
| } |
| |
| /** Abort the copy. */ |
| public void abort() { |
| TreeLogger.log("now abort copy file " + job.fileName); |
| try { |
| close(); |
| } catch (IOException ioe) { |
| } |
| try { |
| dest.deleteFile(job.fileName); |
| } catch (IOException ioe) { |
| } |
| } |
| |
| @Override |
| public void close() throws IOException { |
| IOUtils.close(in, out); |
| if (failed) { |
| dest.deleteFile(job.fileName); |
| throw new IOException("copy failed"); |
| } |
| |
| Long actual = dest.getChecksum(job.fileName); |
| assert actual != null; |
| |
| if (actual.longValue() != job.metaData.checksum) { |
| // Uh oh, bits flipped during copy: |
| dest.deleteFile(job.fileName); |
| throw new IOException("checksum mismatch"); |
| } |
| } |
| } |
| |
| static class CopyResult { |
| public final CountDownLatch done; |
| public final AtomicBoolean failed = new AtomicBoolean(); |
| |
| public CopyResult(int fileCount) { |
| done = new CountDownLatch(fileCount); |
| } |
| } |
| |
| // TODO: abstract this, enable swapping in different |
| // low-level tools ... rsync, bittorrent, etc. |
| |
| /** Simple class to copy low (merged segments) & high |
| * (flushes) priority files. It has a simplistic |
| * "ionice" implementation: if we are copying a low-pri |
| * (merge) file and a high-pri (flush) job shows up, then |
| * we pause the low-pri copy and finish all high-pri |
| * copies, then resume it. */ |
| |
| static class SimpleFileCopier extends Thread implements Closeable { |
| |
| private final Queue<CopyFileJob> highPriorityJobs = new ConcurrentLinkedQueue<>(); |
| private final Queue<CopyFileJob> lowPriorityJobs = new ConcurrentLinkedQueue<>(); |
| private final SlowChecksumDirectory dest; |
| |
| // nocommit make rate limit (10 MB/sec now) controllable: |
| private final RateLimiter mergeRateLimiter = new RateLimiter.SimpleRateLimiter(10.0); |
| private final int id; |
| private boolean stop; |
| |
| /** We always copy files into this dest. */ |
| public SimpleFileCopier(SlowChecksumDirectory dest, int id) { |
| this.dest = dest; |
| this.id = id; |
| } |
| |
| // nocommit use Future here |
| public synchronized CopyResult add(SlowChecksumDirectory src, Map<String,FileMetaData> files, boolean lowPriority, IOContext context) { |
| if (stop) { |
| throw new AlreadyClosedException("closed"); |
| } |
| CopyResult result = new CopyResult(files.size()); |
| Queue<CopyFileJob> queue = lowPriority ? lowPriorityJobs : highPriorityJobs; |
| for(Map.Entry<String,FileMetaData> ent : files.entrySet()) { |
| queue.add(new CopyFileJob(result, context, src, ent.getKey(), ent.getValue())); |
| } |
| notify(); |
| return result; |
| } |
| |
| @Override |
| public void run() { |
| Thread.currentThread().setName("fileCopier id=" + id); |
| TreeLogger.setLogger(new TreeLogger("fileCopier id=" + id)); |
| |
| CopyOneFile curLowPri = null; |
| CopyOneFile curHighPri = null; |
| |
| boolean success = false; |
| |
| try { |
| while (stop == false) { |
| if (curHighPri != null) { |
| if (curHighPri.visit() == false) { |
| TreeLogger.log("id=" + id + " " + curHighPri.job.fileName + " high-priority copy done"); |
| // Finished copying; now close & verify checksums: |
| try { |
| curHighPri.close(); |
| } catch (IOException ioe) { |
| curHighPri.job.result.failed.set(true); |
| TreeLogger.log("WARNING: id=" + id + " " + curHighPri.job.fileName + ": failed to copy high-priority file"); |
| } finally { |
| curHighPri.job.result.done.countDown(); |
| } |
| curHighPri = null; |
| } |
| } else if (highPriorityJobs.isEmpty() == false) { |
| // Start new high-priority copy: |
| CopyFileJob job = highPriorityJobs.poll(); |
| try { |
| curHighPri = new CopyOneFile(job, dest, null); |
| TreeLogger.log("id=" + id + " " + curHighPri.job.fileName + " now start high-priority copy"); |
| } catch (AlreadyClosedException ace) { |
| TreeLogger.log("id=" + id + " " + job.fileName + " skip copy: hit AlreadyClosedException"); |
| job.result.failed.set(true); |
| job.result.done.countDown(); |
| } |
| } else if (curLowPri != null) { |
| if (curLowPri.visit() == false) { |
| TreeLogger.log("id=" + id + " " + curLowPri.job.fileName + " low-priority copy done"); |
| // Finished copying; now close & verify checksums: |
| try { |
| curLowPri.close(); |
| } catch (IOException ioe) { |
| TreeLogger.log("WARNING: id=" + id + " " + curLowPri.job.fileName + ": failed to copy low-priority file"); |
| curLowPri.job.result.failed.set(true); |
| } finally { |
| curLowPri.job.result.done.countDown(); |
| } |
| curLowPri = null; |
| } |
| } else if (lowPriorityJobs.isEmpty() == false) { |
| // Start new low-priority copy: |
| CopyFileJob job = lowPriorityJobs.poll(); |
| try { |
| curLowPri = new CopyOneFile(job, dest, mergeRateLimiter); |
| TreeLogger.log("id=" + id + " " + curLowPri.job.fileName + " now start low-priority copy"); |
| } catch (AlreadyClosedException ace) { |
| TreeLogger.log("id=" + id + " " + job.fileName + " skip copy: hit AlreadyClosedException"); |
| job.result.failed.set(true); |
| job.result.done.countDown(); |
| } |
| } else { |
| // Wait for another job: |
| synchronized (this) { |
| if (highPriorityJobs.isEmpty() && lowPriorityJobs.isEmpty()) { |
| TreeLogger.log("id=" + id + " copy thread now idle"); |
| try { |
| wait(); |
| } catch (InterruptedException ie) { |
| Thread.currentThread().interrupt(); |
| } |
| TreeLogger.log("id=" + id + " copy thread now wake up"); |
| } |
| } |
| } |
| } |
| success = true; |
| } catch (IOException ioe) { |
| // nocommit catch each op & retry instead? |
| throw new RuntimeException(ioe); |
| } finally { |
| synchronized(this) { |
| stop = true; |
| if (curLowPri != null) { |
| curLowPri.abort(); |
| curLowPri.job.result.failed.set(true); |
| curLowPri.job.result.done.countDown(); |
| } |
| if (curHighPri != null) { |
| curHighPri.abort(); |
| curHighPri.job.result.failed.set(true); |
| curHighPri.job.result.done.countDown(); |
| } |
| |
| for(CopyFileJob job : highPriorityJobs) { |
| job.result.failed.set(true); |
| job.result.done.countDown(); |
| } |
| |
| for(CopyFileJob job : lowPriorityJobs) { |
| job.result.failed.set(true); |
| job.result.done.countDown(); |
| } |
| } |
| } |
| } |
| |
| // nocommit cutover all other finishes to closeable |
| |
| @Override |
| public synchronized void close() throws IOException { |
| stop = true; |
| try { |
| notify(); |
| join(); |
| } catch (InterruptedException ie) { |
| // nocommit |
| throw new IOException(ie); |
| } |
| } |
| } |
| |
| // TODO: we could pre-copy merged files out event before |
| // warmMerge is called? e.g. if we "notice" files being |
| // written to the dir ... could give us a "head start" |
| |
| /** Runs in each replica, to handle copying over new |
| * flushes. */ |
| static class CopyThread extends Thread { |
| final Lock lock; |
| final Condition cond; |
| private final Replica replica; |
| private volatile boolean stop; |
| |
| public CopyThread(Replica replica) { |
| this.lock = new ReentrantLock(); |
| this.cond = lock.newCondition(); |
| this.stop = false; |
| this.replica = replica; |
| } |
| |
| @Override |
| public void run() { |
| Thread.currentThread().setName("replica id=" + replica.id); |
| TreeLogger.setLogger(new TreeLogger("replica id=" + replica.id)); |
| |
| try { |
| // While loop to keep pulling newly flushed/merged |
| // segments until we shutdown |
| while (stop == false) { |
| |
| try { |
| |
| // While loop to pull a single new segment: |
| long curVersion = replica.mgr.getCurrentInfos().version; |
| |
| SegmentInfos newInfos = null; |
| CopyState copyState = null; |
| Master curMaster = null; |
| int curMasterMoveCount = -1; |
| while (stop == false) { |
| lock.lock(); |
| try { |
| curMaster = master; |
| if (curMaster != null) { |
| copyState = curMaster.getCopyState(); |
| curMasterMoveCount = masterCount.get(); |
| if (copyState != null) { |
| assert copyState.version >= curVersion: "copyState.version=" + copyState.version + " vs curVersion=" + curVersion; |
| if (copyState.version > curVersion) { |
| TreeLogger.log("got new copyState"); |
| break; |
| } else { |
| TreeLogger.log("skip newInfos"); |
| curMaster.releaseCopyState(copyState); |
| copyState = null; |
| } |
| } else { |
| // Master is closed |
| Thread.sleep(5); |
| } |
| } else { |
| // Master hasn't started yet |
| Thread.sleep(5); |
| } |
| cond.awaitUninterruptibly(); |
| } finally { |
| lock.unlock(); |
| } |
| } |
| |
| // We hold a ref on newInfos at this point |
| |
| // nocommit what if master closes now? |
| |
| if (stop) { |
| if (copyState != null) { |
| curMaster.releaseCopyState(copyState); |
| } |
| break; |
| } |
| |
| // nocommit need to sometimes crash during |
| // replication; ooh: we could just Thread.interrupt()? |
| |
| // nocommit needs to cross the "wire", ie turn into |
| // byte[] and back, and manage the "reservation" |
| // separately on master |
| |
| try { |
| // nocommit just pass copyState |
| replica.sync(curMasterMoveCount, copyState.dir, copyState.files, copyState.infosBytes, copyState.version); |
| } finally { |
| curMaster.releaseCopyState(copyState); |
| } |
| } catch (AlreadyClosedException ace) { |
| // OK: master closed while we were replicating; |
| // we will just retry again against the next master: |
| TreeLogger.log("top: id=" + replica.id + " ignore AlreadyClosedException"); |
| Thread.sleep(5); |
| continue; |
| } |
| } |
| } catch (InterruptedException ie) { |
| throw new RuntimeException(ie); |
| } catch (IOException ioe) { |
| // nocommit how to properly handle... |
| throw new RuntimeException(ioe); |
| } |
| } |
| |
| /** Shuts down the thread and only returns once |
| * it's done. */ |
| public void finish() throws InterruptedException { |
| stop = true; |
| lock.lock(); |
| try { |
| cond.signal(); |
| } finally { |
| lock.unlock(); |
| } |
| join(); |
| } |
| } |
| |
| static class SearchThread extends Thread { |
| private volatile boolean stop; |
| private final InfosSearcherManager mgr; |
| private final Map<Long,Integer> versionDocCounts; |
| private final String id; |
| |
| public SearchThread(String id, InfosSearcherManager mgr, Map<Long,Integer> versionDocCounts) { |
| this.id = id; |
| this.mgr = mgr; |
| this.versionDocCounts = versionDocCounts; |
| setName("SearchThread id=" + id); |
| } |
| |
| @Override |
| public void run() { |
| try { |
| while (stop == false) { |
| IndexSearcher s = mgr.acquire(); |
| try { |
| // Sleep so that sometimes our searcher is "stale": |
| Thread.sleep(_TestUtil.nextInt(random(), 1, 10)); |
| // nocommit do more interesting searches |
| int totalHits = s.search(new MatchAllDocsQuery(), 10).totalHits; |
| if (totalHits > 0) { |
| long version = ((DirectoryReader) s.getIndexReader()).getVersion(); |
| Integer expectedCount = versionDocCounts.get(version); |
| assertNotNull("searcher " + s + " version=" + version + " is missing expected count", expectedCount); |
| // nocommit since master may roll back in time |
| // we cannot assert this: |
| //assertEquals("searcher version=" + version + " replica id=" + id + " searcher=" + s, expectedCount.intValue(), totalHits); |
| } |
| } finally { |
| mgr.release(s); |
| } |
| } |
| } catch (Exception e) { |
| System.out.println("FAILED: id=" + id); |
| e.printStackTrace(System.out); |
| throw new RuntimeException(e); |
| } |
| } |
| |
| public void finish() throws InterruptedException { |
| stop = true; |
| join(); |
| } |
| } |
| |
| static int docCount(SegmentInfos infos) { |
| int totDocCount = 0; |
| for(SegmentCommitInfo info : infos) { |
| totDocCount += info.info.getDocCount() - info.getDelCount(); |
| } |
| return totDocCount; |
| } |
| |
| // nocommit factor/share with IFD |
| static class InfosRefCounts { |
| private final Map<String,Integer> refCounts = new HashMap<String,Integer>(); |
| private final Set<String> pending = new HashSet<String>(); |
| private final Directory dir; |
| private final int id; |
| |
| public InfosRefCounts(int id, Directory dir) throws IOException { |
| this.dir = dir; |
| this.id = id; |
| } |
| |
| public synchronized void incRef(Collection<String> fileNames) { |
| for(String fileName : fileNames) { |
| Integer curCount = refCounts.get(fileName); |
| if (curCount == null) { |
| refCounts.put(fileName, 1); |
| } else { |
| refCounts.put(fileName, curCount.intValue() + 1); |
| } |
| |
| // Necessary in case we had tried to delete this |
| // fileName before, it failed, but then it was later |
| // overwritten: |
| pending.remove(fileName); |
| } |
| } |
| |
| public synchronized void decRef(Collection<String> fileNames) { |
| for(String fileName : fileNames) { |
| Integer curCount = refCounts.get(fileName); |
| assert curCount != null: "fileName=" + fileName; |
| assert curCount.intValue() > 0; |
| if (curCount.intValue() == 1) { |
| refCounts.remove(fileName); |
| delete(fileName); |
| } else { |
| refCounts.put(fileName, curCount.intValue() - 1); |
| } |
| } |
| } |
| |
| private synchronized void delete(String fileName) { |
| try { |
| TreeLogger.log("id=" + id + " " + fileName + " now delete"); |
| dir.deleteFile(fileName); |
| } catch (IOException ioe) { |
| // nocommit why do we keep trying to delete the file |
| // "segments" ... |
| TreeLogger.log("id=" + id + " " + fileName + " delete failed; will retry later"); |
| pending.add(fileName); |
| } |
| } |
| |
| public synchronized void deletePending() { |
| Set<String> copy = new HashSet<String>(pending); |
| pending.clear(); |
| for(String fileName : copy) { |
| delete(fileName); |
| } |
| } |
| |
| public synchronized void deleteUnknownFiles() throws IOException { |
| for(String fileName : dir.listAll()) { |
| if (refCounts.containsKey(fileName) == false && |
| fileName.startsWith("segments") == false && |
| fileName.startsWith("checksum") == false) { |
| TreeLogger.log("id=" + id + " delete unknown file \"" + fileName + "\""); |
| delete(fileName); |
| } |
| } |
| } |
| } |
| } |
