HSEARCH-1699 Batching IndexWriter commits for multiple synchronous worksets

Author: Gustavo Fernandes

engine/src/main/java/org/hibernate/search/backend/impl/lucene/BatchSyncProcessor.java

@@ -0,0 +1,153 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * Copyright (c) 2014, 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.backend.impl.lucene;
+
+import org.hibernate.search.backend.IndexingMonitor;
+import org.hibernate.search.backend.LuceneWork;
+import org.hibernate.search.util.logging.impl.Log;
+import org.hibernate.search.util.logging.impl.LoggerFactory;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.locks.LockSupport;
+
+/**
+ * Processes changesets in batches, maintaining sync guarantees.
+ *<p>
+ * Multiple threads produce one or more {@link org.hibernate.search.backend.LuceneWork}
+ * by calling {@link #submit(java.util.List, org.hibernate.search.backend.IndexingMonitor)},
+ * and get blocked until their changes are applied to the index;</p>
+ * The {@link org.hibernate.search.backend.impl.lucene.BatchSyncProcessor.Consumer} thread will
+ * coalesce changes from multiple threads and apply them in the index, releasing the waiting threads
+ * at the end.
+ * <p>
+ * In the absence of work to be applied, the Consumer thread is parked to avoid busy waiting.</p>
+ *
+ * @author gustavonalle
+ */
+public class BatchSyncProcessor {
+
+	private static final Log log = LoggerFactory.make();
+
+	private final BlockingQueue<Changeset> transferQueue = new LinkedBlockingDeque<Changeset>();
+	private volatile LuceneBackendResources resources;
+	private final String indexName;
+	private volatile boolean stop = false;
+	final Thread consumerThread;
+
+	/**
+	 * Constructor
+	 * @param resources LuceneResources to obtain the workspace
+	 * @param indexName for debugging purposes
+	 */
+	public BatchSyncProcessor(LuceneBackendResources resources, String indexName) {
+		this.resources = resources;
+		this.indexName = indexName;
+		consumerThread = new Thread( new Consumer(), "Hibernate Search sync consumer thread for index " + indexName );
+		consumerThread.setDaemon( true );
+	}
+
+	/**
+	 * Start processing
+	 */
+	public void start() {
+		log.startingSyncConsumerThread( indexName );
+		consumerThread.start();
+	}
+
+	/**
+	 * Submit work and wait for it to be applied to the index
+	 * @param workList list of work
+	 * @param monitor for statistics collection
+	 */
+	public void submit(List<LuceneWork> workList, IndexingMonitor monitor) {
+		Changeset changeset = new Changeset( workList, Thread.currentThread(), monitor );
+		transferQueue.add( changeset );
+		wakeUpConsumer();
+		boolean interrupted = false;
+		while ( ! changeset.isProcessed() && ! interrupted ) {
+			LockSupport.park();
+			if ( Thread.interrupted() ) {
+				interrupted = true;
+			}
+		}
+		if ( interrupted ) {
+			Thread.currentThread().interrupt();
+		}
+	}
+
+	/**
+	 * Wakes up consumer thread if necessary
+	 */
+	private void wakeUpConsumer() {
+		LockSupport.unpark( consumerThread );
+	}
+
+	/**
+	 * Dispose resources
+	 */
+	public void shutdown() {
+		stop = true;
+		LockSupport.unpark( consumerThread );
+	}
+
+	/**
+	 * Handle on the fly rebuilds
+	 * @param resources new instance of {@link org.hibernate.search.backend.impl.lucene.LuceneBackendResources}
+	 */
+	void updateResources(LuceneBackendResources resources) {
+		this.resources = resources;
+	}
+
+	/**
+	 * Consumer thread
+	 */
+	private class Consumer implements Runnable {
+		@Override
+		public void run() {
+			while ( ! stop ) {
+				while ( transferQueue.isEmpty() && ! stop ) {
+					// Avoid busy wait
+					LockSupport.park();
+				}
+				if ( ! transferQueue.isEmpty() ) {
+					List<Changeset> changesets = new LinkedList<Changeset>();
+					transferQueue.drainTo( changesets );
+					ChangesetList changesetList = new ChangesetList( changesets );
+					try {
+						LuceneBackendQueueTask luceneBackendQueueTask = new LuceneBackendQueueTask( changesetList.getWork(), resources, null );
+						luceneBackendQueueTask.run();
+					}
+					finally {
+						changesetList.markProcessed();
+					}
+				}
+			}
+			log.stoppingSyncConsumerThread( indexName );
+		}
+	}
+
+}

engine/src/main/java/org/hibernate/search/backend/impl/lucene/Changeset.java

@@ -0,0 +1,67 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * Copyright (c) 2014, 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.backend.impl.lucene;
+
+import org.hibernate.search.backend.IndexingMonitor;
+import org.hibernate.search.backend.LuceneWork;
+
+import java.util.List;
+import java.util.concurrent.locks.LockSupport;
+
+/**
+ * A Changeset is a work to be applied to the index and its associated producer
+ *
+ * @author gustavonalle
+ */
+final class Changeset {
+
+	private final List<LuceneWork> workList;
+	private final Thread producer;
+	private final IndexingMonitor monitor;
+	private volatile boolean processed = false;
+
+	Changeset(List<LuceneWork> workList, Thread producer, IndexingMonitor monitor) {
+		this.workList = workList;
+		this.producer = producer;
+		this.monitor = monitor;
+	}
+
+	List<LuceneWork> getWorkList() {
+		return workList;
+	}
+
+	IndexingMonitor getMonitor() {
+		return monitor;
+	}
+
+	boolean isProcessed() {
+		return processed;
+	}
+
+	public void markProcessed() {
+		processed = true;
+		LockSupport.unpark( producer );
+	}
+
+}

