[exo-jcr-commits] exo-jcr SVN: r3288 - in jcr/branches/1.14-CNK/exo.jcr.component.core/src: main/java/org/exoplatform/services/jcr/impl/core/query/lucene and 1 other directories.

Author: nzamosenchuk Date: 2010-10-11 08:17:47 -0400 (Mon, 11 Oct 2010) New Revision: 3288 Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/JBossCacheIndexInfos.java jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/MultiIndex.java jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/api/core/query/lucene/IndexingAggregateTest.java Log: EXOJCR-987 Getting rid of Volatile index Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/JBossCacheIndexInfos.java =================================================================== --- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/JBossCacheIndexInfos.java 2010-10-11 11:27:31 UTC (rev 3287) +++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/JBossCacheIndexInfos.java 2010-10-11 12:17:47 UTC (rev 3288) @@ -193,18 +193,18 @@ { setNames(set); // callback multiIndex to refresh lists - try - { - MultiIndex multiIndex = getMultiIndex(); - if (multiIndex != null) - { - multiIndex.refreshIndexList(); - } - } - catch (IOException e) - { - log.error("Failed to update indexes! " + e.getMessage(), e); - } +// try +// { +// MultiIndex multiIndex = getMultiIndex(); +// if (multiIndex != null) +// { +// //multiIndex.refreshIndexList(); +// } +// } +// catch (IOException e) +// { +// log.error("Failed to update indexes! " + e.getMessage(), e); +// } } } } Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/MultiIndex.java =================================================================== --- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/MultiIndex.java 2010-10-11 11:27:31 UTC (rev 3287) +++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/MultiIndex.java 2010-10-11 12:17:47 UTC (rev 3288) @@ -25,7 +25,6 @@ import org.exoplatform.services.jcr.datamodel.NodeData; import org.exoplatform.services.jcr.impl.core.query.IndexerIoMode; import org.exoplatform.services.jcr.impl.core.query.IndexerIoModeHandler; -import org.exoplatform.services.jcr.impl.core.query.IndexerIoModeListener; import org.exoplatform.services.jcr.impl.core.query.IndexingTree; import org.exoplatform.services.jcr.impl.core.query.lucene.directory.DirectoryManager; import org.exoplatform.services.jcr.impl.util.SecurityHelper; @@ -46,8 +45,7 @@ import java.util.List; import java.util.Map; import java.util.Set; -import java.util.Timer; -import java.util.TimerTask; +import java.util.Map.Entry; import javax.jcr.ItemNotFoundException; import javax.jcr.RepositoryException; @@ -79,7 +77,7 @@ * thread and reader threads is done using {@link #updateMonitor} and * {@link #updateInProgress}. */ -public class MultiIndex implements IndexerIoModeListener, IndexUpdateMonitorListener +public class MultiIndex { /** @@ -93,17 +91,12 @@ private IndexInfos indexNames; /** - * Names of index directories that can be deleted. - */ - private final Set deletable = new HashSet(); - - /** * List of open persistent indexes. This list may also contain an open * PersistentIndex owned by the IndexMerger daemon. Such an index is not * registered with indexNames and <b>must not</b> be used in regular index * operations (delete node, etc.)! */ - private final List indexes = new ArrayList(); + private final List<PersistentIndex> indexes = new ArrayList<PersistentIndex>(); /** * The internal namespace mappings of the query manager. @@ -126,18 +119,6 @@ private final SearchIndex handler; /** - * The volatile index. - */ - private VolatileIndex volatileIndex; - - /** - * Flag indicating whether an update operation is in progress. - */ - // private boolean updateInProgress = false; - - private final IndexUpdateMonitor indexUpdateMonitor; - - /** * If not <code>null</code> points to a valid <code>IndexReader</code> that * reads from all indexes, including volatile and persistent indexes. */ @@ -149,33 +130,6 @@ private final DocNumberCache cache; /** - * Monitor to use to synchronize access to {@link #multiReader} and - * {@link #updateInProgress}. - */ - private final Object updateMonitor = new Object(); - - /** - * The time this index was last flushed or a transaction was committed. - */ - private long lastFlushTime; - - /** - * The time this index was last flushed or a transaction was committed. - */ - private long lastFileSystemFlushTime; - - /** - * Timer to schedule flushes of this index after some idle time. - */ - private static final Timer FLUSH_TIMER = new Timer(true); - - /** - * Task that is periodically called by {@link #FLUSH_TIMER} and checks if - * index should be flushed. - */ - private TimerTask flushTask; - - /** * The indexing queue with pending text extraction jobs. */ private IndexingQueue indexingQueue; @@ -186,11 +140,6 @@ private final IndexingTree indexingTree; /** - * The next transaction id. - */ - private long nextTransactionId = 0; - - /** * Flag indicating whether re-indexing is running. */ private boolean reindexing = false; @@ -220,7 +169,6 @@ IndexUpdateMonitor indexUpdateMonitor) throws IOException { this.modeHandler = modeHandler; - this.indexUpdateMonitor = indexUpdateMonitor; this.directoryManager = handler.getDirectoryManager(); // this method is run in privileged mode internally this.indexDir = directoryManager.getDirectory("."); @@ -228,18 +176,15 @@ this.cache = new DocNumberCache(handler.getCacheSize()); this.indexingTree = indexingTree; this.nsMappings = handler.getNamespaceMappings(); - this.flushTask = null; this.indexNames = indexInfos; this.indexNames.setDirectory(indexDir); // this method is run in privileged mode internally this.indexNames.read(); + this.indexNames.addName("1"); + this.indexNames.addName("2"); + this.indexNames.addName("3"); + this.indexNames.addName("4"); - this.lastFileSystemFlushTime = System.currentTimeMillis(); - this.lastFlushTime = System.currentTimeMillis(); - - modeHandler.addIndexerIoModeListener(this); - indexUpdateMonitor.addIndexUpdateMonitorListener(this); - // this method is run in privileged mode internally // as of 1.5 deletable file is not used anymore removeDeletable(); @@ -251,6 +196,7 @@ this.indexingQueue = new IndexingQueue(store); // copy current index names Set<String> currentNames = new HashSet<String>(indexNames.getNames()); + //Set<String> currentNames = new HashSet<String>(); // open persistent indexes for (String name : currentNames) @@ -259,12 +205,14 @@ // it is possible that indexNames still contains a name for // an index that has been deleted, but indexNames has not been // written to disk. - if (!directoryManager.hasDirectory(name)) - { - log.debug("index does not exist anymore: " + name); - // move on to next index - continue; - } + + // TODO THIS CHECK WAS SKIPPED + // if (!directoryManager.hasDirectory(name)) + // { + // log.debug("index does not exist anymore: " + name); + // // move on to next index + // continue; + // } PersistentIndex index = new PersistentIndex(name, handler.getTextAnalyzer(), handler.getSimilarity(), cache, indexingQueue, directoryManager); @@ -275,7 +223,8 @@ } // init volatile index - resetVolatileIndex(); + // TODO commented line below: + //resetVolatileIndex(); // set index format version and at the same time // initialize hierarchy cache if requested. @@ -296,10 +245,6 @@ }); } indexingQueue.initialize(this); - if (modeHandler.getMode() == IndexerIoMode.READ_WRITE) - { - setReadWrite(); - } this.indexNames.setMultiIndex(this); } @@ -314,7 +259,9 @@ { if (indexNames.size() == 0) { - return volatileIndex.getNumDocuments(); + // TODO commented line below: + //return volatileIndex.getNumDocuments(); + return 0; } else { @@ -362,15 +309,16 @@ */ void createInitialIndex(ItemDataConsumer stateMgr) throws IOException { + + // TODO: re-study check!!! // only do an initial index if there are no indexes at all - if (indexNames.size() == 0) + // if (indexNames.size() == 0) { reindexing = true; try { long count = 0; // traverse and index workspace - // TODO: this was removed //executeAndLog(new Start(Action.INTERNAL_TRANSACTION)); @@ -382,7 +330,8 @@ log.info("Created initial index for {} nodes", new Long(count)); releaseMultiReader(); - scheduleFlushTask(); + // TODO commented line below: + //scheduleFlushTask(); } catch (Exception e) { @@ -396,10 +345,10 @@ reindexing = false; } } - else - { - throw new IllegalStateException("Index already present"); - } + // else + // { + // throw new IllegalStateException("Index already present"); + // } } /** @@ -416,7 +365,7 @@ * @throws IOException * if an error occurs while updating the index. */ - synchronized void update(final Collection remove, final Collection add) throws IOException + synchronized void update(final Collection<String> remove, final Collection<Document> add) throws IOException { SecurityHelper.doPriviledgedIOExceptionAction(new PrivilegedExceptionAction<Object>() { @@ -435,26 +384,18 @@ log.warn("unable to prepare index reader " + "for queries during update", e); } } - - synchronized (updateMonitor) - { - //updateInProgress = true; - indexUpdateMonitor.setUpdateInProgress(true, false); - } - boolean flush = false; +// boolean flush = false; try { - long transactionId = nextTransactionId++; - // TODO: this was removed //executeAndLog(new Start(transactionId)); - for (Iterator it = remove.iterator(); it.hasNext();) + for (Iterator<String> it = remove.iterator(); it.hasNext();) { // TODO this was replaced //executeAndLog(new DeleteNode(transactionId, (String)it.next())); - String uuidString = (String)it.next(); + String uuidString = it.next(); // check if indexing queue is still working on // this node from a previous update Document doc = indexingQueue.removeDocument(uuidString); @@ -465,13 +406,15 @@ Term idTerm = new Term(FieldNames.UUID, uuidString); // if the document cannot be deleted from the volatile index // delete it from one of the persistent indexes. - int num = volatileIndex.removeDocument(idTerm); + + // TODO watch here + int num = getChunk(uuidString).removeDocument(idTerm); if (num == 0) { for (int i = indexes.size() - 1; i >= 0; i--) { // only look in registered indexes - PersistentIndex idx = (PersistentIndex)indexes.get(i); + PersistentIndex idx = indexes.get(i); if (indexNames.contains(idx.getName())) { num = idx.removeDocument(idTerm); @@ -484,9 +427,9 @@ } } - for (Iterator it = add.iterator(); it.hasNext();) + for (Iterator<Document> it = add.iterator(); it.hasNext();) { - Document doc = (Document)it.next(); + Document doc = it.next(); if (doc != null) { // TODO: ths is replaced @@ -494,38 +437,27 @@ if (doc != null) { - volatileIndex.addDocuments(new Document[]{doc}); + // TODO watch this + String uuid = doc.get(FieldNames.UUID); + getChunk(uuid).addDocuments(new Document[]{doc}); } // commit volatile index if needed - flush |= checkVolatileCommit(); + // TODO commented line below: + //flush |= checkVolatileCommit(); } } // TODO : this was replaced //executeAndLog(new Commit(transactionId)); - // flush whole index when volatile index has been commited. - if (flush) - { - // if we are going to flush, need to set persistent update - synchronized (updateMonitor) - { - indexUpdateMonitor.setUpdateInProgress(true, true); - } - flush(); - } + // TODO commented line below: + //flush(); + getChunk(null).commit(); } finally { - synchronized (updateMonitor) - { - //updateInProgress = false; - - indexUpdateMonitor.setUpdateInProgress(false, flush); - updateMonitor.notifyAll(); - releaseMultiReader(); - } + releaseMultiReader(); } return null; } @@ -569,11 +501,6 @@ */ synchronized int removeAllDocuments(String uuid) throws IOException { - synchronized (updateMonitor) - { - //updateInProgress = true; - indexUpdateMonitor.setUpdateInProgress(true, false); - } int num; try { @@ -582,7 +509,8 @@ // TODO: this was removed //executeAndLog(new Start(Action.INTERNAL_TRANSACTION)); - num = volatileIndex.removeDocument(idTerm); + // TODO REMOVE WAS HERE + num = getChunk(uuid).removeDocument(idTerm); if (num > 0) { // TODO: removed line @@ -590,7 +518,7 @@ } for (int i = 0; i < indexes.size(); i++) { - PersistentIndex index = (PersistentIndex)indexes.get(i); + PersistentIndex index = indexes.get(i); // only remove documents from registered indexes if (indexNames.contains(index.getName())) { @@ -608,13 +536,7 @@ } finally { - synchronized (updateMonitor) - { - //updateInProgress = false; - indexUpdateMonitor.setUpdateInProgress(false, false); - updateMonitor.notifyAll(); - releaseMultiReader(); - } + releaseMultiReader(); } return num; } @@ -639,14 +561,14 @@ */ synchronized IndexReader[] getIndexReaders(String[] indexNames, IndexListener listener) throws IOException { - Set names = new HashSet(Arrays.asList(indexNames)); - Map indexReaders = new HashMap(); + Set<String> names = new HashSet<String>(Arrays.asList(indexNames)); + Map<ReadOnlyIndexReader, PersistentIndex> indexReaders = new HashMap<ReadOnlyIndexReader, PersistentIndex>(); try { - for (Iterator it = indexes.iterator(); it.hasNext();) + for (Iterator<PersistentIndex> it = indexes.iterator(); it.hasNext();) { - PersistentIndex index = (PersistentIndex)it.next(); + PersistentIndex index = it.next(); if (names.contains(index.getName())) { indexReaders.put(index.getReadOnlyIndexReader(listener), index); @@ -656,10 +578,10 @@ catch (IOException e) { // release readers obtained so far - for (Iterator it = indexReaders.entrySet().iterator(); it.hasNext();) + for (Iterator<Entry<ReadOnlyIndexReader, PersistentIndex>> it = indexReaders.entrySet().iterator(); it.hasNext();) { - Map.Entry entry = (Map.Entry)it.next(); - final ReadOnlyIndexReader reader = (ReadOnlyIndexReader)entry.getKey(); + Map.Entry<ReadOnlyIndexReader, PersistentIndex> entry = it.next(); + final ReadOnlyIndexReader reader = entry.getKey(); try { SecurityHelper.doPriviledgedIOExceptionAction(new PrivilegedExceptionAction<Object>() @@ -675,77 +597,15 @@ { log.warn("Exception releasing index reader: " + ex); } - ((PersistentIndex)entry.getValue()).resetListener(); + (entry.getValue()).resetListener(); } throw e; } - return (IndexReader[])indexReaders.keySet().toArray(new IndexReader[indexReaders.size()]); + return indexReaders.keySet().toArray(new IndexReader[indexReaders.size()]); } /** - * Creates a new Persistent index. The new index is not registered with this - * <code>MultiIndex</code>. - * - * @param indexName - * the name of the index to open, or <code>null</code> if an - * index with a new name should be created. - * @return a new <code>PersistentIndex</code>. - * @throws IOException - * if a new index cannot be created. - */ - synchronized PersistentIndex getOrCreateIndex(String indexName) throws IOException - { - // check existing - for (Iterator it = indexes.iterator(); it.hasNext();) - { - PersistentIndex idx = (PersistentIndex)it.next(); - if (idx.getName().equals(indexName)) - { - return idx; - } - } - - if (modeHandler.getMode() == IndexerIoMode.READ_ONLY) - { - throw new UnsupportedOperationException("Can't create index in READ_ONLY mode."); - } - - // otherwise open / create it - if (indexName == null) - { - do - { - indexName = indexNames.newName(); - } - while (directoryManager.hasDirectory(indexName)); - } - PersistentIndex index; - try - { - index = - new PersistentIndex(indexName, handler.getTextAnalyzer(), handler.getSimilarity(), cache, indexingQueue, - directoryManager); - } - catch (IOException e) - { - // do some clean up - if (!directoryManager.delete(indexName)) - { - deletable.add(indexName); - } - throw e; - } - index.setMaxFieldLength(handler.getMaxFieldLength()); - index.setUseCompoundFile(handler.getUseCompoundFile()); - index.setTermInfosIndexDivisor(handler.getTermInfosIndexDivisor()); - - // add to list of open indexes and return it - indexes.add(index); - return index; - } - - /** * Returns <code>true</code> if this multi index has an index segment with * the given name. This method even returns <code>true</code> if an index * segments has not yet been loaded / initialized but exists on disk. @@ -759,9 +619,9 @@ synchronized boolean hasIndex(String indexName) throws IOException { // check existing - for (Iterator it = indexes.iterator(); it.hasNext();) + for (Iterator<PersistentIndex> it = indexes.iterator(); it.hasNext();) { - PersistentIndex idx = (PersistentIndex)it.next(); + PersistentIndex idx = it.next(); if (idx.getName().equals(indexName)) { return true; @@ -772,131 +632,6 @@ } /** - * Replaces the indexes with names <code>obsoleteIndexes</code> with - * <code>index</code>. Documents that must be deleted in <code>index</code> - * can be identified with <code>Term</code>s in <code>deleted</code>. - * - * @param obsoleteIndexes - * the names of the indexes to replace. - * @param index - * the new index that is the result of a merge of the indexes to - * replace. - * @param deleted - * <code>Term</code>s that identify documents that must be - * deleted in <code>index</code>. - * @throws IOException - * if an exception occurs while replacing the indexes. - */ - void replaceIndexes(String[] obsoleteIndexes, final PersistentIndex index, Collection deleted) throws IOException - { - - if (handler.isInitializeHierarchyCache()) - { - // force initializing of caches - long time = System.currentTimeMillis(); - SecurityHelper.doPriviledgedIOExceptionAction(new PrivilegedExceptionAction<Object>() - { - public Object run() throws Exception - { - index.getReadOnlyIndexReader(true).release(); - return null; - } - }); - time = System.currentTimeMillis() - time; - log.debug("hierarchy cache initialized in {} ms", new Long(time)); - } - - synchronized (this) - { - synchronized (updateMonitor) - { - //updateInProgress = true; - indexUpdateMonitor.setUpdateInProgress(true, true); - } - try - { - // if we are reindexing there is already an active transaction - if (!reindexing) - { - // TODO: this was removed - //executeAndLog(new Start(Action.INTERNAL_TRANS_REPL_INDEXES)); - } - // delete obsolete indexes - Set names = new HashSet(Arrays.asList(obsoleteIndexes)); - for (Iterator it = names.iterator(); it.hasNext();) - { - // do not try to delete indexes that are already gone - String indexName = (String)it.next(); - if (indexNames.contains(indexName)) - { - // TODO this was replaced - //executeAndLog(new DeleteIndex(getTransactionId(), indexName)); - - for (Iterator it2 = indexes.iterator(); it2.hasNext();) - { - PersistentIndex idx = (PersistentIndex)it2.next(); - if (idx.getName().equals(indexName)) - { - idx.close(); - deleteIndex(idx); - break; - } - } - - } - } - - // Index merger does not log an action when it creates the - // target - // index of the merge. We have to do this here. - // TODO this was replaced - ///executeAndLog(new CreateIndex(getTransactionId(), index.getName())); - - PersistentIndex idx = getOrCreateIndex(index.getName()); - - // TODO: this is replaced - //executeAndLog(new AddIndex(getTransactionId(), index.getName())); - if (!indexNames.contains(index.getName())) - { - indexNames.addName(index.getName()); - } - - // delete documents in index - for (Iterator it = deleted.iterator(); it.hasNext();) - { - Term id = (Term)it.next(); - index.removeDocument(id); - } - index.commit(); - - if (!reindexing) - { - // only commit if we are not reindexing - // when reindexing the final commit is done at the very end - - // TODO : this was replaced - //executeAndLog(new Commit(getTransactionId())); - } - } - finally - { - synchronized (updateMonitor) - { - //updateInProgress = false; - indexUpdateMonitor.setUpdateInProgress(false, true); - updateMonitor.notifyAll(); - releaseMultiReader(); - } - } - } - if (reindexing) - { - // do some cleanup right away when reindexing - attemptDelete(); - } - } - - /** * Returns an read-only <code>IndexReader</code> that spans alls indexes of * this <code>MultiIndex</code>. * @@ -926,74 +661,49 @@ { public CachingMultiIndexReader run() throws Exception { - synchronized (updateMonitor) + if (multiReader != null) { - if (multiReader != null) + multiReader.acquire(); + return multiReader; + } + // no reader available + // some other read thread might have created the reader in the + // meantime -> check again + if (multiReader == null) + { + List<ReadOnlyIndexReader> readerList = new ArrayList<ReadOnlyIndexReader>(); + for (int i = 0; i < indexes.size(); i++) { - multiReader.acquire(); - return multiReader; - } - // no reader available - // wait until no update is in progress - while (indexUpdateMonitor.getUpdateInProgress()) - { - try - { - updateMonitor.wait(); - } - catch (InterruptedException e) - { - throw new IOException("Interrupted while waiting to aquire reader"); - } - } - // some other read thread might have created the reader in the - // meantime -> check again - if (multiReader == null) - { - List readerList = new ArrayList(); - for (int i = 0; i < indexes.size(); i++) - { - PersistentIndex pIdx = (PersistentIndex)indexes.get(i); + PersistentIndex pIdx = indexes.get(i); - if (indexNames.contains(pIdx.getName())) + if (indexNames.contains(pIdx.getName())) + { + try { - try + readerList.add(pIdx.getReadOnlyIndexReader(initCache)); + } + catch (FileNotFoundException e) + { + if (directoryManager.hasDirectory(pIdx.getName())) { - readerList.add(pIdx.getReadOnlyIndexReader(initCache)); + throw e; } - catch (FileNotFoundException e) - { - if (directoryManager.hasDirectory(pIdx.getName())) - { - throw e; - } - } } } - readerList.add(volatileIndex.getReadOnlyIndexReader()); - ReadOnlyIndexReader[] readers = - (ReadOnlyIndexReader[])readerList.toArray(new ReadOnlyIndexReader[readerList.size()]); - multiReader = new CachingMultiIndexReader(readers, cache); } - multiReader.acquire(); - return multiReader; + // TODO commented line below: + //readerList.add(volatileIndex.getReadOnlyIndexReader()); + ReadOnlyIndexReader[] readers = readerList.toArray(new ReadOnlyIndexReader[readerList.size()]); + multiReader = new CachingMultiIndexReader(readers, cache); } + multiReader.acquire(); + return multiReader; } }); } /** - * Returns the volatile index. - * - * @return the volatile index. - */ - VolatileIndex getVolatileIndex() - { - return volatileIndex; - } - - /** * Closes this <code>MultiIndex</code>. */ void close() @@ -1007,11 +717,6 @@ synchronized (this) { - // stop timer - if (flushTask != null) - { - flushTask.cancel(); - } // commit / close indexes try @@ -1022,18 +727,21 @@ { log.error("Exception while closing search index.", e); } - try - { - flush(); - } - catch (IOException e) - { - log.error("Exception while closing search index.", e); - } - volatileIndex.close(); + // try + // { + // // TODO commented line below: + // //flush(); + // } + // catch (IOException e) + // { + // log.error("Exception while closing search index.", e); + // } + // TODO commented line below: + //volatileIndex.close(); + // indexes will be closed now: for (int i = 0; i < indexes.size(); i++) { - ((PersistentIndex)indexes.get(i)).close(); + (indexes.get(i)).close(); } // close indexing queue @@ -1112,83 +820,6 @@ } /** - * Removes the <code>index</code> from the list of active sub indexes. The - * Index is not acutally deleted right away, but postponed to the - * transaction commit. - * <p/> - * This method does not close the index, but rather expects that the index - * has already been closed. - * - * @param index - * the index to delete. - */ - synchronized void deleteIndex(PersistentIndex index) - { - // remove it from the lists if index is registered - indexes.remove(index); - indexNames.removeName(index.getName()); - synchronized (deletable) - { - log.debug("Moved " + index.getName() + " to deletable"); - deletable.add(index.getName()); - } - } - - /** - * Flushes this <code>MultiIndex</code>. Persists all pending changes and - * resets the redo log. - * - * @throws IOException - * if the flush fails. - */ - public void flush() throws IOException - { - synchronized (this) - { - // commit volatile index - // TODO: this was removed - //executeAndLog(new Start(Action.INTERNAL_TRANSACTION)); - commitVolatileIndex(); - - // commit persistent indexes - for (int i = indexes.size() - 1; i >= 0; i--) - { - PersistentIndex index = (PersistentIndex)indexes.get(i); - // only commit indexes we own - // index merger also places PersistentIndex instances in - // indexes, - // but does not make them public by registering the name in - // indexNames - if (indexNames.contains(index.getName())) - { - index.commit(); - // check if index still contains documents - if (index.getNumDocuments() == 0) - { - // TODO THIS was replaced - //executeAndLog(new DeleteIndex(getTransactionId(), index.getName())); - - index.close(); - deleteIndex(index); - - } - } - } - // TODO : this was replaced - //executeAndLog(new Commit(getTransactionId())); - - indexNames.write(); - - // reset redo log - lastFlushTime = System.currentTimeMillis(); - lastFileSystemFlushTime = System.currentTimeMillis(); - } - - // delete obsolete indexes - attemptDelete(); - } - - /** * Releases the {@link #multiReader} and sets it <code>null</code>. If the * reader is already <code>null</code> this method does nothing. When this * method returns {@link #multiReader} is guaranteed to be <code>null</code> @@ -1227,143 +858,6 @@ // >------------------------------------- /** - * Enqueues unused segments for deletion in {@link #deletable}. This method - * does not synchronize on {@link #deletable}! A caller must ensure that it - * is the only one acting on the {@link #deletable} map. - * - * @throws IOException - * if an error occurs while reading directories. - */ - private void enqueueUnusedSegments() throws IOException - { - // walk through index segments - String[] dirNames = directoryManager.getDirectoryNames(); - for (int i = 0; i < dirNames.length; i++) - { - if (dirNames[i].startsWith("_") && !indexNames.contains(dirNames[i])) - { - deletable.add(dirNames[i]); - } - } - } - - /** - * Cancel flush task and add new one - */ - private void scheduleFlushTask() - { - SecurityHelper.doPriviledgedAction(new PrivilegedAction<Object>() - { - public Object run() - { - // cancel task - if (flushTask != null) - { - flushTask.cancel(); - } - // clear canceled tasks - FLUSH_TIMER.purge(); - // new flush task, cause canceled can't be re-used - flushTask = new TimerTask() - { - @Override - public void run() - { - // check if there are any indexing jobs finished - checkIndexingQueue(); - // check if volatile index should be flushed - checkFlush(); - } - }; - FLUSH_TIMER.schedule(flushTask, 0, 1000); - lastFlushTime = System.currentTimeMillis(); - lastFileSystemFlushTime = System.currentTimeMillis(); - return null; - } - }); - } - - /** - * Resets the volatile index to a new instance. - */ - private void resetVolatileIndex() throws IOException - { - volatileIndex = new VolatileIndex(handler.getTextAnalyzer(), handler.getSimilarity(), indexingQueue); - volatileIndex.setUseCompoundFile(handler.getUseCompoundFile()); - volatileIndex.setMaxFieldLength(handler.getMaxFieldLength()); - volatileIndex.setBufferSize(handler.getBufferSize()); - } - - /** - * Checks if it is needed to commit the volatile index according to - * {@link SearchIndex#getMaxVolatileIndexSize()}. - * - * @return <code>true</code> if the volatile index has been committed, - * <code>false</code> otherwise. - * @throws IOException - * if an error occurs while committing the volatile index. - */ - private boolean checkVolatileCommit() throws IOException - { - if (volatileIndex.getRamSizeInBytes() >= handler.getMaxVolatileIndexSize()) - { - commitVolatileIndex(); - return true; - } - return false; - } - - /** - * Commits the volatile index to a persistent index. The new persistent - * index is added to the list of indexes but not written to disk. When this - * method returns a new volatile index has been created. - * - * @throws IOException - * if an error occurs while writing the volatile index to disk. - */ - private void commitVolatileIndex() throws IOException - { - - // check if volatile index contains documents at all - if (volatileIndex.getNumDocuments() > 0) - { - - long time = System.currentTimeMillis(); - // create index - - //TODO this was replaced - // CreateIndex create = new CreateIndex(getTransactionId(), null); - // executeAndLog(create); - - PersistentIndex idx = getOrCreateIndex(null); - - // commit volatile index - - // TODO THIS IS REPLACED - //executeAndLog(new VolatileCommit(getTransactionId(), idx.getName())); - - idx.copyIndex(volatileIndex); - resetVolatileIndex(); - - // TODO this is replaced - // add new index - // AddIndex add = new AddIndex(getTransactionId(), create.getIndexName()); - // executeAndLog(add); - - if (!indexNames.contains(idx.getName())) - { - indexNames.addName(idx.getName()); - } - - // create new volatile index - resetVolatileIndex(); - - time = System.currentTimeMillis() - time; - log.debug("Committed in-memory index in " + time + "ms."); - } - } - - /** * Recursively creates an index starting with the NodeState * <code>node</code>. * @@ -1395,7 +889,7 @@ // TODO: this is replaced //executeAndLog(new AddNode(getTransactionId(), node.getIdentifier())); - volatileIndex.addDocuments(new Document[]{createDocument(node)}); + getChunk(node.getIdentifier()).addDocuments(new Document[]{createDocument(node)}); if (++count % 100 == 0) { @@ -1406,7 +900,8 @@ { checkIndexingQueue(true); } - checkVolatileCommit(); + // TODO commented line below: + //checkVolatileCommit(); List<NodeData> children = stateMgr.getChildNodesData(node); for (NodeData nodeData : children) { @@ -1428,28 +923,6 @@ } /** - * Attempts to delete all files recorded in {@link #deletable}. - */ - private void attemptDelete() - { - synchronized (deletable) - { - for (Iterator it = deletable.iterator(); it.hasNext();) - { - String indexName = (String)it.next(); - if (directoryManager.delete(indexName)) - { - it.remove(); - } - else - { - log.info("Unable to delete obsolete index: " + indexName); - } - } - } - } - - /** * Removes the deletable file if it exists. The file is not used anymore in * Jackrabbit versions >= 1.5. */ @@ -1477,52 +950,6 @@ } /** - * Checks the duration between the last commit to this index and the current - * time and flushes the index (if there are changes at all) if the duration - * (idle time) is more than {@link SearchIndex#getVolatileIdleTime()} - * seconds. - */ - private synchronized void checkFlush() - { - long idleTime = System.currentTimeMillis() - lastFlushTime; - long volatileTime = System.currentTimeMillis() - lastFileSystemFlushTime; - // do not flush if volatileIdleTime is zero or negative - if ((handler.getVolatileIdleTime() > 0 && idleTime > handler.getVolatileIdleTime() * 1000) - || (handler.getMaxVolatileTime() > 0 && volatileTime > handler.getMaxVolatileTime() * 1000)) - { - try - { - { - log.debug("Flushing index after being idle for " + idleTime + " ms."); - synchronized (updateMonitor) - { - //updateInProgress = true; - indexUpdateMonitor.setUpdateInProgress(true, true); - } - try - { - flush(); - } - finally - { - synchronized (updateMonitor) - { - //updateInProgress = false; - indexUpdateMonitor.setUpdateInProgress(false, true); - updateMonitor.notifyAll(); - releaseMultiReader(); - } - } - } - } - catch (IOException e) - { - log.error("Unable to commit volatile index", e); - } - } - } - - /** * Checks the indexing queue for finished text extrator jobs and updates the * index accordingly if there are any new ones. This method is synchronized * and should only be called by the timer task that periodically checks if @@ -1547,7 +974,7 @@ private void checkIndexingQueue(boolean transactionPresent) { Document[] docs = indexingQueue.getFinishedDocuments(); - Map finished = new HashMap(); + Map<String, Document> finished = new HashMap<String, Document>(); for (int i = 0; i < docs.length; i++) { String uuid = docs[i].get(FieldNames.UUID); @@ -1560,21 +987,21 @@ log.info("updating index with {} nodes from indexing queue.", new Long(finished.size())); // remove documents from the queue - for (Iterator it = finished.keySet().iterator(); it.hasNext();) + for (Iterator<String> it = finished.keySet().iterator(); it.hasNext();) { - indexingQueue.removeDocument(it.next().toString()); + indexingQueue.removeDocument(it.next()); } try { if (transactionPresent) { - for (Iterator it = finished.keySet().iterator(); it.hasNext();) + for (Iterator<String> it = finished.keySet().iterator(); it.hasNext();) { // TODO this was replaced //executeAndLog(new DeleteNode(getTransactionId(), (String)it.next())); - String uuidString = (String)it.next(); + String uuidString = it.next(); // check if indexing queue is still working on // this node from a previous update Document doc = indexingQueue.removeDocument(uuidString); @@ -1585,13 +1012,13 @@ Term idTerm = new Term(FieldNames.UUID, uuidString); // if the document cannot be deleted from the volatile index // delete it from one of the persistent indexes. - int num = volatileIndex.removeDocument(idTerm); + int num = getChunk(uuidString).removeDocument(idTerm); if (num == 0) { for (int i = indexes.size() - 1; i >= 0; i--) { // only look in registered indexes - PersistentIndex idx = (PersistentIndex)indexes.get(i); + PersistentIndex idx = indexes.get(i); if (indexNames.contains(idx.getName())) { num = idx.removeDocument(idTerm); @@ -1604,11 +1031,13 @@ } } - for (Iterator it = finished.values().iterator(); it.hasNext();) + for (Iterator<Document> it = finished.values().iterator(); it.hasNext();) { // TODO this was replaced //executeAndLog(new AddNode(getTransactionId(), (Document)it.next())); - volatileIndex.addDocuments(new Document[]{(Document)it.next()}); + Document doc = it.next(); + String uuid = doc.get(FieldNames.UUID); + getChunk(uuid).addDocuments(new Document[]{doc}); } } else @@ -1627,154 +1056,14 @@ // ------------------------< Actions // >--------------------------------------- - /** - * @see org.exoplatform.services.jcr.impl.core.query.IndexerIoModeListener#onChangeMode(org.exoplatform.services.jcr.impl.core.query.IndexerIoMode) - */ - public void onChangeMode(IndexerIoMode mode) - { - try - { - switch (mode) - { - case READ_ONLY : - setReadOny(); - break; - case READ_WRITE : - setReadWrite(); - break; - } - } - catch (IOException e) - { - log.error("An error occurs while changing of mode " + mode, e); - } - } - - /** - * Sets mode to READ_ONLY, discarding flush task - */ - protected void setReadOny() - { - // try to stop merger in safe way - flushTask.cancel(); - FLUSH_TIMER.purge(); - } - - /** - * Sets mode to READ_WRITE, initiating recovery process - * - * @throws IOException - */ - protected void setReadWrite() throws IOException - { - // Release all the current threads - synchronized (updateMonitor) - { - indexUpdateMonitor.setUpdateInProgress(false, true); - updateMonitor.notifyAll(); - releaseMultiReader(); - } - - // enqueue unused segments for deletion - enqueueUnusedSegments(); - attemptDelete(); - - // now that we are ready, start index merger - flush(); - - if (indexNames.size() > 0) - { - scheduleFlushTask(); - } - } - - /** - * Refresh list of indexes. Used to be called asynchronously when list changes. New, actual list is read from - * IndexInfos. - * @throws IOException - */ - public void refreshIndexList() throws IOException - { - // TODO: re-study synchronization here. - synchronized (updateMonitor) - { - // release reader if any - releaseMultiReader(); - // prepare added/removed sets - Set<String> newList = new HashSet<String>(indexNames.getNames()); - - // remove removed indexes - Iterator<PersistentIndex> iterator = indexes.iterator(); - while (iterator.hasNext()) - { - PersistentIndex index = iterator.next(); - String name = index.getName(); - // if current index not in new list, close it, cause it is deleted. - if (!newList.contains(name)) - { - index.close(); - iterator.remove(); - } - else - { - // remove from list, cause this segment of index still present and indexes list contains - // PersistentIndex instance related to this index.. - newList.remove(name); - } - } - // now newList contains ONLY new, added indexes, deleted indexes, are removed from list. - for (String name : newList) - { - // only open if it still exists - // it is possible that indexNames still contains a name for - // an index that has been deleted, but indexNames has not been - // written to disk. - if (!directoryManager.hasDirectory(name)) - { - log.debug("index does not exist anymore: " + name); - // move on to next index - continue; - } - PersistentIndex index = - new PersistentIndex(name, handler.getTextAnalyzer(), handler.getSimilarity(), cache, indexingQueue, - directoryManager); - index.setMaxFieldLength(handler.getMaxFieldLength()); - index.setUseCompoundFile(handler.getUseCompoundFile()); - index.setTermInfosIndexDivisor(handler.getTermInfosIndexDivisor()); - indexes.add(index); - } - } - } - - /** - * @see org.exoplatform.services.jcr.impl.core.query.lucene.IndexUpdateMonitorListener#onUpdateInProgressChange(boolean) - */ - public void onUpdateInProgressChange(boolean updateInProgress) - { - if (modeHandler.getMode() == IndexerIoMode.READ_ONLY) - { - if (!updateInProgress) - { - // wake up the sleeping threads - try - { - synchronized (updateMonitor) - { - updateMonitor.notifyAll(); - releaseMultiReader(); - } - } - catch (IOException e) - { - log.error("An erro occurs while trying to wake up the sleeping threads", e); - } - } - } - } - @Deprecated public boolean getRedoLogApplied() { return false; } + + private PersistentIndex getChunk(String uuid) + { + return indexes.get(0); + } } Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/api/core/query/lucene/IndexingAggregateTest.java =================================================================== --- jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/api/core/query/lucene/IndexingAggregateTest.java 2010-10-11 11:27:31 UTC (rev 3287) +++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/java/org/exoplatform/services/jcr/api/core/query/lucene/IndexingAggregateTest.java 2010-10-11 12:17:47 UTC (rev 3288) @@ -106,7 +106,7 @@ { SearchIndex index = (SearchIndex)getQueryHandler(); IndexingQueue queue = index.getIndex().getIndexingQueue(); - index.getIndex().flush(); + //index.getIndex().flush(); synchronized (index.getIndex()) { while (queue.getNumPendingDocuments() > 0)