litvinovg/vitro
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fixing problems with too many threads for indexing.

Browse source
This commit is contained in:
bdc34 2010-07-08 23:43:51 +00:00
parent 499511d5ef
commit 8f4b448dcb
12 changed files with 367 additions and 238 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
webapp/src/edu/cornell/mannlib/vitro/webapp/controller/edit/EntityRetryController.java
View file

@ -35,7 +35,6 @@ import edu.cornell.mannlib.vedit.forwarder.PageForwarder;
import edu.cornell.mannlib.vedit.forwarder.impl.UrlForwarder;
import edu.cornell.mannlib.vedit.util.FormUtils;
import edu.cornell.mannlib.vedit.validator.impl.RequiredFieldValidator;
import edu.cornell.mannlib.vitro.webapp.auth.policy.JenaNetidPolicy.ContextSetup;
import edu.cornell.mannlib.vitro.webapp.beans.DataProperty;
import edu.cornell.mannlib.vitro.webapp.beans.DataPropertyStatement;
import edu.cornell.mannlib.vitro.webapp.beans.Individual;
@ -52,7 +51,6 @@ import edu.cornell.mannlib.vitro.webapp.dao.VClassDao;
import edu.cornell.mannlib.vitro.webapp.dao.VClassGroupDao;
import edu.cornell.mannlib.vitro.webapp.dao.WebappDaoFactory;
import edu.cornell.mannlib.vitro.webapp.edit.listener.impl.IndividualDataPropertyStatementProcessor;
import edu.cornell.mannlib.vitro.webapp.edit.listener.impl.SearchReindexer;
public class EntityRetryController extends BaseEditController {

2
webapp/src/edu/cornell/mannlib/vitro/webapp/controller/edit/UpdateEntityFlagServlet.java
View file

@ -133,7 +133,7 @@ public class UpdateEntityFlagServlet extends VitroHttpServlet {
private void updateSearchIndex(HttpServletRequest request){
IndexBuilder builder = (IndexBuilder)getServletContext().getAttribute(IndexBuilder.class.getName());
if( builder != null )
(new Thread(builder)).start();
builder.doUpdateIndex();
}
}

20
webapp/src/edu/cornell/mannlib/vitro/webapp/dao/jena/SearchReindexingListener.java
View file

@ -24,14 +24,12 @@ import edu.cornell.mannlib.vitro.webapp.search.indexing.IndexBuilder;
* called because the inference system does not seem to send notifyEvents.
*/
public class SearchReindexingListener implements ModelChangedListener {
private HashSet<String> changedUris;
private IndexBuilder indexBuilder;
public SearchReindexingListener(IndexBuilder indexBuilder) {
if(indexBuilder == null )
throw new IllegalArgumentException("Constructor parameter indexBuilder must not be null");
this.indexBuilder = indexBuilder;
this.changedUris = new HashSet<String>();
}
private synchronized void addChange(Statement stmt){
@ -49,8 +47,8 @@ public class SearchReindexingListener implements ModelChangedListener {
}
}
private void doAsyncIndexBuild(){
new Thread(indexBuilder).start();
private void requestAsyncIndexUpdate(){
indexBuilder.doUpdateIndex();
}
@Override
@ -59,7 +57,7 @@ public class SearchReindexingListener implements ModelChangedListener {
EditEvent editEvent = (EditEvent)arg1;
if( !editEvent.getBegin() ){// editEvent is the end of an edit
log.debug("Doing search index build at end of EditEvent");
doAsyncIndexBuild();
requestAsyncIndexUpdate();
}
} else{
log.debug("ignoring event " + arg1.getClass().getName() + " "+ arg1 );
@ -69,13 +67,13 @@ public class SearchReindexingListener implements ModelChangedListener {
@Override
public void addedStatement(Statement stmt) {
addChange(stmt);
doAsyncIndexBuild();
requestAsyncIndexUpdate();
}
@Override
public void removedStatement(Statement stmt){
addChange(stmt);
doAsyncIndexBuild();
requestAsyncIndexUpdate();
}
private static final Log log = LogFactory.getLog(SearchReindexingListener.class.getName());
@ -85,7 +83,7 @@ public class SearchReindexingListener implements ModelChangedListener {
for( Statement s: arg0){
addChange(s);
}
doAsyncIndexBuild();
requestAsyncIndexUpdate();
}
@Override
@ -93,7 +91,7 @@ public class SearchReindexingListener implements ModelChangedListener {
for( Statement s: arg0){
addChange(s);
}
doAsyncIndexBuild();
requestAsyncIndexUpdate();
}
@Override
@ -106,7 +104,7 @@ public class SearchReindexingListener implements ModelChangedListener {
}finally{
arg0.close();
}
doAsyncIndexBuild();
requestAsyncIndexUpdate();
}
@Override
@ -122,7 +120,7 @@ public class SearchReindexingListener implements ModelChangedListener {
if( it != null ) it.close();
m.leaveCriticalSection();
}
doAsyncIndexBuild();
requestAsyncIndexUpdate();
}
@Override

2
webapp/src/edu/cornell/mannlib/vitro/webapp/edit/listener/impl/KeywordSearchReindexer.java
View file

@ -10,7 +10,7 @@ public class KeywordSearchReindexer implements ChangeListener {
public void doInserted(Object newObj, EditProcessObject epo){
IndexBuilder builder = (IndexBuilder)epo.getSession().getServletContext().getAttribute(IndexBuilder.class.getName());
(new Thread(builder)).start();
builder.doUpdateIndex();
}
public void doUpdated(Object oldObj, Object newObj, EditProcessObject epo){

26
webapp/src/edu/cornell/mannlib/vitro/webapp/edit/listener/impl/SearchReindexer.java
View file

@ -1,26 +0,0 @@
/* $This file is distributed under the terms of the license in /doc/license.txt$ */
package edu.cornell.mannlib.vitro.webapp.edit.listener.impl;
import edu.cornell.mannlib.vedit.beans.EditProcessObject;
import edu.cornell.mannlib.vedit.listener.ChangeListener;
import edu.cornell.mannlib.vitro.webapp.beans.Individual;
import edu.cornell.mannlib.vitro.webapp.search.indexing.IndexBuilder;
public class SearchReindexer implements ChangeListener {
public void doInserted(Object newObj, EditProcessObject epo){
IndexBuilder builder = (IndexBuilder)epo.getSession().getServletContext().getAttribute(IndexBuilder.class.getName());
(new Thread(builder)).start();
}
public void doUpdated(Object oldObj, Object newObj, EditProcessObject epo){
doInserted(newObj, epo);
}
public void doDeleted(Object oldObj, EditProcessObject epo){
IndexBuilder builder = (IndexBuilder)epo.getSession().getServletContext().getAttribute(IndexBuilder.class.getName());
builder.entityDeleted(((Individual)oldObj).getURI());
}
}

8
webapp/src/edu/cornell/mannlib/vitro/webapp/search/controller/IndexController.java
View file

@ -20,7 +20,7 @@ import edu.cornell.mannlib.vitro.webapp.search.IndexingException;
import edu.cornell.mannlib.vitro.webapp.search.indexing.IndexBuilder;
/**
* Acepts requests to rebuild or update the search index. It uses
* Accepts requests to rebuild or update the search index. It uses
* an IndexBuilder and finds that IndexBuilder from the servletContext using
* the key "edu.cornel.mannlib.vitro.search.indexing.IndexBuilder"
*
@ -61,12 +61,8 @@ public class IndexController extends HttpServlet {
IndexBuilder builder = (IndexBuilder)getServletContext().getAttribute(IndexBuilder.class.getName());
if( request.getParameter("update") != null ){
builder.doUpdateIndex();
}
if (request.getParameter("clear") != null ) {
builder.clearIndex();
}else{
builder.doIndexBuild();
builder.doIndexRebuild();
}
} catch (IndexingException e) {

218
webapp/src/edu/cornell/mannlib/vitro/webapp/search/indexing/IndexBuilder.java
View file

@ -2,7 +2,9 @@
package edu.cornell.mannlib.vitro.webapp.search.indexing;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
@ -28,8 +30,9 @@ import edu.cornell.mannlib.vitro.webapp.search.beans.ProhibitedFromSearch;
* It uses an implementation of a back-end through an object that
* implements IndexerIface. An example of a back-end is LuceneIndexer.
*
* The IndexBuilder implements the EntityChangeListener so it can
* be registered for Entity changes from the GenericDB classes.
* See the class SearchReindexingListener for an example of how a model change
* listener can use an IndexBuilder to keep the full text index in sncy with
* updates to a model.
*
* There should be an IndexBuilder in the servlet context, try:
*
@ -40,19 +43,28 @@ import edu.cornell.mannlib.vitro.webapp.search.beans.ProhibitedFromSearch;
* @author bdc34
*
*/
public class IndexBuilder implements Runnable {
List<ObjectSourceIface> sourceList = new LinkedList<ObjectSourceIface>();
IndexerIface indexer = null;
ServletContext context = null;
ProhibitedFromSearch classesProhibitedFromSearch = null;
public class IndexBuilder {
private List<ObjectSourceIface> sourceList = new LinkedList<ObjectSourceIface>();
private IndexerIface indexer = null;
private ServletContext context = null;
private ProhibitedFromSearch classesProhibitedFromSearch = null;
long lastRun = 0;
Collection<String> changedUris = null;
private long lastRun = 0;
private HashSet<String> changedUris = null;
private List<Individual> updatedInds = null;
private List<Individual> deletedInds = null;
private IndexBuilderThread indexingThread = null;
//shared with IndexBuilderThread
private boolean reindexRequested = false;
public static final boolean UPDATE_DOCS = false;
public static final boolean NEW_DOCS = true;
private static final Log log = LogFactory.getLog(IndexBuilder.class.getName());
private static final Log log = LogFactory.getLog(IndexBuilder.class);
public IndexBuilder(ServletContext context,
IndexerIface indexer,
@ -61,7 +73,9 @@ public class IndexBuilder implements Runnable {
this.sourceList = sources;
this.context = context;
changedUris = new HashSet<String>();
this.changedUris = new HashSet<String>();
this.indexingThread = new IndexBuilderThread(this);
this.indexingThread.start();
}
public void addObjectSource(ObjectSourceIface osi) {
@ -77,9 +91,69 @@ public class IndexBuilder implements Runnable {
return sourceList;
}
public void doIndexBuild() throws IndexingException {
log.debug(this.getClass().getName()
+ " performing doFullRebuildIndex()\n");
public void doIndexRebuild() throws IndexingException {
//set up full index rebuild
setReindexRequested( true );
//wake up indexing thread
synchronized (this.indexingThread) {
this.indexingThread.notifyAll();
}
}
/**
* This will re-index Individuals that changed because of modtime or because they
* were added with addChangedUris().
*/
public void doUpdateIndex() {
//wake up thread
synchronized (this.indexingThread) {
this.indexingThread.notifyAll();
}
}
public synchronized void addToChangedUris(String uri){
changedUris.add(uri);
}
public synchronized void addToChangedUris(Collection<String> uris){
changedUris.addAll(uris);
}
public synchronized boolean isReindexRequested() {
return reindexRequested;
}
public synchronized boolean isThereWorkToDo(){
return isReindexRequested() || ! changedUris.isEmpty() ;
}
public ProhibitedFromSearch getClassesProhibitedFromSearch() {
return classesProhibitedFromSearch;
}
public void setClassesProhibitedFromSearch(
ProhibitedFromSearch classesProhibitedFromSearch) {
this.classesProhibitedFromSearch = classesProhibitedFromSearch;
}
public void killIndexingThread() {
this.indexingThread.kill();
}
/* ******************** non-public methods ************************* */
private synchronized void setReindexRequested(boolean reindexRequested) {
this.reindexRequested = reindexRequested;
}
private synchronized Collection<String> getAndEmptyChangedUris(){
Collection<String> out = changedUris;
changedUris = new HashSet<String>();
return out;
}
protected void indexRebuild() throws IndexingException {
setReindexRequested(false);
log.debug("performing indexRebuild()");
Iterator<ObjectSourceIface> sources = sourceList.iterator();
List listOfIterators = new LinkedList();
@ -98,15 +172,13 @@ public class IndexBuilder implements Runnable {
getAndEmptyChangedUris();
if( listOfIterators.size() == 0){ log.debug("Warning: no ObjectSources found.");}
doBuild( listOfIterators, true, NEW_DOCS );
doBuild( listOfIterators, Collections.EMPTY_LIST, true, NEW_DOCS );
log.debug(this.getClass().getName() + ".doFullRebuildIndex() Done \n");
}
public void run() {
doUpdateIndex();
}
public void doUpdateIndex() {
protected void updatedIndex() throws IndexingException{
log.debug("Starting updateIndex()");
long since = indexer.getModified() - 60000;
Iterator<ObjectSourceIface> sources = sourceList.iterator();
@ -125,16 +197,43 @@ public class IndexBuilder implements Runnable {
+ "\tIt doesn not implement " + "ObjectSourceIface.\n");
}
List<Individual> changedInds = addDepResourceClasses(checkForDeletes(getAndEmptyChangedUris()));
listOfIterators.add( (new IndexBuilder.BuilderObjectSource(changedInds)).getUpdatedSinceIterator(0) );
buildAddAndDeleteLists( getAndEmptyChangedUris());
listOfIterators.add( (new IndexBuilder.BuilderObjectSource(updatedInds)).getUpdatedSinceIterator(0) );
doBuild( listOfIterators, false, UPDATE_DOCS );
doBuild( listOfIterators, deletedInds, false, UPDATE_DOCS );
}
/**
* Sets updatedUris and deletedUris.
* @param changedUris
*/
private void buildAddAndDeleteLists( Collection<String> uris){
/* clear updateInds and deletedUris. This is the only method that should set these. */
this.updatedInds = new ArrayList<Individual>();
this.deletedInds = new ArrayList<Individual>();
WebappDaoFactory wdf = (WebappDaoFactory)context.getAttribute("webappDaoFactory");
for( String uri: uris){
if( uri != null ){
Individual ind = wdf.getIndividualDao().getIndividualByURI(uri);
if( ind != null)
this.updatedInds.add(ind);
else{
log.debug("found delete in changed uris");
this.deletedInds.add(ind);
}
}
}
this.updatedInds = addDepResourceClasses(updatedInds);
}
private List<Individual> addDepResourceClasses(List<Individual> inds) {
WebappDaoFactory wdf = (WebappDaoFactory)context.getAttribute("webappDaoFactory");
VClassDao vClassDao = wdf.getVClassDao();
java.util.ListIterator<Individual> it = inds.listIterator();
Iterator<Individual> it = inds.iterator();
VClass depResVClass = new VClass(VitroVocabulary.DEPENDENT_RESORUCE);
while(it.hasNext()){
Individual ind = it.next();
@ -167,14 +266,6 @@ public class IndexBuilder implements Runnable {
return inds;
}
public void clearIndex(){
try {
indexer.clearIndex();
} catch (IndexingException e) {
log.error("error while clearing index", e);
}
}
/**
* For each sourceIterator, get all of the objects and attempt to
* index them.
@ -189,12 +280,17 @@ public class IndexBuilder implements Runnable {
* to false, and a check is made before adding, it will work fine; but
* checking if an object is on the index is slow.
*/
private void doBuild(List sourceIterators, boolean wipeIndexFirst, boolean newDocs ){
private void doBuild(List sourceIterators, Collection<Individual> deletes, boolean wipeIndexFirst, boolean newDocs ){
try {
indexer.startIndexing();
if( wipeIndexFirst )
indexer.clearIndex();
else{
for(Individual deleteMe : deletes ){
indexer.removeFromIndex(deleteMe);
}
}
//get an iterator for all of the sources of indexable objects
Iterator sourceIters = sourceIterators.iterator();
@ -224,36 +320,19 @@ public class IndexBuilder implements Runnable {
* @param items
* @return
*/
protected void indexForSource(Iterator<Individual> individuals , boolean newDocs){
private void indexForSource(Iterator<Individual> individuals , boolean newDocs){
if( individuals == null ) return;
while(individuals.hasNext()){
indexItem(individuals.next(), newDocs);
}
}
private List<Individual> checkForDeletes(List<String> uris){
WebappDaoFactory wdf = (WebappDaoFactory)context.getAttribute("webappDaoFactory");
List<Individual> nonDeletes = new LinkedList<Individual>();
for( String uri: uris){
if( uri != null ){
Individual ind = wdf.getIndividualDao().getIndividualByURI(uri);
if( ind != null)
nonDeletes.add(ind);
else{
log.debug("found delete in changed uris");
entityDeleted(uri);
}
}
}
return nonDeletes;
}
/**
* Use the backend indexer to index a single item.
* @param item
* @return
*/
protected void indexItem( Individual ind, boolean newDoc){
private void indexItem( Individual ind, boolean newDoc){
try{
if( ind == null )
return;
@ -277,32 +356,6 @@ public class IndexBuilder implements Runnable {
return ;
}
public void entityDeleted(String entityURI) {
if( log.isDebugEnabled())
log.debug("IndexBuilder.entityDeleted() " + entityURI);
Individual ent = new IndividualImpl(entityURI);
try {
indexer.removeFromIndex(ent);
} catch (IndexingException e) {
log.debug("IndexBuilder.entityDeleted failed: " + e);
}
}
public synchronized void addToChangedUris(String uri){
changedUris.add(uri);
}
public synchronized void addToChangedUris(Collection<String> uris){
changedUris.addAll(uris);
}
private synchronized List<String> getAndEmptyChangedUris(){
LinkedList<String> out = new LinkedList<String>();
out.addAll( changedUris );
changedUris = new HashSet<String>();
return out;
}
private class BuilderObjectSource implements ObjectSourceIface {
private final List<Individual> individuals;
public BuilderObjectSource( List<Individual> individuals){
@ -330,12 +383,7 @@ public class IndexBuilder implements Runnable {
}
}
public ProhibitedFromSearch getClassesProhibitedFromSearch() {
return classesProhibitedFromSearch;
}
public void setClassesProhibitedFromSearch(
ProhibitedFromSearch classesProhibitedFromSearch) {
this.classesProhibitedFromSearch = classesProhibitedFromSearch;
}
}

65
webapp/src/edu/cornell/mannlib/vitro/webapp/search/indexing/IndexBuilderThread.java Normal file
View file

@ -0,0 +1,65 @@
/* $This file is distributed under the terms of the license in /doc/license.txt$ */
package edu.cornell.mannlib.vitro.webapp.search.indexing;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* Thread that executes the methods in IndexBuilder.
*
* @author bdc34
*
*/
public class IndexBuilderThread extends Thread{
private IndexBuilder indexBuilder;
protected boolean stopRequested = false;
protected long reindexInterval = 1000 * 60 /* msec */ ;
private static final Log log = LogFactory.getLog(IndexBuilderThread.class.getName());
public IndexBuilderThread(IndexBuilder ib){
super("IndexBuilderThread");
this.indexBuilder = ib;
}
@Override
public void run() {
while(true){
if( stopRequested ){
log.info("Stopping IndexBuilderThread ");
return;
}
try{
if( indexBuilder.isReindexRequested() ){
log.debug("full re-index requested");
indexBuilder.indexRebuild();
}else{
log.debug("updated requested");
Thread.sleep(250);
indexBuilder.updatedIndex();
}
}catch (Throwable e) {
log.error(e,e);
}
if( indexBuilder != null && ! indexBuilder.isThereWorkToDo() ){
log.debug("there is no indexing working to do, going to sleep");
try {
synchronized (this) {
this.wait(reindexInterval);
}
} catch (InterruptedException e) {
log.debug(" woken up",e);
}
}
}
}
public synchronized void kill(){
log.debug("Attempting to kill IndexBuilderThread ");
stopRequested = true;
this.notifyAll();
}
}

12
webapp/src/edu/cornell/mannlib/vitro/webapp/search/lucene/LuceneIndexer.java
View file

@ -5,6 +5,7 @@ package edu.cornell.mannlib.vitro.webapp.search.lucene;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
@ -40,6 +41,7 @@ public class LuceneIndexer implements IndexerIface {
List<Searcher> searchers = null;
IndexWriter writer = null;
boolean indexing = false;
HashSet<String> urisIndexed;
//JODA timedate library can use java date format strings.
//http://java.sun.com/j2se/1.3/docs/api/java/text/SimpleDateFormat.html
@ -136,6 +138,7 @@ public class LuceneIndexer implements IndexerIface {
writer =
new IndexWriter(indexDir,analyzer,false, MAX_FIELD_LENGTH);
indexing = true;
urisIndexed = new HashSet<String>();
} catch(Throwable ioe){
try{
makeNewIndex();
@ -155,6 +158,7 @@ public class LuceneIndexer implements IndexerIface {
return;
}
try {
urisIndexed = null;
log.info("ending index");
if( writer != null )
writer.optimize();
@ -189,7 +193,15 @@ public class LuceneIndexer implements IndexerIface {
if( writer == null )
throw new IndexingException("LuceneIndexer: cannot build index," +
"IndexWriter is null.");
if( ind == null )
log.debug("Individual to index was null, ignoring.");
try {
if( urisIndexed.contains(ind.getURI()) ){
log.debug("already indexed " + ind.getURI() );
return;
}else
urisIndexed.add(ind.getURI());
Iterator<Obj2DocIface> it = getObj2DocList().iterator();
while (it.hasNext()) {
Obj2DocIface obj2doc = (Obj2DocIface) it.next();

2
webapp/src/edu/cornell/mannlib/vitro/webapp/search/lucene/LuceneSetup.java
View file

@ -145,6 +145,8 @@ public class LuceneSetup implements javax.servlet.ServletContextListener {
*/
public void contextDestroyed(ServletContextEvent sce) {
log.info("**** Running "+this.getClass().getName()+".contextDestroyed()");
IndexBuilder builder = (IndexBuilder)sce.getServletContext().getAttribute(IndexBuilder.class.getName());
builder.killIndexingThread();
}
/**

3
webapp/src/edu/cornell/mannlib/vitro/webapp/search/lucene/LuceneSetupCJK.java
View file

@ -127,7 +127,10 @@ public class LuceneSetupCJK implements javax.servlet.ServletContextListener {
* Gets run when the webApp Context gets destroyed.
*/
public void contextDestroyed(ServletContextEvent sce) {
log.info("**** Running "+this.getClass().getName()+".contextDestroyed()");
IndexBuilder builder = (IndexBuilder)sce.getServletContext().getAttribute(IndexBuilder.class.getName());
builder.killIndexingThread();
}
/**

33
webapp/test/edu/cornell/mannlib/vitro/webapp/search/indexing/IndexBuilderThreadTest.java Normal file
View file

@ -0,0 +1,33 @@
/* $This file is distributed under the terms of the license in /doc/license.txt$ */
package edu.cornell.mannlib.vitro.webapp.search.indexing;
import junit.framework.Assert;
import org.apache.log4j.Level;
import org.junit.Test;
import edu.cornell.mannlib.vitro.testing.AbstractTestClass;
public class IndexBuilderThreadTest extends AbstractTestClass {
@Test
public void testStoppingTheThread(){
setLoggerLevel(IndexBuilderThread.class, Level.OFF);
IndexBuilderThread ibt = new IndexBuilderThread(null);
ibt.start();
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Assert.fail(e.getMessage());
}
ibt.kill();
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Assert.fail(e.getMessage());
}
Assert.assertFalse(ibt.isAlive());
}
}