[hibernate-commits] Hibernate SVN: r16967 - in search/trunk/src: main/java/org/hibernate/search/backend and 10 other directories.

Author: sannegrinovero Date: 2009-06-29 18:25:43 -0400 (Mon, 29 Jun 2009) New Revision: 16967 Added: search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/ search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/BatchBackend.java search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/DirectoryProviderWorkspace.java search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/LuceneBatchBackend.java search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/PerDirectoryWorkProcessor.java search/trunk/src/main/java/org/hibernate/search/batchindexing/BatchCoordinator.java search/trunk/src/test/java/org/hibernate/search/test/batchindexing/IndexingGeneratedCorpusTest.java search/trunk/src/test/java/org/hibernate/search/test/batchindexing/TitleAble.java search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/ search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/SentenceInventor.java search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/TextProductionTest.java search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/WordDictionary.java Removed: search/trunk/src/main/java/org/hibernate/search/batchindexing/IndexWritingJob.java Modified: search/trunk/src/main/java/org/hibernate/search/Environment.java search/trunk/src/main/java/org/hibernate/search/Indexer.java search/trunk/src/main/java/org/hibernate/search/backend/Workspace.java search/trunk/src/main/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/DpSelectionDelegate.java search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/DpSelectionVisitor.java search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/LuceneBackendQueueProcessor.java search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/PerDPQueueProcessor.java search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/QueueProcessors.java search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/AddWorkDelegate.java search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/DeleteWorkDelegate.java search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/LuceneWorkDelegate.java search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/OptimizeWorkDelegate.java search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/PurgeAllWorkDelegate.java search/trunk/src/main/java/org/hibernate/search/batchindexing/BatchIndexingWorkspace.java search/trunk/src/main/java/org/hibernate/search/batchindexing/EntityConsumerLuceneworkProducer.java search/trunk/src/main/java/org/hibernate/search/batchindexing/Executors.java search/trunk/src/main/java/org/hibernate/search/batchindexing/IdentifierConsumerEntityProducer.java search/trunk/src/main/java/org/hibernate/search/batchindexing/IdentifierProducer.java search/trunk/src/main/java/org/hibernate/search/batchindexing/ProducerConsumerQueue.java search/trunk/src/main/java/org/hibernate/search/engine/SearchFactoryImplementor.java search/trunk/src/main/java/org/hibernate/search/impl/IndexerImpl.java search/trunk/src/main/java/org/hibernate/search/impl/SearchFactoryImpl.java search/trunk/src/main/java/org/hibernate/search/impl/SimpleIndexingProgressMonitor.java search/trunk/src/test/java/org/hibernate/search/test/batchindexing/AncientBook.java search/trunk/src/test/java/org/hibernate/search/test/batchindexing/Book.java search/trunk/src/test/java/org/hibernate/search/test/batchindexing/Dvd.java search/trunk/src/test/java/org/hibernate/search/test/batchindexing/SearchIndexerTest.java Log: HSEARCH-218 [add indexAll( Class type ) to rebuild indexes from all data] Adding batch backend and coupling the batch API to the new backend. Modified: search/trunk/src/main/java/org/hibernate/search/Environment.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/Environment.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/Environment.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -64,12 +64,19 @@ public static final String READER_STRATEGY = READER_PREFIX + "strategy"; /** - * filter caching strategy class (must have a no-arg constructor and implements FilterCachingStrategy) + * filter caching strategy class (must have a no-arg constructor and implement FilterCachingStrategy) */ public static final String FILTER_CACHING_STRATEGY = "hibernate.search.filter.cache_strategy"; /** * number of docidresults cached in hard reference. */ - public static final String CACHE_DOCIDRESULTS_SIZE = "hibernate.search.filter.cache_docidresults.size"; + public static final String CACHE_DOCIDRESULTS_SIZE = "hibernate.search.filter.cache_docidresults.size"; + + /** + * batch backend implementation class (must have a no-arg constructor and implement BatchBackend) + * also prefix for configuration settings of the batch backend + */ + public static final String BATCH_BACKEND = "hibernate.search.batchbackend"; + } Modified: search/trunk/src/main/java/org/hibernate/search/Indexer.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/Indexer.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/Indexer.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -1,6 +1,6 @@ package org.hibernate.search; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.Future; import org.hibernate.CacheMode; @@ -29,7 +29,13 @@ */ Indexer documentBuilderThreads(int numberOfThreads); - //not supported yet + /** + * Sets the number of threads to be used to analyze the documents + * and write to the index. + * @param numberOfThreads + * @return + */ + //TODO not yet implemented //Indexer indexWriterThreads(int numberOfThreads); /** @@ -68,21 +74,28 @@ Indexer purgeAllAtStart(boolean purgeAll); /** + * Will stop indexing after having indexed this amount of objects. + * As a results the index will not be consistent + * with the database: use only for testing. + * @param maximum + * @return + */ + Indexer limitObjects(int maximum); + + /** * Starts the indexing process in background (asynchronous). * Can be called only once. + * @return a Future to control task canceling. get() will always return null, + * blocking until completion. */ - void start(); + Future<?> start(); /** * Starts the indexing process, and then block until it's finished. * Can be called only once. - * @param timeout the maximum time to wait - * @param unit the time unit of the <tt>timeout</tt> argument. - * @return <tt>true</tt> if the process finished and <tt>false</tt> - * if the waiting time elapsed before the process was finished. * @throws InterruptedException if the current thread is interrupted * while waiting. */ - boolean startAndWait( long timeout, TimeUnit unit ) throws InterruptedException; + void startAndWait() throws InterruptedException; } Modified: search/trunk/src/main/java/org/hibernate/search/backend/Workspace.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/Workspace.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/Workspace.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -11,7 +11,6 @@ import org.apache.lucene.index.IndexWriter; import org.slf4j.Logger; -import org.hibernate.annotations.common.AssertionFailure; import org.hibernate.search.SearchException; import org.hibernate.search.SearchFactory; import org.hibernate.search.engine.DocumentBuilderIndexedEntity; @@ -78,16 +77,19 @@ /** * If optimization has not been forced give a chance to configured OptimizerStrategy * to optimize the index. - * To enter the optimization phase you need to acquire the lock first. - * @throws AssertionFailure if the lock is not owned. */ public void optimizerPhase() { - assertOwnLock(); - // used getAndSet(0) because Workspace is going to be reused by next transaction. - synchronized (optimizerStrategy) { - optimizerStrategy.addTransaction( operations.getAndSet( 0L ) ); - optimizerStrategy.optimize( this ); + lock.lock(); + try { + // used getAndSet(0) because Workspace is going to be reused by next transaction. + synchronized ( optimizerStrategy ) { + optimizerStrategy.addTransaction( operations.getAndSet( 0L ) ); + optimizerStrategy.optimize( this ); + } } + finally { + lock.unlock(); + } } /** @@ -98,17 +100,22 @@ * @see SearchFactory#optimize(Class) */ public void optimize() { - // Needs to ensure the optimizerStrategy is accessed in threadsafe way - synchronized (optimizerStrategy) { - optimizerStrategy.optimizationForced(); + lock.lock(); + try { + //Needs to ensure the optimizerStrategy is accessed in threadsafe way + synchronized (optimizerStrategy) { + optimizerStrategy.optimizationForced(); + } } + finally { + lock.unlock(); + } } /** * Gets the IndexWriter, opening one if needed. * @param batchmode when true the indexWriter settings for batch mode will be applied. * Ignored if IndexWriter is open already. - * @throws AssertionFailure if the lock is not owned. * @throws SearchException on a IOException during index opening. * @return a new IndexWriter or one already open. */ @@ -130,8 +137,8 @@ /** * Commits changes to a previously opened IndexWriter. * - * @throws SearchException on IOException during Lucene close operation. - * @throws AssertionFailure if there is no IndexWriter to close. + * @throws SearchException on IOException during Lucene close operation, + * or if there is no IndexWriter to close. */ public synchronized void commitIndexWriter() { if ( writer != null ) { @@ -143,15 +150,11 @@ throw new SearchException( "Exception while commiting index changes", e ); } } - else { - throw new AssertionFailure( "No open IndexWriter to commit changes." ); - } } /** * Closes a previously opened IndexWriter. - * @throws SearchException on IOException during Lucene close operation. - * @throws AssertionFailure if there is no IndexWriter to close. + * @throws SearchException on IOException during Lucene close operation */ public synchronized void closeIndexWriter() { IndexWriter toClose = writer; @@ -165,9 +168,6 @@ throw new SearchException( "Exception while closing IndexWriter", e ); } } - else { - throw new AssertionFailure( "No open IndexWriter to close" ); - } } /** @@ -186,27 +186,5 @@ public Set<Class<?>> getEntitiesInDirectory() { return entitiesInDirectory; } - - /** - * Acquires a lock on the DirectoryProvider backing this Workspace; - * this is required to use optimizerPhase() - * @see #optimizerPhase() - */ - public void lock() { - lock.lock(); - } - /** - * Releases the lock obtained by calling lock(). The caller must own the lock. - * @see #lock() - */ - public void unlock() { - lock.unlock(); - } - - private void assertOwnLock() { - if ( ! lock.isHeldByCurrentThread() ) - throw new AssertionFailure( "Not owning DirectoryProvider Lock" ); - } - } Modified: search/trunk/src/main/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -89,7 +89,7 @@ } else { try { - Class processorFactoryClass = ReflectHelper.classForName( backend, BatchedQueueingProcessor.class ); + Class<?> processorFactoryClass = ReflectHelper.classForName( backend, BatchedQueueingProcessor.class ); backendQueueProcessorFactory = ( BackendQueueProcessorFactory ) processorFactoryClass.newInstance(); } catch ( ClassNotFoundException e ) { Added: search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/BatchBackend.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/BatchBackend.java (rev 0) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/BatchBackend.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -0,0 +1,56 @@ +package org.hibernate.search.backend.impl.batchlucene; + +import java.util.Properties; +import java.util.concurrent.TimeUnit; + +import org.hibernate.search.backend.LuceneWork; +import org.hibernate.search.batchindexing.IndexerProgressMonitor; +import org.hibernate.search.engine.SearchFactoryImplementor; + +/** + * Implementors of this interface are not drop-in replacements for the standard BackendQueueProcessorFactory + * but are meant to be used only during batch processing. + * The order of LuceneWork(s) processed is not guaranteed as the queue is consumed by several + * concurrent workers. + * + * @author Sanne Grinovero + */ +public interface BatchBackend { + + /** + * Used at startup, called once as first method. + * @param props all configuration properties + * @param searchFactory the client + */ + void initialize(Properties props, IndexerProgressMonitor monitor, SearchFactoryImplementor searchFactory); + + /** + * Enqueues one work to be processed asynchronously + * @param work + * @throws InterruptedException if the current thread is interrupted while + * waiting for the work queue to have enough space. + */ + void enqueueAsyncWork(LuceneWork work) throws InterruptedException; + + /** + * Does one work in sync + * @param work + * @throws InterruptedException + */ + void doWorkInSync(LuceneWork work); + + /** + * Waits until all work is done and terminates the executors. + * IndexWriter is not closed yet: work in sync can still be processed. + * @throws InterruptedException if the current thread is interrupted + * while waiting for the enqueued tasks to be finished. + */ + void stopAndFlush(long timeout, TimeUnit unit) throws InterruptedException; + + /** + * Used to shutdown and release resources. + * No other method should be used after this one. + */ + void close(); + +} Added: search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/DirectoryProviderWorkspace.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/DirectoryProviderWorkspace.java (rev 0) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/DirectoryProviderWorkspace.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -0,0 +1,124 @@ +package org.hibernate.search.backend.impl.batchlucene; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.hibernate.search.SearchException; +import org.hibernate.search.backend.LuceneWork; +import org.hibernate.search.backend.Workspace; +import org.hibernate.search.backend.impl.lucene.works.LuceneWorkDelegate; +import org.hibernate.search.backend.impl.lucene.works.LuceneWorkVisitor; +import org.hibernate.search.batchindexing.Executors; +import org.hibernate.search.batchindexing.IndexerProgressMonitor; +import org.hibernate.search.engine.SearchFactoryImplementor; +import org.hibernate.search.store.DirectoryProvider; +import org.hibernate.search.util.LoggerFactory; +import org.slf4j.Logger; + +/** + * Collects all resources needed to apply changes to one index. + * They are reused across the processing of all LuceneWork. + * + * !! Be careful to ensure the IndexWriter is eventually closed, + * or the index will stay locked. + * @see close(); + * + * @author Sanne Grinovero + */ +class DirectoryProviderWorkspace { + + private static final Logger log = LoggerFactory.make(); + + private final ExecutorService executor; + private final LuceneWorkVisitor visitor; + private final Workspace workspace; + private final IndexerProgressMonitor monitor; + + private final AtomicBoolean closed = new AtomicBoolean( false ); + + DirectoryProviderWorkspace(SearchFactoryImplementor searchFactoryImp, DirectoryProvider<?> dp, IndexerProgressMonitor monitor, int maxThreads) { + if ( maxThreads < 1 ) { + throw new IllegalArgumentException( "maxThreads needs to be at least 1" ); + } + this.monitor = monitor; + workspace = new Workspace( searchFactoryImp, dp ); + visitor = new LuceneWorkVisitor( workspace ); + executor = Executors.newFixedThreadPool( maxThreads, "indexwriter" ); + } + + /** + * Notify the indexwriting threads that they should quit at the end of the enqueued + * tasks. Waits for the end of the current queue, then commits changes. + * @throws InterruptedException + */ + public void stopAndFlush(long timeout, TimeUnit unit) throws InterruptedException { + checkIsNotClosed(); + executor.shutdown(); //it becomes illegal to add more work + executor.awaitTermination( timeout, unit ); + workspace.commitIndexWriter(); //commits changes if any + //does not yet close the IndexWriter ! + } + + /** + * Used to do some tasks at the beginning and/or at the end of the main batch + * operations. This work is not done async. + * @param work + */ + public void doWorkInSync(LuceneWork work) { + checkIsNotClosed(); + LuceneWorkDelegate delegate = work.getWorkDelegate( visitor ); + delegate.performWork( work, workspace.getIndexWriter( true ) ); + delegate.logWorkDone( work , monitor ); + //if the IndexWriter was opened, it's not closed now. + } + + public void enqueueAsyncWork(LuceneWork work) { + //no need to check if we are closed here, better check inside the async method + executor.execute( new AsyncIndexRunnable( work ) ); + } + + /** + * Makes sure the executor is closed and closes the IndexWriter. + */ + public void close() { + if ( closed.compareAndSet( false, true ) ) { + try { + if ( ! executor.isShutdown() ) { + log.error( "Terminating batch work! Index might end up in inconsistent state." ); + executor.shutdownNow(); + } + } + finally { + workspace.closeIndexWriter(); + } + } + else { + checkIsNotClosed(); //will throw an appropriate exception + } + } + + /** + * Verifies this is not closed yet, or throws an exception. + */ + private void checkIsNotClosed() { + if ( closed.get() ) { + throw new SearchException( "Batch DirectoryProviderWorkspace is closed already" ); + } + } + + private class AsyncIndexRunnable implements Runnable { + + private final LuceneWork work; + + AsyncIndexRunnable(LuceneWork work) { + this.work = work; + } + + public void run() { + doWorkInSync( work ); + } + + } + +} Property changes on: search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/DirectoryProviderWorkspace.java ___________________________________________________________________ Name: svn:executable + * Added: search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/LuceneBatchBackend.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/LuceneBatchBackend.java (rev 0) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/batchlucene/LuceneBatchBackend.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -0,0 +1,120 @@ +package org.hibernate.search.backend.impl.batchlucene; + +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.TimeUnit; + +import org.hibernate.search.Environment; +import org.hibernate.search.SearchException; +import org.hibernate.search.backend.LuceneWork; +import org.hibernate.search.backend.configuration.ConfigurationParseHelper; +import org.hibernate.search.backend.impl.lucene.DpSelectionVisitor; +import org.hibernate.search.backend.impl.lucene.PerDirectoryWorkProcessor; +import org.hibernate.search.batchindexing.IndexerProgressMonitor; +import org.hibernate.search.engine.DocumentBuilderIndexedEntity; +import org.hibernate.search.engine.SearchFactoryImplementor; +import org.hibernate.search.store.DirectoryProvider; +import org.hibernate.search.store.IndexShardingStrategy; + +/** + * First EXPERIMENTAL BatchBackend; this is not meant to be used as a regular + * backend, only to apply batch changes to the index. Several threads + * are used to make changes to each index, so order of Work processing is not guaranteed. + * + * @author Sanne Grinovero + */ +public class LuceneBatchBackend implements BatchBackend { + + public static final String CONCURRENT_WRITERS = Environment.BATCH_BACKEND + ".concurrent_writers"; + + private static final DpSelectionVisitor providerSelectionVisitor = new DpSelectionVisitor(); + + private SearchFactoryImplementor searchFactoryImplementor; + private final Map<DirectoryProvider<?>,DirectoryProviderWorkspace> resourcesMap = new HashMap<DirectoryProvider<?>,DirectoryProviderWorkspace>(); + private final PerDirectoryWorkProcessor asyncWorker = new AsyncBatchPerDirectoryWorkProcessor(); + private final PerDirectoryWorkProcessor syncWorker = new SyncBatchPerDirectoryWorkProcessor(); + + public void initialize(Properties cfg, IndexerProgressMonitor monitor, SearchFactoryImplementor searchFactoryImplementor) { + this.searchFactoryImplementor = searchFactoryImplementor; + int maxThreadsPerIndex = ConfigurationParseHelper.getIntValue( cfg, "concurrent_writers", 2 ); + if ( maxThreadsPerIndex < 1 ) { + throw new SearchException( "concurrent_writers for batch backend must be at least 1." ); + } + for ( DirectoryProvider<?> dp : searchFactoryImplementor.getDirectoryProviders() ) { + DirectoryProviderWorkspace resources = new DirectoryProviderWorkspace( searchFactoryImplementor, dp, monitor, maxThreadsPerIndex ); + resourcesMap.put( dp, resources ); + } + } + + public void enqueueAsyncWork(LuceneWork work) throws InterruptedException { + sendWorkToShards( work, asyncWorker ); + } + + public void doWorkInSync(LuceneWork work) { + try { + sendWorkToShards( work, syncWorker ); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + //doesn't happen, see SyncBatchPerDirectoryWorkProcessor below: is missing the throws. + throw new SearchException( "AssertionFailure" ); + } + } + + /** + * Stops the background threads and flushes changes; + * Please note the timeout is applied to each index in + * sequence, so it might take as much time as timeout*directoryproviders + */ + public void stopAndFlush(long timeout, TimeUnit unit) throws InterruptedException { + for ( DirectoryProviderWorkspace res : resourcesMap.values() ) { + res.stopAndFlush( timeout, unit ); + } + } + + public void close() { + Throwable error = null; + for ( DirectoryProviderWorkspace res : resourcesMap.values() ) { + try { + res.close(); + } + catch (Throwable t) { + //make sure to try closing all IndexWriters + error = t; + } + } + if ( error != null ) { + throw new SearchException( "Error while closing batch indexer", error ); + } + } + + private void sendWorkToShards(LuceneWork work, PerDirectoryWorkProcessor worker) throws InterruptedException { + final Class<?> entityType = work.getEntityClass(); + DocumentBuilderIndexedEntity<?> documentBuilder = searchFactoryImplementor.getDocumentBuilderIndexedEntity( entityType ); + IndexShardingStrategy shardingStrategy = documentBuilder.getDirectoryProviderSelectionStrategy(); + work.getWorkDelegate( providerSelectionVisitor ).addAsPayLoadsToQueue( work, shardingStrategy, worker ); + } + + /** + * Implements a PerDirectoryWorkProcessor to enqueue work Asynchronously. + */ + private class AsyncBatchPerDirectoryWorkProcessor implements PerDirectoryWorkProcessor { + + public void addWorkToDpProcessor(DirectoryProvider<?> dp, LuceneWork work) throws InterruptedException { + resourcesMap.get( dp ).enqueueAsyncWork( work ); + } + + } + + /** + * Implements a PerDirectoryWorkProcessor to enqueue work Synchronously. + */ + private class SyncBatchPerDirectoryWorkProcessor implements PerDirectoryWorkProcessor { + + public void addWorkToDpProcessor(DirectoryProvider<?> dp, LuceneWork work) { + resourcesMap.get( dp ).doWorkInSync( work ); + } + + } + +} Modified: search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/DpSelectionDelegate.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/DpSelectionDelegate.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/DpSelectionDelegate.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -6,15 +6,16 @@ /** * @author Sanne Grinovero */ -interface DpSelectionDelegate { +public interface DpSelectionDelegate { /** * The LuceneWork must be applied to different indexes. * @param work the work to split. * @param queues the target queue to add work to. * @param shardingStrategy the Sharding strategy is usually needed to identify affected Directories. + * @throws InterruptedException */ void addAsPayLoadsToQueue(LuceneWork work, - IndexShardingStrategy shardingStrategy, QueueProcessors queues); + IndexShardingStrategy shardingStrategy, PerDirectoryWorkProcessor queues) throws InterruptedException; } Modified: search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/DpSelectionVisitor.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/DpSelectionVisitor.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/DpSelectionVisitor.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -17,7 +17,7 @@ * * @author Sanne Grinovero */ -class DpSelectionVisitor implements WorkVisitor<DpSelectionDelegate> { +public class DpSelectionVisitor implements WorkVisitor<DpSelectionDelegate> { private final AddSelectionDelegate addDelegate = new AddSelectionDelegate(); private final DeleteSelectionDelegate deleteDelegate = new DeleteSelectionDelegate(); @@ -43,8 +43,8 @@ private static class AddSelectionDelegate implements DpSelectionDelegate { public void addAsPayLoadsToQueue(LuceneWork work, - IndexShardingStrategy shardingStrategy, QueueProcessors queues) { - DirectoryProvider provider = shardingStrategy.getDirectoryProviderForAddition( + IndexShardingStrategy shardingStrategy, PerDirectoryWorkProcessor queues) throws InterruptedException { + DirectoryProvider<?> provider = shardingStrategy.getDirectoryProviderForAddition( work.getEntityClass(), work.getId(), work.getIdInString(), @@ -58,13 +58,13 @@ private static class DeleteSelectionDelegate implements DpSelectionDelegate { public void addAsPayLoadsToQueue(LuceneWork work, - IndexShardingStrategy shardingStrategy, QueueProcessors queues) { - DirectoryProvider[] providers = shardingStrategy.getDirectoryProvidersForDeletion( + IndexShardingStrategy shardingStrategy, PerDirectoryWorkProcessor queues) throws InterruptedException { + DirectoryProvider<?>[] providers = shardingStrategy.getDirectoryProvidersForDeletion( work.getEntityClass(), work.getId(), work.getIdInString() ); - for (DirectoryProvider provider : providers) { + for (DirectoryProvider<?> provider : providers) { queues.addWorkToDpProcessor( provider, work ); } } @@ -74,9 +74,9 @@ private static class OptimizeSelectionDelegate implements DpSelectionDelegate { public void addAsPayLoadsToQueue(LuceneWork work, - IndexShardingStrategy shardingStrategy, QueueProcessors queues) { - DirectoryProvider[] providers = shardingStrategy.getDirectoryProvidersForAllShards(); - for (DirectoryProvider provider : providers) { + IndexShardingStrategy shardingStrategy, PerDirectoryWorkProcessor queues) throws InterruptedException { + DirectoryProvider<?>[] providers = shardingStrategy.getDirectoryProvidersForAllShards(); + for (DirectoryProvider<?> provider : providers) { queues.addWorkToDpProcessor( provider, work ); } } @@ -86,13 +86,13 @@ private static class PurgeAllSelectionDelegate implements DpSelectionDelegate { public void addAsPayLoadsToQueue(LuceneWork work, - IndexShardingStrategy shardingStrategy, QueueProcessors queues) { - DirectoryProvider[] providers = shardingStrategy.getDirectoryProvidersForDeletion( + IndexShardingStrategy shardingStrategy, PerDirectoryWorkProcessor queues) throws InterruptedException { + DirectoryProvider<?>[] providers = shardingStrategy.getDirectoryProvidersForDeletion( work.getEntityClass(), work.getId(), work.getIdInString() ); - for (DirectoryProvider provider : providers) { + for (DirectoryProvider<?> provider : providers) { queues.addWorkToDpProcessor( provider, work ); } } Modified: search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/LuceneBackendQueueProcessor.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/LuceneBackendQueueProcessor.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/LuceneBackendQueueProcessor.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -43,13 +43,13 @@ public void run() { QueueProcessors processors = new QueueProcessors( resourcesMap ); // divide the queue in tasks, adding to QueueProcessors by affected Directory. - for ( LuceneWork work : queue ) { - final Class<?> entityType = work.getEntityClass(); - DocumentBuilderIndexedEntity<?> documentBuilder = searchFactoryImplementor.getDocumentBuilderIndexedEntity( entityType ); - IndexShardingStrategy shardingStrategy = documentBuilder.getDirectoryProviderSelectionStrategy(); - work.getWorkDelegate( providerSelectionVisitor ).addAsPayLoadsToQueue( work, shardingStrategy, processors ); - } try { + for ( LuceneWork work : queue ) { + final Class<?> entityType = work.getEntityClass(); + DocumentBuilderIndexedEntity<?> documentBuilder = searchFactoryImplementor.getDocumentBuilderIndexedEntity( entityType ); + IndexShardingStrategy shardingStrategy = documentBuilder.getDirectoryProviderSelectionStrategy(); + work.getWorkDelegate( providerSelectionVisitor ).addAsPayLoadsToQueue( work, shardingStrategy, processors ); + } //this Runnable splits tasks in more runnables and then runs them: processors.runAll( sync ); } catch (InterruptedException e) { Modified: search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/PerDPQueueProcessor.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/PerDPQueueProcessor.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/PerDPQueueProcessor.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -76,21 +76,12 @@ } private void performOptimizations() { - log.trace( "Locking Workspace (or waiting to...)" ); - workspace.lock(); - try { - log.trace( "Workspace lock aquired." ); - //TODO next line is assuming the OptimizerStrategy will need an IndexWriter; - // would be nicer to have the strategy put an OptimizeWork on the queue, - // or just return "yes please" (true) to some method? - //FIXME will not have a chance to trigger when no "add" activity is done. - // this is correct until we enable modification counts for deletions too. - workspace.optimizerPhase(); - } - finally { - workspace.unlock(); - log.trace( "Unlocked Workspace" ); - } + //TODO next line is assuming the OptimizerStrategy will need an IndexWriter; + // would be nicer to have the strategy put an OptimizeWork on the queue, + // or just return "yes please" (true) to some method? + //FIXME will not have a chance to trigger when no "add" activity is done. + // this is correct until we enable modification counts for deletions too. + workspace.optimizerPhase(); } /** Added: search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/PerDirectoryWorkProcessor.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/PerDirectoryWorkProcessor.java (rev 0) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/PerDirectoryWorkProcessor.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -0,0 +1,18 @@ +package org.hibernate.search.backend.impl.lucene; + +import org.hibernate.search.backend.LuceneWork; +import org.hibernate.search.store.DirectoryProvider; + +/** + * Interface to implement visitor pattern in combination + * with DpSelectionVisitor and DpSelectionDelegate to + * send LuceneWork to the appropriate queues, as defined + * by an IndexShardingStrategy. + * + * @author Sanne Grinovero + */ +public interface PerDirectoryWorkProcessor { + + public void addWorkToDpProcessor(DirectoryProvider<?> dp, LuceneWork work) throws InterruptedException; + +} Modified: search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/QueueProcessors.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/QueueProcessors.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/QueueProcessors.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -18,7 +18,7 @@ * them concurrently. * @author Sanne Grinovero */ -class QueueProcessors { +class QueueProcessors implements PerDirectoryWorkProcessor { private final Map<DirectoryProvider, PerDPResources> resourcesMap; private final Map<DirectoryProvider, PerDPQueueProcessor> dpProcessors = new HashMap<DirectoryProvider, PerDPQueueProcessor>(); @@ -27,7 +27,7 @@ this.resourcesMap = resourcesMap; } - void addWorkToDpProcessor(DirectoryProvider dp, LuceneWork work) { + public void addWorkToDpProcessor(DirectoryProvider dp, LuceneWork work) { if ( ! dpProcessors.containsKey( dp ) ) { dpProcessors.put( dp, new PerDPQueueProcessor( resourcesMap.get( dp ) ) ); } Modified: search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/AddWorkDelegate.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/AddWorkDelegate.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/AddWorkDelegate.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -12,6 +12,7 @@ import org.hibernate.search.backend.AddLuceneWork; import org.hibernate.search.backend.LuceneWork; import org.hibernate.search.backend.Workspace; +import org.hibernate.search.batchindexing.IndexerProgressMonitor; import org.hibernate.search.engine.DocumentBuilderIndexedEntity; import org.hibernate.search.util.LoggerFactory; import org.hibernate.search.util.ScopedAnalyzer; @@ -95,4 +96,8 @@ return analyzerClone; } + public void logWorkDone(LuceneWork work, IndexerProgressMonitor monitor) { + monitor.documentsAdded( 1 ); + } + } Modified: search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/DeleteWorkDelegate.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/DeleteWorkDelegate.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/DeleteWorkDelegate.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -12,6 +12,7 @@ import org.hibernate.search.SearchException; import org.hibernate.search.backend.LuceneWork; import org.hibernate.search.backend.Workspace; +import org.hibernate.search.batchindexing.IndexerProgressMonitor; import org.hibernate.search.engine.DocumentBuilderIndexedEntity; import org.hibernate.search.engine.DocumentBuilder; import org.hibernate.search.util.LoggerFactory; @@ -59,4 +60,8 @@ } } + public void logWorkDone(LuceneWork work, IndexerProgressMonitor monitor) { + // TODO Auto-generated method stub + } + } Modified: search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/LuceneWorkDelegate.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/LuceneWorkDelegate.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/LuceneWorkDelegate.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -2,6 +2,7 @@ import org.apache.lucene.index.IndexWriter; import org.hibernate.search.backend.LuceneWork; +import org.hibernate.search.batchindexing.IndexerProgressMonitor; /** * @author Sanne Grinovero @@ -16,4 +17,12 @@ */ void performWork(LuceneWork work, IndexWriter writer); + /** + * Used for stats and performance counters, use the monitor + * to keep track of activity done on the index. + * @param work the work which was done. + * @param monitor the monitor tracking activity. + */ + void logWorkDone(LuceneWork work, IndexerProgressMonitor monitor); + } Modified: search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/OptimizeWorkDelegate.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/OptimizeWorkDelegate.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/OptimizeWorkDelegate.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -8,6 +8,7 @@ import org.hibernate.search.SearchException; import org.hibernate.search.backend.LuceneWork; import org.hibernate.search.backend.Workspace; +import org.hibernate.search.batchindexing.IndexerProgressMonitor; import org.hibernate.search.util.LoggerFactory; /** @@ -42,4 +43,9 @@ } } + public void logWorkDone(LuceneWork work, IndexerProgressMonitor monitor) { + // TODO Auto-generated method stub + + } + } Modified: search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/PurgeAllWorkDelegate.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/PurgeAllWorkDelegate.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/backend/impl/lucene/works/PurgeAllWorkDelegate.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -6,6 +6,7 @@ import org.hibernate.search.SearchException; import org.hibernate.search.backend.LuceneWork; +import org.hibernate.search.batchindexing.IndexerProgressMonitor; import org.hibernate.search.engine.DocumentBuilder; import org.hibernate.search.util.LoggerFactory; @@ -37,4 +38,9 @@ } } + public void logWorkDone(LuceneWork work, IndexerProgressMonitor monitor) { + // TODO Auto-generated method stub + + } + } Added: search/trunk/src/main/java/org/hibernate/search/batchindexing/BatchCoordinator.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/batchindexing/BatchCoordinator.java (rev 0) +++ search/trunk/src/main/java/org/hibernate/search/batchindexing/BatchCoordinator.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -0,0 +1,125 @@ +package org.hibernate.search.batchindexing; + +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; + +import org.hibernate.CacheMode; +import org.hibernate.SessionFactory; +import org.hibernate.search.backend.OptimizeLuceneWork; +import org.hibernate.search.backend.PurgeAllLuceneWork; +import org.hibernate.search.backend.impl.batchlucene.BatchBackend; +import org.hibernate.search.engine.SearchFactoryImplementor; +import org.hibernate.search.util.LoggerFactory; +import org.slf4j.Logger; + +/** + * Makes sure that several different BatchIndexingWorkspace(s) + * can be started concurrently, sharing the same batch-backend + * and IndexWriters. + * + * @author Sanne Grinovero + */ +public class BatchCoordinator implements Runnable { + + private static final Logger log = LoggerFactory.make(); + + private final Class<?>[] rootEntities; + private final SearchFactoryImplementor searchFactoryImplementor; + private final SessionFactory sessionFactory; + private final int objectLoadingThreads; + private final int collectionLoadingThreads; + private final CacheMode cacheMode; + private final int objectLoadingBatchSize; + private final boolean optimizeAtEnd; + private final boolean purgeAtStart; + private final boolean optimizeAfterPurge; + private final CountDownLatch endAllSignal; + private final IndexerProgressMonitor monitor; + private final int objectsLimit; + + private BatchBackend backend; + + public BatchCoordinator(Set<Class<?>> rootEntities, + SearchFactoryImplementor searchFactoryImplementor, + SessionFactory sessionFactory, int objectLoadingThreads, + int collectionLoadingThreads, CacheMode cacheMode, + int objectLoadingBatchSize, int objectsLimit, + boolean optimizeAtEnd, + boolean purgeAtStart, boolean optimizeAfterPurge, + IndexerProgressMonitor monitor) { + this.rootEntities = rootEntities.toArray( new Class<?>[ rootEntities.size() ] ); + this.searchFactoryImplementor = searchFactoryImplementor; + this.sessionFactory = sessionFactory; + this.objectLoadingThreads = objectLoadingThreads; + this.collectionLoadingThreads = collectionLoadingThreads; + this.cacheMode = cacheMode; + this.objectLoadingBatchSize = objectLoadingBatchSize; + this.optimizeAtEnd = optimizeAtEnd; + this.purgeAtStart = purgeAtStart; + this.optimizeAfterPurge = optimizeAfterPurge; + this.monitor = monitor; + this.objectsLimit = objectsLimit; + this.endAllSignal = new CountDownLatch( rootEntities.size() ); + } + + public void run() { + backend = searchFactoryImplementor.makeBatchBackend( monitor ); + try { + beforeBatch(); // purgeAll and pre-optimize activities + doBatchWork(); + backend.stopAndFlush( 60L*60*24, TimeUnit.SECONDS ); //1 day : enough to flush to indexes? +// backend.stopAndFlush( 10, TimeUnit.SECONDS ); + afterBatch(); + } catch (InterruptedException e) { + log.error( "Batch indexing was interrupted" ); + Thread.currentThread().interrupt(); + } + finally { + backend.close(); + } + } + + private void doBatchWork() throws InterruptedException { + ExecutorService executor = Executors.newFixedThreadPool( rootEntities.length, "BatchIndexingWorkspace" ); + for ( Class<?> type : rootEntities ) { + executor.execute( new BatchIndexingWorkspace( + searchFactoryImplementor, sessionFactory, type, + objectLoadingThreads, collectionLoadingThreads, + cacheMode, objectLoadingBatchSize, + endAllSignal, monitor, backend, objectsLimit ) ); + } + executor.shutdown(); + endAllSignal.await(); //waits for the executor to finish + } + + private void afterBatch() { + if ( this.optimizeAtEnd ) { + Set<Class<?>> targetedClasses = searchFactoryImplementor.getIndexedTypesPolymorphic( rootEntities ); + optimize( targetedClasses ); + } + } + + private void beforeBatch() { + if ( this.purgeAtStart ) { + //purgeAll for affected entities + Set<Class<?>> targetedClasses = searchFactoryImplementor.getIndexedTypesPolymorphic( rootEntities ); + for ( Class<?> clazz : targetedClasses ) { + //needs do be in-sync work to make sure we wait for the end of it. + backend.doWorkInSync( new PurgeAllLuceneWork( clazz ) ); + } + if ( this.optimizeAfterPurge ) { + optimize( targetedClasses ); + } + } + } + + private void optimize(Set<Class<?>> targetedClasses) { + for ( Class<?> clazz : targetedClasses ) { + //TODO the backend should remove duplicate optimize work to the same DP (as entities might share indexes) + backend.doWorkInSync( new OptimizeLuceneWork( clazz ) ); + } + } + +} Modified: search/trunk/src/main/java/org/hibernate/search/batchindexing/BatchIndexingWorkspace.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/batchindexing/BatchIndexingWorkspace.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/batchindexing/BatchIndexingWorkspace.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -1,19 +1,14 @@ package org.hibernate.search.batchindexing; -import java.util.ArrayList; +import java.io.Serializable; import java.util.List; -import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.atomic.AtomicBoolean; import org.hibernate.CacheMode; import org.hibernate.SessionFactory; import org.hibernate.search.SearchException; -import org.hibernate.search.backend.BackendQueueProcessorFactory; -import org.hibernate.search.backend.LuceneWork; -import org.hibernate.search.backend.OptimizeLuceneWork; -import org.hibernate.search.backend.PurgeAllLuceneWork; +import org.hibernate.search.backend.impl.batchlucene.BatchBackend; import org.hibernate.search.engine.SearchFactoryImplementor; import org.hibernate.search.util.LoggerFactory; import org.slf4j.Logger; @@ -33,21 +28,17 @@ //following order shows the 4 stages of an entity flowing to the index: private final ThreadPoolExecutor execIdentifiersLoader; - private final ProducerConsumerQueue fromIdentifierListToEntities; + private final ProducerConsumerQueue<List<Serializable>> fromIdentifierListToEntities; private final ThreadPoolExecutor execFirstLoader; - private final ProducerConsumerQueue fromEntityToAddwork; + private final ProducerConsumerQueue<Object> fromEntityToAddwork; private final ThreadPoolExecutor execDocBuilding; - private final ProducerConsumerQueue fromAddworkToIndex; - private final ThreadPoolExecutor execWriteIndex; private final int objectLoadingThreadNum; private final int luceneworkerBuildingThreadNum; - private final int indexWritingThreadNum; private final Class<?> indexedType; // status control - private final AtomicBoolean started = new AtomicBoolean( false ); - private final CountDownLatch endWritersSignal; //released when we stop adding Documents to Index + private final CountDownLatch producerEndSignal; //released when we stop adding Documents to Index private final CountDownLatch endAllSignal; //released when we release all locks and IndexWriter // progress monitor @@ -57,16 +48,17 @@ private final CacheMode cacheMode; private final int objectLoadingBatchSize; - private final boolean purgeAtStart; - private final boolean optimizeAfterPurge; - private final boolean optimizeAtEnd; + private final BatchBackend backend; + + private final int objectsLimit; public BatchIndexingWorkspace(SearchFactoryImplementor searchFactoryImplementor, SessionFactory sessionFactory, Class<?> entityType, - int objectLoadingThreads, int collectionLoadingThreads, int writerThreads, + int objectLoadingThreads, int collectionLoadingThreads, CacheMode cacheMode, int objectLoadingBatchSize, - boolean optimizeAtEnd, boolean purgeAtStart, boolean optimizeAfterPurge, CountDownLatch endAllSignal, - IndexerProgressMonitor monitor) { + CountDownLatch endAllSignal, + IndexerProgressMonitor monitor, BatchBackend backend, + int objectsLimit) { this.indexedType = entityType; this.searchFactory = searchFactoryImplementor; @@ -75,116 +67,69 @@ //thread pool sizing: this.objectLoadingThreadNum = objectLoadingThreads; this.luceneworkerBuildingThreadNum = collectionLoadingThreads;//collections are loaded as needed by building the document -// this.indexWritingThreadNum = writerThreads; //FIXME enable this line and remove the next line after solving HSEARCH-367 - this.indexWritingThreadNum = 1; //loading options: this.cacheMode = cacheMode; this.objectLoadingBatchSize = objectLoadingBatchSize; + this.backend = backend; //executors: (quite expensive constructor) //execIdentifiersLoader has size 1 and is not configurable: ensures the list is consistent as produced by one transaction this.execIdentifiersLoader = Executors.newFixedThreadPool( 1, "identifierloader" ); this.execFirstLoader = Executors.newFixedThreadPool( objectLoadingThreadNum, "entityloader" ); this.execDocBuilding = Executors.newFixedThreadPool( luceneworkerBuildingThreadNum, "collectionsloader" ); - this.execWriteIndex = Executors.newFixedThreadPool( indexWritingThreadNum, "analyzers" ); //pipelining queues: - this.fromIdentifierListToEntities = new ProducerConsumerQueue( 1 ); - this.fromEntityToAddwork = new ProducerConsumerQueue( objectLoadingThreadNum ); - this.fromAddworkToIndex = new ProducerConsumerQueue( luceneworkerBuildingThreadNum ); + this.fromIdentifierListToEntities = new ProducerConsumerQueue<List<Serializable>>( 1 ); + this.fromEntityToAddwork = new ProducerConsumerQueue<Object>( objectLoadingThreadNum ); //end signal shared with other instances: this.endAllSignal = endAllSignal; - this.endWritersSignal = new CountDownLatch( indexWritingThreadNum ); + this.producerEndSignal = new CountDownLatch( luceneworkerBuildingThreadNum ); - //behaviour options: - this.optimizeAtEnd = optimizeAtEnd; - this.optimizeAfterPurge = optimizeAfterPurge; - this.purgeAtStart = purgeAtStart; - this.monitor = monitor; + this.objectsLimit = objectsLimit; } public void run() { - if ( ! started.compareAndSet( false, true ) ) - throw new IllegalStateException( "BatchIndexingWorkspace can be started only once." ); - boolean interrupted = false; try { - //TODO switch to batch mode in backend - if ( purgeAtStart ) { - purgeAll(); - if ( optimizeAfterPurge ) { - optimize(); - } - } - BackendQueueProcessorFactory backendQueueProcessorFactory = searchFactory.getBackendQueueProcessorFactory(); //first start the consumers, then the producers (reverse order): - for ( int i=0; i < indexWritingThreadNum; i++ ) { - //from LuceneWork to IndexWriters: - execWriteIndex.execute( new IndexWritingJob( - fromAddworkToIndex, endWritersSignal, monitor, backendQueueProcessorFactory ) ); - } for ( int i=0; i < luceneworkerBuildingThreadNum; i++ ) { - //from entity to LuceneWork: + //from entity to LuceneWork: execDocBuilding.execute( new EntityConsumerLuceneworkProducer( - fromEntityToAddwork, fromAddworkToIndex, monitor, - sessionFactory, searchFactory, cacheMode) ); + fromEntityToAddwork, monitor, + sessionFactory, producerEndSignal, searchFactory, + cacheMode, backend) ); } for ( int i=0; i < objectLoadingThreadNum; i++ ) { - //from primary key to loaded entity: + //from primary key to loaded entity: execFirstLoader.execute( new IdentifierConsumerEntityProducer( fromIdentifierListToEntities, fromEntityToAddwork, monitor, sessionFactory, cacheMode, indexedType) ); } + //from class definition to all primary keys: + execIdentifiersLoader.execute( new IdentifierProducer( + fromIdentifierListToEntities, sessionFactory, + objectLoadingBatchSize, indexedType, monitor, + objectsLimit ) ); - execIdentifiersLoader.execute( new IdentifierProducer(fromIdentifierListToEntities, sessionFactory, objectLoadingBatchSize, indexedType, monitor) ); //shutdown all executors: execIdentifiersLoader.shutdown(); execFirstLoader.shutdown(); execDocBuilding.shutdown(); - execWriteIndex.shutdown(); - log.debug( "helper executors are shutting down" ); try { - endWritersSignal.await(); //await all indexing is done. + producerEndSignal.await(); //await for all work being sent to the backend + log.debug( "All work for type {} has been produced", indexedType.getName() ); } catch (InterruptedException e) { - interrupted = true; - throw new SearchException( "Interrupted on batch Indexing; index will be left in unknown state!", e); + //restore interruption signal: + Thread.currentThread().interrupt(); + throw new SearchException( "Interrupted on batch Indexing; index will be left in unknown state!", e ); } - log.debug( "index writing finished" ); - if ( optimizeAtEnd ) { - log.debug( "starting optimization" ); - optimize(); - } } finally { endAllSignal.countDown(); - if ( interrupted ) { - //restore interruption signal: - Thread.currentThread().interrupt(); - } } } - private void optimize() { - Set<Class<?>> targetedClasses = searchFactory.getIndexedTypesPolymorphic( new Class[] {indexedType} ); - List<LuceneWork> queue = new ArrayList<LuceneWork>( targetedClasses.size() ); - for ( Class<?> clazz : targetedClasses ) { - queue.add( new OptimizeLuceneWork( clazz ) ); - } - //TODO use the batch backend - searchFactory.getBackendQueueProcessorFactory().getProcessor( queue ).run(); - } - - private void purgeAll() { - Set<Class<?>> targetedClasses = searchFactory.getIndexedTypesPolymorphic( new Class[] {indexedType} ); - List<LuceneWork> queue = new ArrayList<LuceneWork>( targetedClasses.size() ); - for ( Class<?> clazz : targetedClasses ) { - queue.add( new PurgeAllLuceneWork( clazz ) ); - } - //TODO use the batch backend - searchFactory.getBackendQueueProcessorFactory().getProcessor( queue ).run(); - } - } Modified: search/trunk/src/main/java/org/hibernate/search/batchindexing/EntityConsumerLuceneworkProducer.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/batchindexing/EntityConsumerLuceneworkProducer.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/batchindexing/EntityConsumerLuceneworkProducer.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -2,6 +2,7 @@ import java.io.Serializable; import java.util.Map; +import java.util.concurrent.CountDownLatch; import org.hibernate.CacheMode; import org.hibernate.FlushMode; @@ -11,6 +12,7 @@ import org.hibernate.SessionFactory; import org.hibernate.Transaction; import org.hibernate.search.backend.AddLuceneWork; +import org.hibernate.search.backend.impl.batchlucene.BatchBackend; import org.hibernate.search.bridge.TwoWayFieldBridge; import org.hibernate.search.engine.DocumentBuilderIndexedEntity; import org.hibernate.search.engine.SearchFactoryImplementor; @@ -28,26 +30,30 @@ private static final Logger log = LoggerFactory.make(); - private final ProducerConsumerQueue source; - private final ProducerConsumerQueue destination; + private final ProducerConsumerQueue<Object> source; private final SessionFactory sessionFactory; private final Map<Class<?>, DocumentBuilderIndexedEntity<?>> documentBuilders; private final IndexerProgressMonitor monitor; private static final int CLEAR_PERIOD = 50; private final CacheMode cacheMode; + + private final CountDownLatch producerEndSignal; + + private final BatchBackend backend; public EntityConsumerLuceneworkProducer( - ProducerConsumerQueue entitySource, - ProducerConsumerQueue fromAddworkToIndex, + ProducerConsumerQueue<Object> entitySource, IndexerProgressMonitor monitor, SessionFactory sessionFactory, - SearchFactoryImplementor searchFactory, CacheMode cacheMode) { + CountDownLatch producerEndSignal, + SearchFactoryImplementor searchFactory, CacheMode cacheMode, BatchBackend backend) { this.source = entitySource; - this.destination = fromAddworkToIndex; this.monitor = monitor; this.sessionFactory = sessionFactory; + this.producerEndSignal = producerEndSignal; this.cacheMode = cacheMode; + this.backend = backend; this.documentBuilders = searchFactory.getDocumentBuildersIndexedEntities(); } @@ -61,6 +67,7 @@ transaction.commit(); } finally { + producerEndSignal.countDown(); session.close(); } log.debug( "finished" ); @@ -89,10 +96,8 @@ } catch (InterruptedException e) { // just quit + Thread.currentThread().interrupt(); } - finally { - destination.producerStopping(); - } } @SuppressWarnings("unchecked") @@ -105,7 +110,7 @@ //depending on the complexity of the object graph going to be indexed it's possible //that we hit the database several times during work construction. AddLuceneWork addWork = docBuilder.createAddWork( clazz, entity, id, idInString, true ); - destination.put( addWork ); + backend.enqueueAsyncWork( addWork ); } } Modified: search/trunk/src/main/java/org/hibernate/search/batchindexing/Executors.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/batchindexing/Executors.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/batchindexing/Executors.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -8,27 +8,34 @@ /** * Helper class to create threads; - * these threads are grouped and named to be readily identified in a profiler. + * these threads are grouped and named to be identified in a profiler. * * @author Sanne Grinovero */ public class Executors { - private static final String THREAD_GROUP_PREFIX = "Hibernate Search indexer: "; + private static final String THREAD_GROUP_PREFIX = "Hibernate Search: "; + private static final int QUEUE_MAX_LENGTH = 1000; //TODO have it configurable? /** - * Creates a new fixed size ThreadPoolExecutor - * @param threads - * @param groupname - * @return + * Creates a new fixed size ThreadPoolExecutor. + * It's using a blockingqueue of maximum 1000 elements and the rejection + * policy is set to CallerRunsPolicy for the case the queue is full. + * These settings are required to cap the queue, to make sure the + * timeouts are reasonable for most jobs. + * + * @param threads the number of threads + * @param groupname a label to identify the threadpool; useful for profiling. + * @return the new ExecutorService */ public static ThreadPoolExecutor newFixedThreadPool(int threads, String groupname) { return new ThreadPoolExecutor( threads, threads, 0L, TimeUnit.MILLISECONDS, - new LinkedBlockingQueue<Runnable>(), - new SearchThreadFactory( groupname ) ); + new LinkedBlockingQueue<Runnable>( QUEUE_MAX_LENGTH ), + new SearchThreadFactory( groupname ), + new ThreadPoolExecutor.CallerRunsPolicy() ); } /** @@ -42,15 +49,15 @@ SearchThreadFactory(String groupname) { SecurityManager s = System.getSecurityManager(); - group = (s != null)? s.getThreadGroup() : + group = ( s != null ) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); namePrefix = THREAD_GROUP_PREFIX + groupname + "-"; } public Thread newThread(Runnable r) { - Thread t = new Thread(group, r, + Thread t = new Thread( group, r, namePrefix + threadNumber.getAndIncrement(), - 0); + 0 ); return t; } Modified: search/trunk/src/main/java/org/hibernate/search/batchindexing/IdentifierConsumerEntityProducer.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/batchindexing/IdentifierConsumerEntityProducer.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/batchindexing/IdentifierConsumerEntityProducer.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -26,16 +26,16 @@ private static final Logger log = LoggerFactory.make(); - private final ProducerConsumerQueue source; - private final ProducerConsumerQueue destination; + private final ProducerConsumerQueue<List<Serializable>> source; + private final ProducerConsumerQueue<Object> destination; private final SessionFactory sessionFactory; private final CacheMode cacheMode; private final Class<?> type; private final IndexerProgressMonitor monitor; public IdentifierConsumerEntityProducer( - ProducerConsumerQueue fromIdentifierListToEntities, - ProducerConsumerQueue fromEntityToAddwork, + ProducerConsumerQueue<List<Serializable>> fromIdentifierListToEntities, + ProducerConsumerQueue<Object> fromEntityToAddwork, IndexerProgressMonitor monitor, SessionFactory sessionFactory, CacheMode cacheMode, Class<?> type) { @@ -70,6 +70,7 @@ do { take = source.take(); if ( take != null ) { + @SuppressWarnings("unchecked") List<Serializable> listIds = (List<Serializable>) take; log.trace( "received list of ids {}", listIds ); loadList( listIds, session ); @@ -79,6 +80,7 @@ } catch (InterruptedException e) { // just quit + Thread.currentThread().interrupt(); } finally { destination.producerStopping(); Modified: search/trunk/src/main/java/org/hibernate/search/batchindexing/IdentifierProducer.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/batchindexing/IdentifierProducer.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/batchindexing/IdentifierProducer.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -30,29 +30,34 @@ private static final Logger log = LoggerFactory.make(); - private final ProducerConsumerQueue destination; + private final ProducerConsumerQueue<List<Serializable>> destination; private final SessionFactory sessionFactory; private final int batchSize; private final Class<?> indexedType; private final IndexerProgressMonitor monitor; + private final int objectsLimit; + /** * @param fromIdentifierListToEntities the target queue where the produced identifiers are sent to * @param sessionFactory the Hibernate SessionFactory to use to load entities * @param objectLoadingBatchSize affects mostly the next consumer: IdentifierConsumerEntityProducer * @param indexedType the entity type to be loaded * @param monitor to monitor indexing progress + * @param objectsLimit if not zero */ public IdentifierProducer( - ProducerConsumerQueue fromIdentifierListToEntities, + ProducerConsumerQueue<List<Serializable>> fromIdentifierListToEntities, SessionFactory sessionFactory, int objectLoadingBatchSize, - Class<?> indexedType, IndexerProgressMonitor monitor) { + Class<?> indexedType, IndexerProgressMonitor monitor, + int objectsLimit) { this.destination = fromIdentifierListToEntities; this.sessionFactory = sessionFactory; this.batchSize = objectLoadingBatchSize; this.indexedType = indexedType; this.monitor = monitor; + this.objectsLimit = objectsLimit; log.trace( "created" ); } @@ -82,15 +87,18 @@ } private void loadAllIdentifiers(final StatelessSession session) throws InterruptedException { - Integer count = (Integer) session + Integer totalCount = (Integer) session .createCriteria( indexedType ) .setProjection( Projections.count( "id" ) ) .setCacheable( false ) .uniqueResult(); - - log.debug( "going to fetch {} primary keys", count); - monitor.addToTotalCount( count ); + if ( objectsLimit != 0 && objectsLimit < totalCount.intValue() ) { + totalCount = objectsLimit; + } + log.debug( "going to fetch {} primary keys", totalCount); + monitor.addToTotalCount( totalCount ); + Criteria criteria = session .createCriteria( indexedType ) .setProjection( Projections.id() ) @@ -99,6 +107,7 @@ ScrollableResults results = criteria.scroll( ScrollMode.FORWARD_ONLY ); ArrayList<Serializable> destinationList = new ArrayList<Serializable>( batchSize ); + int counter = 0; try { while ( results.next() ) { Serializable id = (Serializable) results.get( 0 ); @@ -107,6 +116,10 @@ enqueueList( destinationList ); destinationList = new ArrayList<Serializable>( batchSize ); } + counter++; + if ( counter == totalCount ) { + break; + } } } finally { Deleted: search/trunk/src/main/java/org/hibernate/search/batchindexing/IndexWritingJob.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/batchindexing/IndexWritingJob.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/batchindexing/IndexWritingJob.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -1,76 +0,0 @@ -package org.hibernate.search.batchindexing; - -import java.util.ArrayList; -import java.util.concurrent.CountDownLatch; - -import org.hibernate.search.backend.BackendQueueProcessorFactory; -import org.hibernate.search.backend.LuceneWork; -import org.hibernate.search.util.LoggerFactory; -import org.slf4j.Logger; - -/** - * This Runnable is meant to push to the backend - * all LuceneWork it can take from the ProducerConsumerQueue. - * - * FIXME: - * Is currently more an adaptor to bridge the queue approach - * to the current backend implementations. This is not by any means - * the most efficient way to index documents, it has been built only - * to test the approach. - * - */ -class IndexWritingJob implements Runnable { - - private static final Logger log = LoggerFactory.make(); - - private final ProducerConsumerQueue pleq; - private final BackendQueueProcessorFactory backend; - private final CountDownLatch endSignal; - private final IndexerProgressMonitor monitor; - - /** - * @param pleq This queue contains the LuceneWork to be send to the backend - * @param endSignal - * @param monitor - * @param backendQueueProcessorFactory - */ - IndexWritingJob(ProducerConsumerQueue pleq, CountDownLatch endSignal, IndexerProgressMonitor monitor, BackendQueueProcessorFactory backendQueueProcessorFactory) { - this.pleq = pleq; - this.monitor = monitor; - this.backend = backendQueueProcessorFactory; - this.endSignal = endSignal; - log.trace( "created" ); - } - - public void run() { - log.debug( "Start" ); - try { - while ( true ) { - Object take = pleq.take(); - if ( take == null ) { - break; - } - else { - LuceneWork work = (LuceneWork) take; - log.trace( "received lucenework {}", work ); - //TODO group work in bigger lists of size #batch and introduce the CommitLuceneWork to avoid waiting more work - ArrayList<LuceneWork> list = new ArrayList<LuceneWork>( 1 ); - list.add( work ); - Runnable processor = backend.getProcessor( list ); - processor.run(); - monitor.documentsAdded( 1L ); - } - } - log.debug( "Finished" ); - } - catch (InterruptedException e) { - // normal quit: no need to propagate interruption. - log.debug( "Interrupted" ); - } - finally { - //notify everybody we have finished. - endSignal.countDown(); - } - } - -} \ No newline at end of file Modified: search/trunk/src/main/java/org/hibernate/search/batchindexing/ProducerConsumerQueue.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/batchindexing/ProducerConsumerQueue.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/batchindexing/ProducerConsumerQueue.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -11,11 +11,13 @@ * * @author Sanne Grinovero */ -public class ProducerConsumerQueue { +public class ProducerConsumerQueue<T> { private static final int DEFAULT_BUFF_LENGHT = 1000; private static final Object exitToken = new Object(); + //doesn't use generics here as exitToken needs to be put in the queue too: + @SuppressWarnings("unchecked") private final BlockingQueue queue; private final AtomicInteger producersToWaitFor; @@ -26,6 +28,7 @@ this( DEFAULT_BUFF_LENGHT, producersToWaitFor ); } + @SuppressWarnings("unchecked") public ProducerConsumerQueue( int queueLenght, int producersToWaitFor ) { queue = new ArrayBlockingQueue( queueLenght ); this.producersToWaitFor = new AtomicInteger( producersToWaitFor ); @@ -37,7 +40,8 @@ * @return the next object in the queue, or null to exit * @throws InterruptedException */ - public Object take() throws InterruptedException { + @SuppressWarnings("unchecked") + public T take() throws InterruptedException { Object obj = queue.take(); if ( obj == exitToken ) { //restore exit signal for other threads @@ -45,7 +49,7 @@ return null; } else { - return obj; + return (T)obj; } } @@ -55,7 +59,8 @@ * @param obj * @throws InterruptedException */ - public void put(Object obj) throws InterruptedException { + @SuppressWarnings("unchecked") + public void put(T obj) throws InterruptedException { queue.put( obj ); } @@ -64,9 +69,10 @@ * finished. After doing it can safely terminate. * After all producer threads have called producerStopping() * a token will be inserted in the blocking queue to eventually - * awake sleaping consumers and have them quit, after the + * awake sleeping consumers and have them quit, after the * queue has been processed. */ + @SuppressWarnings("unchecked") public void producerStopping() { int activeProducers = producersToWaitFor.decrementAndGet(); //last producer must close consumers Modified: search/trunk/src/main/java/org/hibernate/search/engine/SearchFactoryImplementor.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/engine/SearchFactoryImplementor.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/engine/SearchFactoryImplementor.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -9,6 +9,8 @@ import org.hibernate.search.backend.BackendQueueProcessorFactory; import org.hibernate.search.backend.LuceneIndexingParameters; import org.hibernate.search.backend.Worker; +import org.hibernate.search.backend.impl.batchlucene.BatchBackend; +import org.hibernate.search.batchindexing.IndexerProgressMonitor; import org.hibernate.search.filter.FilterCachingStrategy; import org.hibernate.search.store.DirectoryProvider; import org.hibernate.search.store.optimization.OptimizerStrategy; @@ -61,4 +63,6 @@ int getFilterCacheBitResultsSize(); Set<Class<?>> getIndexedTypesPolymorphic(Class<?>[] classes); + + BatchBackend makeBatchBackend(IndexerProgressMonitor progressMonitor); } Modified: search/trunk/src/main/java/org/hibernate/search/impl/IndexerImpl.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/impl/IndexerImpl.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/impl/IndexerImpl.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -1,16 +1,15 @@ package org.hibernate.search.impl; import java.util.HashSet; -import java.util.LinkedList; -import java.util.List; import java.util.Set; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; import org.hibernate.CacheMode; import org.hibernate.SessionFactory; import org.hibernate.search.Indexer; -import org.hibernate.search.batchindexing.BatchIndexingWorkspace; +import org.hibernate.search.batchindexing.BatchCoordinator; +import org.hibernate.search.batchindexing.Executors; import org.hibernate.search.batchindexing.IndexerProgressMonitor; import org.hibernate.search.engine.SearchFactoryImplementor; import org.hibernate.search.util.LoggerFactory; @@ -30,15 +29,13 @@ private final SessionFactory sessionFactory; protected Set<Class<?>> rootEntities = new HashSet<Class<?>>(); - protected List<BatchIndexingWorkspace> indexers = new LinkedList<BatchIndexingWorkspace>(); - boolean started = false; - private CountDownLatch endAllSignal; // default settings defined here: private int objectLoadingThreads = 2; //loading the main entity private int collectionLoadingThreads = 4; //also responsible for loading of lazy @IndexedEmbedded collections private int writerThreads = 1; //also running the Analyzers private int objectLoadingBatchSize = 10; + private int objectsLimit = 0; //means no limit at all private CacheMode cacheMode = CacheMode.IGNORE; private boolean optimizeAtEnd = true; private boolean purgeAtStart = true; @@ -87,9 +84,7 @@ } } cleaned.removeAll( toRemove ); - if ( log.isDebugEnabled() ) { - log.debug( "Targets for indexing job: {}", cleaned ); - } + log.debug( "Targets for indexing job: {}", cleaned ); return cleaned; } @@ -143,30 +138,37 @@ return this; } - public void start() { - if ( started ) { - throw new IllegalStateException( "Can be started only once" ); + public Future<?> start() { + BatchCoordinator coordinator = createCoordinator(); + ExecutorService executor = Executors.newFixedThreadPool( 1, "batch coordinator" ); + try { + Future<?> submit = executor.submit( coordinator ); + return submit; } - else { - started = true; - endAllSignal = new CountDownLatch( rootEntities.size() ); - for ( Class<?> type : rootEntities ) { - indexers.add( new BatchIndexingWorkspace( - searchFactoryImplementor, sessionFactory, type, - objectLoadingThreads, collectionLoadingThreads, writerThreads, - cacheMode, objectLoadingBatchSize, - optimizeAtEnd, purgeAtStart, optimizeAfterPurge, - endAllSignal, monitor) ); - } - for ( BatchIndexingWorkspace batcher : indexers ) { - new Thread( batcher ).start(); - } + finally { + executor.shutdown(); } } + + public void startAndWait() throws InterruptedException { + BatchCoordinator coordinator = createCoordinator(); + coordinator.run(); + if ( Thread.currentThread().isInterrupted() ) { + throw new InterruptedException(); + } + } + + protected BatchCoordinator createCoordinator() { + return new BatchCoordinator( rootEntities, searchFactoryImplementor, sessionFactory, + objectLoadingThreads, collectionLoadingThreads, + cacheMode, objectLoadingBatchSize, objectsLimit, + optimizeAtEnd, purgeAtStart, optimizeAfterPurge, + monitor ); + } - public boolean startAndWait(long timeout, TimeUnit unit) throws InterruptedException { - start(); - return endAllSignal.await( timeout, unit ); + public Indexer limitObjects(int maximum) { + this.objectsLimit = maximum; + return this; } } Modified: search/trunk/src/main/java/org/hibernate/search/impl/SearchFactoryImpl.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/impl/SearchFactoryImpl.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/impl/SearchFactoryImpl.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -43,6 +43,10 @@ import org.hibernate.search.backend.Worker; import org.hibernate.search.backend.WorkerFactory; import org.hibernate.search.backend.configuration.ConfigurationParseHelper; +import org.hibernate.search.backend.configuration.MaskedProperty; +import org.hibernate.search.backend.impl.batchlucene.BatchBackend; +import org.hibernate.search.backend.impl.batchlucene.LuceneBatchBackend; +import org.hibernate.search.batchindexing.IndexerProgressMonitor; import org.hibernate.search.cfg.SearchConfiguration; import org.hibernate.search.cfg.SearchMapping; import org.hibernate.search.engine.DocumentBuilderIndexedEntity; @@ -84,6 +88,7 @@ private Map<String, Analyzer> analyzers; private final AtomicBoolean stopped = new AtomicBoolean( false ); private final int cacheBitResultsSize; + private final Properties configurationProperties; private final PolymorphicIndexHierarchy indexHierarchy = new PolymorphicIndexHierarchy(); @@ -137,6 +142,7 @@ this.cacheBitResultsSize = ConfigurationParseHelper.getIntValue( cfg.getProperties(), Environment.CACHE_DOCIDRESULTS_SIZE, CachingWrapperFilter.DEFAULT_SIZE ); + this.configurationProperties = cfg.getProperties(); this.barrier = 1; //write barrier } @@ -542,6 +548,34 @@ } //read barrier return indexHierarchy.getIndexedClasses( classes ); } + + public BatchBackend makeBatchBackend(IndexerProgressMonitor progressMonitor) { + BatchBackend batchBackend; + String impl = configurationProperties.getProperty( Environment.BATCH_BACKEND ); + if ( StringHelper.isEmpty( impl ) || "LuceneBatch".equalsIgnoreCase( impl ) ) { + batchBackend = new LuceneBatchBackend(); + } + else { + try { + Class batchBackendClass = ReflectHelper + .classForName( impl, SearchFactoryImpl.class ); + batchBackend = ( BatchBackend ) batchBackendClass.newInstance(); + } + catch ( ClassNotFoundException e ) { + throw new SearchException( "Unable to find batchbackend implementation class: " + impl, e ); + } + catch ( IllegalAccessException e ) { + throw new SearchException( "Unable to instantiate batchbackend class: " + impl, e ); + } + catch ( InstantiationException e ) { + throw new SearchException( "Unable to instantiate batchbackend class: " + impl, e ); + } + } + Properties batchBackendConfiguration = new MaskedProperty( + this.configurationProperties, Environment.BATCH_BACKEND ); + batchBackend.initialize( batchBackendConfiguration, progressMonitor, this ); + return batchBackend; + } /** * Helper class which keeps track of all super classes and interfaces of the indexed entities. Modified: search/trunk/src/main/java/org/hibernate/search/impl/SimpleIndexingProgressMonitor.java =================================================================== --- search/trunk/src/main/java/org/hibernate/search/impl/SimpleIndexingProgressMonitor.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/main/java/org/hibernate/search/impl/SimpleIndexingProgressMonitor.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -42,14 +42,14 @@ } protected int getStatusMessagePeriod() { - return 1000; + return 50; } protected void printStatusMessage(long starttimems, long totalTodoCount, long doneCount) { long elapsedMs = System.currentTimeMillis() - starttimems; log.info( "{} documents indexed in {} ms", doneCount, elapsedMs ); - double estimateSpeed = ( (double) doneCount * 1000f ) / elapsedMs ; - double estimatePercentileComplete = ( (double) doneCount*100 ) / (double)totalTodoCount ; + float estimateSpeed = (float) doneCount * 1000f / elapsedMs ; + float estimatePercentileComplete = (float) doneCount * 100f / (float) totalTodoCount ; log.info( "Indexing speed: {} documents/second; progress: {}%", estimateSpeed, estimatePercentileComplete ); } Modified: search/trunk/src/test/java/org/hibernate/search/test/batchindexing/AncientBook.java =================================================================== --- search/trunk/src/test/java/org/hibernate/search/test/batchindexing/AncientBook.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/test/java/org/hibernate/search/test/batchindexing/AncientBook.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -10,4 +10,12 @@ public String catalogueGroupName = ""; + public String getCatalogueGroupName() { + return catalogueGroupName; + } + + public void setCatalogueGroupName(String catalogueGroupName) { + this.catalogueGroupName = catalogueGroupName; + } + } Modified: search/trunk/src/test/java/org/hibernate/search/test/batchindexing/Book.java =================================================================== --- search/trunk/src/test/java/org/hibernate/search/test/batchindexing/Book.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/test/java/org/hibernate/search/test/batchindexing/Book.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -1,15 +1,36 @@ package org.hibernate.search.test.batchindexing; import javax.persistence.Entity; +import javax.persistence.GeneratedValue; import javax.persistence.Id; +import org.hibernate.search.annotations.Field; import org.hibernate.search.annotations.Indexed; @Indexed @Entity -public class Book { +public class Book implements TitleAble { - @Id - public long id; + private long id; + + private String title; + @Id @GeneratedValue + public long getId() { + return id; + } + + public void setId(long id) { + this.id = id; + } + + @Field + public String getTitle() { + return title; + } + + public void setTitle(String title) { + this.title = title; + } + } Modified: search/trunk/src/test/java/org/hibernate/search/test/batchindexing/Dvd.java =================================================================== --- search/trunk/src/test/java/org/hibernate/search/test/batchindexing/Dvd.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/test/java/org/hibernate/search/test/batchindexing/Dvd.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -9,7 +9,7 @@ @Indexed @Entity -public class Dvd { +public class Dvd implements TitleAble { public long unusuallyNamedIdentifier; public String title; Added: search/trunk/src/test/java/org/hibernate/search/test/batchindexing/IndexingGeneratedCorpusTest.java =================================================================== --- search/trunk/src/test/java/org/hibernate/search/test/batchindexing/IndexingGeneratedCorpusTest.java (rev 0) +++ search/trunk/src/test/java/org/hibernate/search/test/batchindexing/IndexingGeneratedCorpusTest.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -0,0 +1,147 @@ +package org.hibernate.search.test.batchindexing; + +import junit.framework.TestCase; + +import org.apache.lucene.search.MatchAllDocsQuery; +import org.apache.lucene.search.Query; +import org.hibernate.Transaction; +import org.hibernate.search.Environment; +import org.hibernate.search.FullTextQuery; +import org.hibernate.search.FullTextSession; +import org.hibernate.search.backend.impl.batchlucene.LuceneBatchBackend; +import org.hibernate.search.test.util.FullTextSessionBuilder; +import org.hibernate.search.test.util.textbuilder.SentenceInventor; + +/** + * Tests the fullTextSession.createIndexer() API + * for basic functionality. + * + * @author Sanne Grinovero + */ +public class IndexingGeneratedCorpusTest extends TestCase { + + private final int BOOK_NUM = 3000; + private final int ANCIENTBOOK_NUM = 200; + private final int DVD_NUM = 2000; + + private SentenceInventor sentenceInventor = new SentenceInventor( 7L, 10000 ); + private FullTextSessionBuilder builder; + private int totalEntitiesInDB = 0; + + @Override + public void setUp() throws Exception { + super.setUp(); + builder = new FullTextSessionBuilder(); + builder + .addAnnotatedClass( Book.class ) + .addAnnotatedClass( Dvd.class ) + .addAnnotatedClass( AncientBook.class ) + .setProperty( "hibernate.show_sql", "false" ) //too verbose for this test + .setProperty( LuceneBatchBackend.CONCURRENT_WRITERS, "4" ) + .build(); + createMany( Book.class, BOOK_NUM ); + createMany( Dvd.class, DVD_NUM ); + createMany( AncientBook.class, ANCIENTBOOK_NUM ); + } + + private void createMany(Class<? extends TitleAble> entityType, int amount ) throws InstantiationException, IllegalAccessException { + FullTextSession fullTextSession = builder.openFullTextSession(); + try { + Transaction tx = fullTextSession.beginTransaction(); + for ( int i = 0; i < amount; i++ ) { + TitleAble instance = entityType.newInstance(); + instance.setTitle( sentenceInventor.nextSentence() ); + fullTextSession.persist( instance ); + totalEntitiesInDB++; + if ( i % 250 == 249 ) { + tx.commit(); + fullTextSession.clear(); + System.out.println( "Test preparation: " + totalEntitiesInDB + " entities persisted" ); + tx = fullTextSession.beginTransaction(); + } + } + tx.commit(); + } + finally { + fullTextSession.close(); + } + } + + public void testBatchIndexing() throws InterruptedException { + verifyResultNumbers(); //initial count of entities should match expectations + purgeAll(); // empty indexes + verifyIsEmpty(); + reindexAll(); // rebuild the indexes + verifyResultNumbers(); // verify the count match again + reindexAll(); //tests that purgeAll is automatic: + verifyResultNumbers(); //..same numbers again + } + + private void reindexAll() throws InterruptedException { + FullTextSession fullTextSession = builder.openFullTextSession(); + try { + fullTextSession.createIndexer( Object.class ) + .documentBuilderThreads( 8 ) + .objectLoadingThreads( 4 ) + .objectLoadingBatchSize( 30 ) + .startAndWait(); + } + finally { + fullTextSession.close(); + } + } + + private void purgeAll() { + FullTextSession fullTextSession = builder.openFullTextSession(); + try { + Transaction tx = fullTextSession.beginTransaction(); + fullTextSession.purgeAll( Object.class ); + tx.commit(); + } + finally { + fullTextSession.close(); + } + } + + @SuppressWarnings("unchecked") + private void verifyResultNumbers() { + assertEquals( DVD_NUM, + countByFT( Dvd.class ) ); + assertEquals( ANCIENTBOOK_NUM + BOOK_NUM, + countByFT( Book.class ) ); + assertEquals( ANCIENTBOOK_NUM, + countByFT( AncientBook.class ) ); + assertEquals( DVD_NUM + ANCIENTBOOK_NUM + BOOK_NUM, + countByFT( AncientBook.class, Book.class, Dvd.class ) ); + assertEquals( DVD_NUM + ANCIENTBOOK_NUM, + countByFT( AncientBook.class, Dvd.class ) ); + } + + @SuppressWarnings("unchecked") + private void verifyIsEmpty() { + assertEquals( 0, countByFT( Dvd.class ) ); + assertEquals( 0, countByFT( Book.class ) ); + assertEquals( 0, countByFT( AncientBook.class ) ); + assertEquals( 0, countByFT( AncientBook.class, Book.class, Dvd.class ) ); + } + + private int countByFT(Class<? extends TitleAble>... types) { + Query findAll = new MatchAllDocsQuery(); + int bySize = 0; + int byResultSize = 0; + FullTextSession fullTextSession = builder.openFullTextSession(); + try { + Transaction tx = fullTextSession.beginTransaction(); + FullTextQuery fullTextQuery = fullTextSession.createFullTextQuery( findAll, types ); + bySize = fullTextQuery.list().size(); + byResultSize = fullTextQuery.getResultSize(); + tx.commit(); + } + finally { + fullTextSession.close(); + } + assertEquals( bySize, byResultSize ); + return bySize; + } + +} Property changes on: search/trunk/src/test/java/org/hibernate/search/test/batchindexing/IndexingGeneratedCorpusTest.java ___________________________________________________________________ Name: svn:keywords + Id Modified: search/trunk/src/test/java/org/hibernate/search/test/batchindexing/SearchIndexerTest.java =================================================================== --- search/trunk/src/test/java/org/hibernate/search/test/batchindexing/SearchIndexerTest.java 2009-06-29 20:18:57 UTC (rev 16966) +++ search/trunk/src/test/java/org/hibernate/search/test/batchindexing/SearchIndexerTest.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -1,7 +1,6 @@ package org.hibernate.search.test.batchindexing; import java.util.Set; -import java.util.concurrent.TimeUnit; import junit.framework.TestCase; @@ -18,6 +17,10 @@ public class SearchIndexerTest extends TestCase { + /** + * test that the Indexer is properly identifying the root entities + * from the selection of classes to be indexed. + */ public void testEntityHierarchy() { FullTextSessionBuilder ftsb = new FullTextSessionBuilder() .addAnnotatedClass( ModernBook.class ) @@ -56,7 +59,7 @@ } } - private class TestableSearchIndexerImpl extends IndexerImpl { + private static class TestableSearchIndexerImpl extends IndexerImpl { protected TestableSearchIndexerImpl(SearchFactoryImplementor searchFactory, Class<?>... types) { super( searchFactory, null, types ); @@ -68,6 +71,10 @@ } + /** + * Test to verify that the identifier loading works even when + * the property is not called "id" + */ public void testIdentifierNaming() throws InterruptedException { //disable automatic indexing, to test manual index creation. FullTextSessionBuilder ftsb = new FullTextSessionBuilder() @@ -95,7 +102,7 @@ { FullTextSession fullTextSession = ftsb.openFullTextSession(); fullTextSession.createIndexer( Dvd.class ) - .startAndWait( 10, TimeUnit.SECONDS ); + .startAndWait(); fullTextSession.close(); } { @@ -104,8 +111,9 @@ } } - public int countResults( Term termForQuery, FullTextSessionBuilder ftSessionBuilder, Class<?> type ) { - TermQuery fullTextQuery = new TermQuery( new Term( "title", "trek" ) ); + //helper method + private int countResults( Term termForQuery, FullTextSessionBuilder ftSessionBuilder, Class<?> type ) { + TermQuery fullTextQuery = new TermQuery( termForQuery ); FullTextSession fullTextSession = ftSessionBuilder.openFullTextSession(); Transaction transaction = fullTextSession.beginTransaction(); FullTextQuery query = fullTextSession.createFullTextQuery( fullTextQuery, type ); Added: search/trunk/src/test/java/org/hibernate/search/test/batchindexing/TitleAble.java =================================================================== --- search/trunk/src/test/java/org/hibernate/search/test/batchindexing/TitleAble.java (rev 0) +++ search/trunk/src/test/java/org/hibernate/search/test/batchindexing/TitleAble.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -0,0 +1,9 @@ +package org.hibernate.search.test.batchindexing; + +public interface TitleAble { + + public String getTitle(); + + public void setTitle(String title); + +} Property changes on: search/trunk/src/test/java/org/hibernate/search/test/batchindexing/TitleAble.java ___________________________________________________________________ Name: svn:keywords + Id Added: search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/SentenceInventor.java =================================================================== --- search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/SentenceInventor.java (rev 0) +++ search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/SentenceInventor.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -0,0 +1,154 @@ +package org.hibernate.search.test.util.textbuilder; + +import java.util.Random; +import java.util.Set; +import java.util.TreeSet; + +/** + * Test utility meant to produce sentences of a randomly generated language, + * having some properties of natural languages. + * The goal is to produce sentences which look like a western text, + * but are not. + * All sentences from the same SentenceInventor will share + * a limited dictionary, making the frequencies suitable to test + * with Lucene. + * Sentences produced depend from the constructor arguments, + * making the output predictable for testing purposes. + * + * @author Sanne Grinovero + */ +public class SentenceInventor { + + private final Random r; + private final WordDictionary dictionary; + //array contains repeated object for probability distribution (more chance for a ",") + private final char[] sentenceSeparators = new char[] { ',', ',', ',' , ';', ':', ':' }; + + /** + * @param randomSeed the seed to use for random generator + * @param dictionarySize the number of terms to insert in the dictionary used to build sentences + */ + public SentenceInventor(long randomSeed, int dictionarySize) { + r = new Random( randomSeed ); + dictionary = randomDictionary( dictionarySize ); + } + + /** + * @return a random character from the ASCII table (text chars only) + */ + public char randomCharacter() { + return (char) (r.nextInt( 26 ) + 65); + } + + /** + * @param length the desired length + * @return a randomly generated String + */ + public String randomString(int length) { + char[] chars = new char[length]; + for ( int i=0; i<length; i++ ) { + chars[i] = randomCharacter(); + } + return new String( chars ); + } + + /** + * Produces a randomly generated String, using + * only western alphabet characters and selecting + * the length as a normal distribution of natural languages. + * @return the generated String + */ + public String randomString() { + double d = r.nextGaussian() * 6.3d; + int l = (int) d + 6; + if ( l > 0 ) + return randomString( l ); + else + return randomString(); + } + + /** + * Produces a random String, which might be lowercase, + * completely uppercase, or uppercasing the first char + * (randomly selected) + * @return produced String + */ + public String randomTerm() { + int i = r.nextInt( 200 ); + String term = randomString(); + if ( i > 10 ) + //completely lowercase 189/200 cases + return term.toLowerCase(); + else if ( i < 2 ) + //completely uppercase in 2/200 cases + return term; + else + //first letter upercase in 9/200 cases + return term.substring( 0, 1 ) + term.substring( 1 ).toLowerCase(); + } + + private WordDictionary randomDictionary(int size) { + Set<String> tree = new TreeSet<String>(); + while ( tree.size() != size ) { + tree.add( randomTerm() ); + } + return new WordDictionary( tree ); + } + + /** + * Builds a sentence concatenating terms from the generated dictionary and spaces + * @return a sentence + */ + public String nextSentence() { + int sentenceLength = r.nextInt( 3 ) + r.nextInt( 10 ) + 1; + String[] sentence = new String[sentenceLength]; + for ( int i=0; i<sentenceLength; i++ ) { + sentence[i] = dictionary.randomWord(); + } + if ( sentenceLength == 1 ) { + return sentence[0]; + } + else { + StringBuilder sb = new StringBuilder( sentence[0]); + for ( int i=1; i<sentenceLength; i++) { + sb.append( " " ); + sb.append( sentence[i] ); + } + return sb.toString(); + } + } + + /** + * Combines a random (gaussian) number of sentences in a period, + * using some punctuation symbols and + * capitalizing first char, terminating with dot and newline. + * @return + */ + public String nextPeriod() { + int periodLengthSentences = r.nextInt( 7 ) - 2; + periodLengthSentences = ( periodLengthSentences < 1 ) ? 1 : periodLengthSentences; + String firstsentence = nextSentence(); + StringBuilder sb = new StringBuilder() + .append( firstsentence.substring( 0,1 ).toUpperCase() ) + .append( firstsentence.substring( 1 ) ); + for ( int i=1; i<periodLengthSentences; i++ ) { + int separatorCharIndex = r.nextInt( sentenceSeparators.length ); + sb + .append( sentenceSeparators[separatorCharIndex] ) + .append( ' ' ) + .append( nextSentence() ); + } + sb.append( ".\n" ); + return sb.toString(); + } + + //run it to get an idea of what this class is going to produce + public static void main(String[] args) { + SentenceInventor wi = new SentenceInventor( 7L, 10000 ); + for (int i=0; i<30; i++) { + System.out.print( wi.nextPeriod() ); + } + } + +} + Property changes on: search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/SentenceInventor.java ___________________________________________________________________ Name: svn:keywords + Id Added: search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/TextProductionTest.java =================================================================== --- search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/TextProductionTest.java (rev 0) +++ search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/TextProductionTest.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -0,0 +1,23 @@ +package org.hibernate.search.test.util.textbuilder; + +import junit.framework.TestCase; + +/** + * Tests WordDictionary and WordInventor, + * these are test utilities not part of the Search distribution; + * the test exists to spot if the text they produce is unchanged, so + * that other tests can rely on working test utilities. + * + * @see WordDictionary + * @see SentenceInventor + * + * @author Sanne Grinovero + */ +public class TextProductionTest extends TestCase { + + public void testSomeWordsGetBuilt() { + SentenceInventor wi = new SentenceInventor( 7L, 10000 ); + assertEquals( "Qoswo, orrmi ag ybwp bbtb kw qgtqaon lyhk nbv: qrqm flyui hyshm jmpqyb qmolml fjxw gnumocv Twwg.\n", wi.nextPeriod() ); + } + +} Property changes on: search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/TextProductionTest.java ___________________________________________________________________ Name: svn:keywords + Id Added: search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/WordDictionary.java =================================================================== --- search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/WordDictionary.java (rev 0) +++ search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/WordDictionary.java 2009-06-29 22:25:43 UTC (rev 16967) @@ -0,0 +1,50 @@ +package org.hibernate.search.test.util.textbuilder; + +import java.util.Arrays; +import java.util.Comparator; +import java.util.Random; +import java.util.Set; + +/** + * Test utility meant to build a consistent dictionary of words. + * This is not just a random generator: like in natural + * languages shorter terms have a higher frequency in a text corpus + * and the dictionary size is limited. + * + * @author Sanne Grinovero + */ +public class WordDictionary { + + private final String[] positionalWords; + private final int maxSize; + private final double gaussFactor; + + private static final Random r = new Random( 12L ); + + public WordDictionary(Set<String> words) { + this.positionalWords = words.toArray( new String[0] ); + //sort by String length. Languages use shorter terms more often. + Arrays.sort( positionalWords, new StringLengthComparator() ); + maxSize = positionalWords.length; + gaussFactor = ((double)maxSize +1 ) / 4d ; + } + + private static class StringLengthComparator implements Comparator<String> { + + public int compare(String o1, String o2) { + return o1.length()-o2.length(); + } + + } + + public String randomWord() { + int position = Math.abs((int) ( r.nextGaussian() * gaussFactor ) ); + if ( position < maxSize ) { + return positionalWords[position]; + } + else { + return randomWord(); + } + } + +} Property changes on: search/trunk/src/test/java/org/hibernate/search/test/util/textbuilder/WordDictionary.java ___________________________________________________________________ Name: svn:keywords + Id