Author: sannegrinovero Date: 2010-05-02 10:20:53 -0400 (Sun, 02 May 2010) New Revision: 19341 Added: search/trunk/hibernate-search/src/test/java/org/hibernate/search/test/backend/ search/trunk/hibernate-search/src/test/java/org/hibernate/search/test/backend/AsyncBackendLongWorklistsStressTest.java search/trunk/hibernate-search/src/test/java/org/hibernate/search/test/backend/SyncBackendLongWorklistsStressTest.java Modified: search/trunk/hibernate-search/src/main/java/org/hibernate/search/batchindexing/Executors.java search/trunk/hibernate-search/src/test/resources/log4j.properties Log: HSEARCH-515 Lucene AlreadyClosedExceptions During Batch Processing With ASYNC Workers Modified: search/trunk/hibernate-search/src/main/java/org/hibernate/search/batchindexing/Executors.java =================================================================== --- search/trunk/hibernate-search/src/main/java/org/hibernate/search/batchindexing/Executors.java 2010-05-02 10:47:19 UTC (rev 19340) +++ search/trunk/hibernate-search/src/main/java/org/hibernate/search/batchindexing/Executors.java 2010-05-02 14:20:53 UTC (rev 19341) @@ -25,11 +25,15 @@ package org.hibernate.search.batchindexing; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import org.hibernate.search.util.LoggerFactory; +import org.slf4j.Logger; + /** * Helper class to create threads; * these threads are grouped and named to be identified in a profiler. @@ -41,6 +45,8 @@ private static final String THREAD_GROUP_PREFIX = "Hibernate Search: "; private static final int QUEUE_MAX_LENGTH = 1000; //TODO have it configurable? + private static final Logger log = LoggerFactory.make(); + /** * Creates a new fixed size ThreadPoolExecutor. * It's using a blockingqueue of maximum 1000 elements and the rejection @@ -70,7 +76,7 @@ 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>( queueSize ), new SearchThreadFactory( groupname ), - new ThreadPoolExecutor.CallerRunsPolicy() ); + new BlockPolicy() ); } /** @@ -97,5 +103,32 @@ } } + + /** + * A handler for rejected tasks that will have the caller block until + * space is available. + */ + public static class BlockPolicy implements RejectedExecutionHandler { + + /** + * Creates a <tt>BlockPolicy</tt>. + */ + public BlockPolicy() { } + + /** + * Puts the Runnable to the blocking queue, effectively blocking + * the delegating thread until space is available. + * @param r the runnable task requested to be executed + * @param e the executor attempting to execute this task + */ + public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { + try { + e.getQueue().put( r ); + } + catch (InterruptedException e1) { + log.error( "Work discarded, thread was interrupted while waiting for space to schedule: {}", r ); + } + } + } } Added: search/trunk/hibernate-search/src/test/java/org/hibernate/search/test/backend/AsyncBackendLongWorklistsStressTest.java =================================================================== --- search/trunk/hibernate-search/src/test/java/org/hibernate/search/test/backend/AsyncBackendLongWorklistsStressTest.java (rev 0) +++ search/trunk/hibernate-search/src/test/java/org/hibernate/search/test/backend/AsyncBackendLongWorklistsStressTest.java 2010-05-02 14:20:53 UTC (rev 19341) @@ -0,0 +1,36 @@ +/* $Id$ + * + * Hibernate, Relational Persistence for Idiomatic Java + * + * Copyright (c) 2009, Red Hat, Inc. and/or its affiliates or third-party contributors as + * indicated by the @author tags or express copyright attribution + * statements applied by the authors. All third-party contributions are + * distributed under license by Red Hat, Inc. + * + * This copyrighted material is made available to anyone wishing to use, modify, + * copy, or redistribute it subject to the terms and conditions of the GNU + * Lesser General Public License, as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License + * for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this distribution; if not, write to: + * Free Software Foundation, Inc. + * 51 Franklin Street, Fifth Floor + * Boston, MA 02110-1301 USA + */ +package org.hibernate.search.test.backend; + +import org.hibernate.search.Environment; + +public class AsyncBackendLongWorklistsStressTest extends SyncBackendLongWorklistsStressTest { + + protected void configure(org.hibernate.cfg.Configuration cfg) { + super.configure( cfg ); + cfg.setProperty( Environment.WORKER_EXECUTION, "async" ); + } + +} Added: search/trunk/hibernate-search/src/test/java/org/hibernate/search/test/backend/SyncBackendLongWorklistsStressTest.java =================================================================== --- search/trunk/hibernate-search/src/test/java/org/hibernate/search/test/backend/SyncBackendLongWorklistsStressTest.java (rev 0) +++ search/trunk/hibernate-search/src/test/java/org/hibernate/search/test/backend/SyncBackendLongWorklistsStressTest.java 2010-05-02 14:20:53 UTC (rev 19341) @@ -0,0 +1,100 @@ +/* $Id$ + * + * Hibernate, Relational Persistence for Idiomatic Java + * + * Copyright (c) 2009, Red Hat, Inc. and/or its affiliates or third-party contributors as + * indicated by the @author tags or express copyright attribution + * statements applied by the authors. All third-party contributions are + * distributed under license by Red Hat, Inc. + * + * This copyrighted material is made available to anyone wishing to use, modify, + * copy, or redistribute it subject to the terms and conditions of the GNU + * Lesser General Public License, as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License + * for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this distribution; if not, write to: + * Free Software Foundation, Inc. + * 51 Franklin Street, Fifth Floor + * Boston, MA 02110-1301 USA + */ +package org.hibernate.search.test.backend; + +import java.io.File; + +import junit.framework.Assert; + +import org.apache.lucene.analysis.StopAnalyzer; +import org.apache.lucene.search.MatchAllDocsQuery; +import org.hibernate.Transaction; +import org.hibernate.criterion.Projections; +import org.hibernate.search.Environment; +import org.hibernate.search.FullTextSession; +import org.hibernate.search.Search; +import org.hibernate.search.store.FSDirectoryProvider; +import org.hibernate.search.test.Clock; +import org.hibernate.search.test.SearchTestCase; +import org.hibernate.search.util.FileHelper; + +public class SyncBackendLongWorklistsStressTest extends SearchTestCase { + + /* needs to be sensibly higher than org.hibernate.search.batchindexing.Executors.QUEUE_MAX_LENGTH */ + private static final int NUM_SAVED_ENTITIES = 3000; + + public void testWorkLongerThanMaxQueueSize() throws Exception { + FullTextSession s = Search.getFullTextSession( openSession() ); + for (int i = 0; i < NUM_SAVED_ENTITIES; i++ ) { + Transaction tx = s.beginTransaction(); + Clock clock = new Clock( i, "brand num° " + i); + s.persist( clock ); + tx.commit(); + s.clear(); + } + + Transaction tx = s.beginTransaction(); + // count of entities in database needs to be checked before SF is closed (HSQLDB will forget the entities) + Number count = (Number) s.createCriteria( Clock.class ).setProjection( Projections.rowCount() ).uniqueResult(); + Assert.assertEquals( NUM_SAVED_ENTITIES, count.intValue() ); + tx.commit(); + s.close(); + + //we need to close the session to wait for all async work to be flushed + sessions.close(); + buildSessionFactory( getMappings(), getAnnotatedPackages(), getXmlFiles() ); + + s = Search.getFullTextSession( openSession() ); + tx = s.beginTransaction(); + int fullTextCount = s.createFullTextQuery( new MatchAllDocsQuery(), Clock.class ).getResultSize(); + Assert.assertEquals( NUM_SAVED_ENTITIES, fullTextCount ); + tx.commit(); + s.close(); + } + + @Override + protected Class<?>[] getMappings() { + return new Class[] { Clock.class }; + } + + protected void configure(org.hibernate.cfg.Configuration cfg) { + super.configure( cfg ); + File sub = getBaseIndexDir(); + cfg.setProperty( "hibernate.search.default.indexBase", sub.getAbsolutePath() ); + //needs FSDirectory to have the index contents survive the SessionFactory close + cfg.setProperty( "hibernate.search.default.directory_provider", FSDirectoryProvider.class.getName() ); + cfg.setProperty( Environment.ANALYZER_CLASS, StopAnalyzer.class.getName() ); + cfg.setProperty( "hibernate.show_sql", "false" ); + cfg.setProperty( "hibernate.format_sql", "false" ); + } + + protected void tearDown() throws Exception { + super.tearDown(); + File sub = getBaseIndexDir(); + FileHelper.delete( sub ); + } + +} + Modified: search/trunk/hibernate-search/src/test/resources/log4j.properties =================================================================== --- search/trunk/hibernate-search/src/test/resources/log4j.properties 2010-05-02 10:47:19 UTC (rev 19340) +++ search/trunk/hibernate-search/src/test/resources/log4j.properties 2010-05-02 14:20:53 UTC (rev 19341) @@ -2,13 +2,13 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.Target=System.out log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %c{1}:%L - %m%n +log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} (%t) %5p %c{1}:%L - %m%n ### direct messages to file hibernate.log ### log4j.appender.file=org.apache.log4j.FileAppender log4j.appender.file.File=hibernate.log log4j.appender.file.layout=org.apache.log4j.PatternLayout -log4j.appender.file.layout.ConversionPattern=%d{ABSOLUTE} %5p %c{1}:%L - %m%n +log4j.appender.file.layout.ConversionPattern=%d{ABSOLUTE} (%t) %5p %c{1}:%L - %m%n ### direct messages to socket - chainsaw ### log4j.appender.socket=org.apache.log4j.net.SocketAppender