engine/src/main/java/org/hibernate/search/backend/impl/lucene/ChangesetList.java

@@ -0,0 +1,58 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * Copyright (c) 2014, 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.backend.impl.lucene;
+
+import org.hibernate.search.backend.LuceneWork;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Aggregator for {@link org.hibernate.search.backend.impl.lucene.Changeset}
+ *
+ * @author gustavonalle
+ */
+final class ChangesetList {
+
+	private final List<Changeset> changesets;
+
+	ChangesetList(List<Changeset> changesets) {
+		this.changesets = changesets;
+	}
+
+	List<LuceneWork> getWork() {
+		ArrayList<LuceneWork> luceneWorks = new ArrayList<LuceneWork>();
+		for ( Changeset changeset : changesets ) {
+			luceneWorks.addAll( changeset.getWorkList() );
+		}
+		return luceneWorks;
+	}
+
+	void markProcessed() {
+		for ( Changeset changeset : changesets ) {
+			changeset.markProcessed();
+		}
+	}
+
+}

engine/src/main/java/org/hibernate/search/backend/impl/lucene/LuceneBackendQueueProcessor.java

@@ -25,11 +25,8 @@
 
 import java.util.List;
 import java.util.Properties;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
 import java.util.concurrent.locks.Lock;
 
-import org.hibernate.search.SearchException;
 import org.hibernate.search.backend.BackendFactory;
 import org.hibernate.search.backend.IndexingMonitor;
 import org.hibernate.search.backend.LuceneWork;
@@ -57,6 +54,7 @@ public class LuceneBackendQueueProcessor implements BackendQueueProcessor {
 	private boolean sync;
 	private AbstractWorkspaceImpl workspaceOverride;
 	private LuceneBackendTaskStreamer streamWorker;
+	private BatchSyncProcessor batchSyncProcessor;
 
 	@Override
 	public void initialize(Properties props, WorkerBuildContext context, DirectoryBasedIndexManager indexManager) {
@@ -68,11 +66,14 @@ public void initialize(Properties props, WorkerBuildContext context, DirectoryBa
 		}
 		resources = new LuceneBackendResources( context, indexManager, props, workspaceOverride );
 		streamWorker = new LuceneBackendTaskStreamer( resources );
+		batchSyncProcessor = new BatchSyncProcessor( resources, indexManager.getIndexName() );
+		batchSyncProcessor.start();
 	}
 
 	@Override
 	public void close() {
 		resources.shutdown();
+		batchSyncProcessor.shutdown();
 	}
 
 	@Override
@@ -88,25 +89,15 @@ public void applyWork(List<LuceneWork> workList, IndexingMonitor monitor) {
 		if ( workList == null ) {
 			throw new IllegalArgumentException( "workList should not be null" );
 		}
-		LuceneBackendQueueTask luceneBackendQueueProcessor = new LuceneBackendQueueTask(
-				workList,
-				resources,
-				monitor
-		);
 		if ( sync ) {
-			Future<?> future = resources.getQueueingExecutor().submit( luceneBackendQueueProcessor );
-			try {
-				future.get();
-			}
-			catch (InterruptedException e) {
-				log.interruptedWhileWaitingForIndexActivity( e );
-				Thread.currentThread().interrupt();
-			}
-			catch (ExecutionException e) {
-				throw new SearchException( "Error applying updates to the Lucene index", e.getCause() );
-			}
+			batchSyncProcessor.submit( workList, monitor );
 		}
 		else {
+			LuceneBackendQueueTask luceneBackendQueueProcessor = new LuceneBackendQueueTask(
+					workList,
+					resources,
+					monitor
+			);
 			resources.getQueueingExecutor().execute( luceneBackendQueueProcessor );
 		}
 	}
@@ -133,6 +124,7 @@ public void setCustomWorkspace(AbstractWorkspaceImpl workspace) {
 	@Override
 	public void indexMappingChanged() {
 		resources = resources.onTheFlyRebuild();
+		batchSyncProcessor.updateResources( resources );
 	}
 
 }

engine/src/main/java/org/hibernate/search/util/logging/impl/Log.java

@@ -774,4 +774,11 @@ public interface Log extends BasicLogger {
 	@Message(id = 229, value = "Property " + Environment.INDEX_FLUSH_INTERVAL + "for the index '%s' needs to be positive." )
 	SearchException flushIntervalNeedsToBePositive(String indexName);
 
+	@LogMessage(level = INFO)
+	@Message(id = 230, value = "Starting sync consumer thread for index '%s'" )
+	void startingSyncConsumerThread(String indexName);
+
+	@LogMessage(level = INFO)
+	@Message(id = 231, value = "Stopping sync consumer thread for index '%s'" )
+	void stoppingSyncConsumerThread(String indexName);
 }