Author: sannegrinovero Date: 2008-06-26 20:25:45 -0400 (Thu, 26 Jun 2008) New Revision: 14817 Modified: search/trunk/src/java/org/hibernate/search/backend/Workspace.java search/trunk/src/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java search/trunk/src/java/org/hibernate/search/engine/SearchFactoryImplementor.java search/trunk/src/java/org/hibernate/search/impl/FullTextSessionImpl.java search/trunk/src/java/org/hibernate/search/impl/SearchFactoryImpl.java search/trunk/src/java/org/hibernate/search/reader/SharedReaderProvider.java search/trunk/src/java/org/hibernate/search/reader/SharingBufferReaderProvider.java search/trunk/src/java/org/hibernate/search/store/DirectoryProviderFactory.java search/trunk/src/java/org/hibernate/search/store/FSMasterDirectoryProvider.java search/trunk/src/test/org/hibernate/search/test/reader/performance/ReaderPerformance.java Log: minor DirectoryProviders's Locks reorganization (and relative API) Modified: search/trunk/src/java/org/hibernate/search/backend/Workspace.java =================================================================== --- search/trunk/src/java/org/hibernate/search/backend/Workspace.java 2008-06-26 22:15:36 UTC (rev 14816) +++ search/trunk/src/java/org/hibernate/search/backend/Workspace.java 2008-06-27 00:25:45 UTC (rev 14817) @@ -4,7 +4,7 @@ import java.io.IOException; import java.util.HashMap; import java.util.Map; -import java.util.concurrent.locks.ReentrantLock; +import java.util.concurrent.locks.Lock; import org.apache.lucene.analysis.Analyzer; import org.apache.lucene.analysis.SimpleAnalyzer; @@ -189,7 +189,7 @@ private class DPWorkspace { private final DirectoryProvider directoryProvider; - private final ReentrantLock lock; + private final Lock lock; private IndexReader reader; private IndexWriter writer; @@ -199,7 +199,7 @@ DPWorkspace(DirectoryProvider dp) { this.directoryProvider = dp; - this.lock = searchFactoryImplementor.getLockableDirectoryProviders().get( dp ); + this.lock = searchFactoryImplementor.getDirectoryProviderLock( dp ); } public boolean needsOptimization() { Modified: search/trunk/src/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java =================================================================== --- search/trunk/src/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java 2008-06-26 22:15:36 UTC (rev 14816) +++ search/trunk/src/java/org/hibernate/search/backend/impl/BatchedQueueingProcessor.java 2008-06-27 00:25:45 UTC (rev 14817) @@ -38,14 +38,13 @@ private static final Logger log = LoggerFactory.getLogger( BatchedQueueingProcessor.class ); - private boolean sync; - private int batchSize; - private ExecutorService executorService; - private BackendQueueProcessorFactory backendQueueProcessorFactory; - private SearchFactoryImplementor searchFactoryImplementor; + private final boolean sync; + private final int batchSize; + private final ExecutorService executorService; + private final BackendQueueProcessorFactory backendQueueProcessorFactory; + private final SearchFactoryImplementor searchFactoryImplementor; - public BatchedQueueingProcessor(SearchFactoryImplementor searchFactoryImplementor, - Properties properties) { + public BatchedQueueingProcessor(SearchFactoryImplementor searchFactoryImplementor, Properties properties) { this.searchFactoryImplementor = searchFactoryImplementor; //default to sync if none defined this.sync = !"async".equalsIgnoreCase( properties.getProperty( Environment.WORKER_EXECUTION ) ); @@ -71,6 +70,9 @@ new ThreadPoolExecutor.CallerRunsPolicy() ); } + else { + executorService = null; + } String backend = properties.getProperty( Environment.WORKER_BACKEND ); if ( StringHelper.isEmpty( backend ) || "lucene".equalsIgnoreCase( backend ) ) { backendQueueProcessorFactory = new LuceneBackendQueueProcessorFactory(); Modified: search/trunk/src/java/org/hibernate/search/engine/SearchFactoryImplementor.java =================================================================== --- search/trunk/src/java/org/hibernate/search/engine/SearchFactoryImplementor.java 2008-06-26 22:15:36 UTC (rev 14816) +++ search/trunk/src/java/org/hibernate/search/engine/SearchFactoryImplementor.java 2008-06-27 00:25:45 UTC (rev 14817) @@ -3,7 +3,7 @@ import java.util.Map; import java.util.Set; -import java.util.concurrent.locks.ReentrantLock; +import java.util.concurrent.locks.Lock; import org.hibernate.search.SearchFactory; import org.hibernate.search.backend.BackendQueueProcessorFactory; @@ -26,8 +26,6 @@ Map<Class, DocumentBuilder<Object>> getDocumentBuilders(); - Map<DirectoryProvider, ReentrantLock> getLockableDirectoryProviders(); - Worker getWorker(); void addOptimizerStrategy(DirectoryProvider<?> provider, OptimizerStrategy optimizerStrategy); @@ -49,4 +47,11 @@ void addClassToDirectoryProvider(Class clazz, DirectoryProvider<?> directoryProvider); Set<Class> getClassesInDirectoryProvider(DirectoryProvider<?> directoryProvider); + + Set<DirectoryProvider> getDirectoryProviders(); + + Lock getDirectoryProviderLock(DirectoryProvider dp); + + void addDirectoryProvider(DirectoryProvider<?> provider); + } Modified: search/trunk/src/java/org/hibernate/search/impl/FullTextSessionImpl.java =================================================================== --- search/trunk/src/java/org/hibernate/search/impl/FullTextSessionImpl.java 2008-06-26 22:15:36 UTC (rev 14816) +++ search/trunk/src/java/org/hibernate/search/impl/FullTextSessionImpl.java 2008-06-27 00:25:45 UTC (rev 14817) @@ -43,7 +43,6 @@ import org.hibernate.search.FullTextQuery; import org.hibernate.search.FullTextSession; import org.hibernate.search.SearchFactory; -import org.hibernate.search.SearchException; import org.hibernate.search.backend.Work; import org.hibernate.search.backend.WorkType; import org.hibernate.search.engine.DocumentBuilder; @@ -153,9 +152,9 @@ //need to add elements in a queue kept at the Session level //the queue will be processed by a Lucene(Auto)FlushEventListener //note that we could keep this queue somewhere in the event listener in the mean time but that requires - // a synchronized hashmap holding this queue on a per session basis plus some session house keeping (yuk) - //an other solution would be to subclass SessionImpl instead of having this LuceneSession delecation model - // this is an open discussion + //a synchronized hashmap holding this queue on a per session basis plus some session house keeping (yuk) + //another solution would be to subclass SessionImpl instead of having this LuceneSession delegation model + //this is an open discussion } public SearchFactory getSearchFactory() { Modified: search/trunk/src/java/org/hibernate/search/impl/SearchFactoryImpl.java =================================================================== --- search/trunk/src/java/org/hibernate/search/impl/SearchFactoryImpl.java 2008-06-26 22:15:36 UTC (rev 14816) +++ search/trunk/src/java/org/hibernate/search/impl/SearchFactoryImpl.java 2008-06-27 00:25:45 UTC (rev 14817) @@ -13,6 +13,8 @@ import java.util.Properties; import java.util.Set; import java.util.WeakHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import org.apache.lucene.analysis.Analyzer; @@ -60,29 +62,25 @@ Version.touch(); } - private static final Logger log = LoggerFactory.getLogger( SearchFactoryImpl.class ); + private final Logger log = LoggerFactory.getLogger( SearchFactoryImpl.class ); private final Map<Class, DocumentBuilder<Object>> documentBuilders = new HashMap<Class, DocumentBuilder<Object>>(); //keep track of the index modifiers per DirectoryProvider since multiple entity can use the same directory provider - //TODO move the ReentrantLock into DirectoryProviderData.lock, add a getDPLock(DP) and add a Set<DP> getDirectoryProviders() method. - private final Map<DirectoryProvider, ReentrantLock> lockableDirectoryProviders = - new HashMap<DirectoryProvider, ReentrantLock>(); - private final Map<DirectoryProvider, DirectoryProviderData> dirProviderData = - new HashMap<DirectoryProvider, DirectoryProviderData>(); - private Worker worker; - private ReaderProvider readerProvider; + private final Map<DirectoryProvider, DirectoryProviderData> dirProviderData = new HashMap<DirectoryProvider, DirectoryProviderData>(); + private final Worker worker; + private final ReaderProvider readerProvider; private BackendQueueProcessorFactory backendQueueProcessorFactory; private final Map<String, FilterDef> filterDefinitions = new HashMap<String, FilterDef>(); - private FilterCachingStrategy filterCachingStrategy; + private final FilterCachingStrategy filterCachingStrategy; private Map<String, Analyzer> analyzers; - private boolean stopped = false; + private final AtomicBoolean stopped = new AtomicBoolean( false ); /** * Each directory provider (index) can have its own performance settings. */ private Map<DirectoryProvider, LuceneIndexingParameters> dirProviderIndexingParams = new HashMap<DirectoryProvider, LuceneIndexingParameters>(); - private String indexingStrategy; + private final String indexingStrategy; public BackendQueueProcessorFactory getBackendQueueProcessorFactory() { @@ -97,24 +95,24 @@ public SearchFactoryImpl(Configuration cfg) { //yuk ReflectionManager reflectionManager = getReflectionManager( cfg ); - setIndexingStrategy(cfg); //need to be done before the document builds - InitContext context = new InitContext(cfg); - initDocumentBuilders(cfg, reflectionManager, context ); + this.indexingStrategy = defineIndexingStrategy( cfg ); //need to be done before the document builds + initDocumentBuilders( cfg, reflectionManager ); Set<Class> indexedClasses = documentBuilders.keySet(); for (DocumentBuilder builder : documentBuilders.values()) { builder.postInitialize( indexedClasses ); } - worker = WorkerFactory.createWorker( cfg, this ); - readerProvider = ReaderProviderFactory.createReaderProvider( cfg, this ); - buildFilterCachingStrategy( cfg.getProperties() ); + this.worker = WorkerFactory.createWorker( cfg, this ); + this.readerProvider = ReaderProviderFactory.createReaderProvider( cfg, this ); + this.filterCachingStrategy = buildFilterCachingStrategy( cfg.getProperties() ); } - private void setIndexingStrategy(Configuration cfg) { - indexingStrategy = cfg.getProperties().getProperty( Environment.INDEXING_STRATEGY, "event" ); + private static String defineIndexingStrategy(Configuration cfg) { + String indexingStrategy = cfg.getProperties().getProperty( Environment.INDEXING_STRATEGY, "event" ); if ( ! ("event".equals( indexingStrategy ) || "manual".equals( indexingStrategy ) ) ) { - throw new SearchException(Environment.INDEXING_STRATEGY + " unknown: " + indexingStrategy); + throw new SearchException( Environment.INDEXING_STRATEGY + " unknown: " + indexingStrategy ); } + return indexingStrategy; } public String getIndexingStrategy() { @@ -122,8 +120,7 @@ } public void close() { - if (!stopped) { - stopped = true; + if ( stopped.compareAndSet( false, true) ) { try { worker.close(); } @@ -131,7 +128,7 @@ log.error( "Worker raises an exception on close()", e ); } //TODO move to DirectoryProviderFactory for cleaner - for (DirectoryProvider dp : lockableDirectoryProviders.keySet() ) { + for (DirectoryProvider dp : getDirectoryProviders() ) { try { dp.stop(); } @@ -223,7 +220,6 @@ SearchFactoryImpl searchFactory = contextMap.get( cfg ); if ( searchFactory == null ) { searchFactory = new SearchFactoryImpl( cfg ); - contextMap.put( cfg, searchFactory ); } return searchFactory; @@ -234,8 +230,8 @@ return documentBuilders; } - public Map<DirectoryProvider, ReentrantLock> getLockableDirectoryProviders() { - return lockableDirectoryProviders; + public Set<DirectoryProvider> getDirectoryProviders() { + return this.dirProviderData.keySet(); } public Worker getWorker() { @@ -272,7 +268,7 @@ ReflectionManager reflectionManager; try { //TODO introduce a ReflectionManagerHolder interface to avoid reflection - //I want to avoid hard link between HAN and Validator for usch a simple need + //I want to avoid hard link between HAN and Validator for such a simple need //reuse the existing reflectionManager one when possible reflectionManager = (ReflectionManager) cfg.getClass().getMethod( "getReflectionManager" ).invoke( cfg ); @@ -311,11 +307,12 @@ return analyzer; } - private void initDocumentBuilders(Configuration cfg, ReflectionManager reflectionManager, InitContext context) { + private void initDocumentBuilders(Configuration cfg, ReflectionManager reflectionManager) { + InitContext context = new InitContext( cfg ); Iterator iter = cfg.getClassMappings(); DirectoryProviderFactory factory = new DirectoryProviderFactory(); - while (iter.hasNext()) { + while ( iter.hasNext() ) { PersistentClass clazz = (PersistentClass) iter.next(); Class<?> mappedClass = clazz.getMappedClass(); if (mappedClass != null) { @@ -340,7 +337,8 @@ factory.startDirectoryProviders(); } - private void buildFilterCachingStrategy(Properties properties) { + private static FilterCachingStrategy buildFilterCachingStrategy(Properties properties) { + FilterCachingStrategy filterCachingStrategy; String impl = properties.getProperty( Environment.FILTER_CACHING_STRATEGY ); if ( StringHelper.isEmpty( impl ) || "mru".equalsIgnoreCase( impl ) ) { filterCachingStrategy = new MRUFilterCachingStrategy(); @@ -361,6 +359,7 @@ } } filterCachingStrategy.initialize( properties ); + return filterCachingStrategy; } public FilterCachingStrategy getFilterCachingStrategy() { @@ -372,7 +371,17 @@ } private static class DirectoryProviderData { + public final Lock dirLock = new ReentrantLock(); public OptimizerStrategy optimizerStrategy; public Set<Class> classes = new HashSet<Class>(2); } + + public Lock getDirectoryProviderLock(DirectoryProvider dp) { + return this.dirProviderData.get( dp ).dirLock; + } + + public void addDirectoryProvider(DirectoryProvider<?> provider) { + this.dirProviderData.put( provider, new DirectoryProviderData() ); + } + } Modified: search/trunk/src/java/org/hibernate/search/reader/SharedReaderProvider.java =================================================================== --- search/trunk/src/java/org/hibernate/search/reader/SharedReaderProvider.java 2008-06-26 22:15:36 UTC (rev 14816) +++ search/trunk/src/java/org/hibernate/search/reader/SharedReaderProvider.java 2008-06-27 00:25:45 UTC (rev 14817) @@ -282,7 +282,7 @@ } public void initialize(Properties props, SearchFactoryImplementor searchFactoryImplementor) { - Set<DirectoryProvider> providers = searchFactoryImplementor.getLockableDirectoryProviders().keySet(); + Set<DirectoryProvider> providers = searchFactoryImplementor.getDirectoryProviders(); perDirectoryProviderManipulationLocks = new HashMap<DirectoryProvider, Lock>( providers.size() ); for (DirectoryProvider dp : providers) { perDirectoryProviderManipulationLocks.put( dp, new ReentrantLock() ); Modified: search/trunk/src/java/org/hibernate/search/reader/SharingBufferReaderProvider.java =================================================================== --- search/trunk/src/java/org/hibernate/search/reader/SharingBufferReaderProvider.java 2008-06-26 22:15:36 UTC (rev 14816) +++ search/trunk/src/java/org/hibernate/search/reader/SharingBufferReaderProvider.java 2008-06-27 00:25:45 UTC (rev 14817) @@ -63,7 +63,7 @@ public void initialize(Properties props, SearchFactoryImplementor searchFactoryImplementor) { Map<DirectoryProvider,PerDirectoryLatestReader> map = new HashMap<DirectoryProvider,PerDirectoryLatestReader>(); - Set<DirectoryProvider> providers = searchFactoryImplementor.getLockableDirectoryProviders().keySet(); + Set<DirectoryProvider> providers = searchFactoryImplementor.getDirectoryProviders(); for ( DirectoryProvider provider : providers ) { try { map.put( provider, new PerDirectoryLatestReader( provider ) ); Modified: search/trunk/src/java/org/hibernate/search/store/DirectoryProviderFactory.java =================================================================== --- search/trunk/src/java/org/hibernate/search/store/DirectoryProviderFactory.java 2008-06-26 22:15:36 UTC (rev 14816) +++ search/trunk/src/java/org/hibernate/search/store/DirectoryProviderFactory.java 2008-06-27 00:25:45 UTC (rev 14817) @@ -47,7 +47,7 @@ public class DirectoryProviderFactory { private final List<DirectoryProvider<?>> providers = new ArrayList<DirectoryProvider<?>>(); - private static String DEFAULT_DIRECTORY_PROVIDER = FSDirectoryProvider.class.getName(); + private static final String DEFAULT_DIRECTORY_PROVIDER = FSDirectoryProvider.class.getName(); private static final String SHARDING_STRATEGY = "sharding_strategy"; private static final String NBR_OF_SHARDS = SHARDING_STRATEGY + ".nbr_of_shards"; @@ -149,8 +149,8 @@ configureIndexingParameters( searchFactoryImplementor, indexProps, provider ); providers.add( provider ); searchFactoryImplementor.addClassToDirectoryProvider(entity, provider); - if ( !searchFactoryImplementor.getLockableDirectoryProviders().containsKey( provider ) ) { - searchFactoryImplementor.getLockableDirectoryProviders().put( provider, new ReentrantLock() ); + if ( ! searchFactoryImplementor.getDirectoryProviders().contains( provider ) ) { + searchFactoryImplementor.addDirectoryProvider( provider ); } return provider; } Modified: search/trunk/src/java/org/hibernate/search/store/FSMasterDirectoryProvider.java =================================================================== --- search/trunk/src/java/org/hibernate/search/store/FSMasterDirectoryProvider.java 2008-06-26 22:15:36 UTC (rev 14816) +++ search/trunk/src/java/org/hibernate/search/store/FSMasterDirectoryProvider.java 2008-06-27 00:25:45 UTC (rev 14817) @@ -169,7 +169,7 @@ long start = System.currentTimeMillis(); inProgress = true; if ( directoryProviderLock == null ) { - directoryProviderLock = searchFactory.getLockableDirectoryProviders().get( directoryProvider ); + directoryProviderLock = searchFactory.getDirectoryProviderLock( directoryProvider ); directoryProvider = null; searchFactory = null; //get rid of any useless link (help hot redeployment?) } Modified: search/trunk/src/test/org/hibernate/search/test/reader/performance/ReaderPerformance.java =================================================================== --- search/trunk/src/test/org/hibernate/search/test/reader/performance/ReaderPerformance.java 2008-06-26 22:15:36 UTC (rev 14816) +++ search/trunk/src/test/org/hibernate/search/test/reader/performance/ReaderPerformance.java 2008-06-27 00:25:45 UTC (rev 14817) @@ -29,13 +29,15 @@ private static final File baseIndexDir = new File( new File( "." ), "indextemp" ); //more iterations for more reliable measures: - private static final int TOTAL_WORK_BATCHES = 1000; + private static final int TOTAL_WORK_BATCHES = 10; //the next 3 define the kind of workload mix to test on: private static final int SEARCHERS_PER_BATCH = 10; private static final int UPDATES_PER_BATCH = 2; private static final int INSERTIONS_PER_BATCH = 1; - private static final int WORKER_THREADS = 30; + private static final int WORKER_THREADS = 20; + + private static final int WARMUP_CYCLES = 6; protected void setUp() throws Exception { baseIndexDir.mkdir(); @@ -44,8 +46,6 @@ FileHelper.delete( file ); } super.setUp(); - //enable this line: -// buildBigIndex(); } public void testFakeTest(){ @@ -53,15 +53,17 @@ } private void buildBigIndex() throws InterruptedException, CorruptIndexException, LockObtainFailedException, IOException { + System.out.println( "Going to create fake index..." ); FSDirectory directory = FSDirectory.getDirectory(new File(baseIndexDir, Detective.class.getCanonicalName())); IndexWriter iw = new IndexWriter( directory, new SimpleAnalyzer(), true ); IndexFillRunnable filler = new IndexFillRunnable( iw ); ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool( WORKER_THREADS ); - for (int batch=0; batch<=100000; batch++){ + for (int batch=0; batch<=5000000; batch++){ executor.execute( filler ); } executor.shutdown(); executor.awaitTermination( 600, TimeUnit.SECONDS ); + iw.optimize(); iw.close(); System.out.println( "Index created." ); } @@ -75,7 +77,7 @@ protected void tearDown() throws Exception { super.tearDown(); -// FileHelper.delete( baseIndexDir ); + FileHelper.delete( baseIndexDir ); } protected void configure(org.hibernate.cfg.Configuration cfg) { @@ -90,7 +92,14 @@ protected abstract String getReaderStrategyName(); //this test is disabled as it is very slow (and someone should read the output) - public final void disabled_testPerformance() throws InterruptedException{ + public final void disabled_testPerformance() throws InterruptedException, CorruptIndexException, LockObtainFailedException, IOException { + buildBigIndex(); + for (int i=0; i<WARMUP_CYCLES; i++) { + timeMs(); + } + } + + private final void timeMs() throws InterruptedException { ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool( WORKER_THREADS ); CountDownLatch startSignal = new CountDownLatch(1); InsertActivity insertionTask = new InsertActivity( getSessions(), startSignal );