Apache
/
lucene
mirror of https://github.com/apache/lucene.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

SOLR-1711: fix hang when queue is full but there are no runners 

git-svn-id: https://svn.apache.org/repos/asf/lucene/dev/trunk@1063869 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Yonik Seeley 2011-01-26 20:57:05 +00:00
parent d494562133
commit 326ab7d577
2 changed files with 21 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
solr/CHANGES.txt
View File

@ -516,7 +516,8 @@ Bug Fixes
* SOLR-1711: SolrJ - StreamingUpdateSolrServer had a race condition that * SOLR-1711: SolrJ - StreamingUpdateSolrServer had a race condition that
could halt the streaming of documents. The original patch to fix this could halt the streaming of documents. The original patch to fix this
(never officially released) introduced another hanging bug due to (never officially released) introduced another hanging bug due to
connections not being released. (Attila Babo, Erik Hetzner via yonik) connections not being released.
(Attila Babo, Erik Hetzner, Johannes Tuchscherer via yonik)
* SOLR-1748, SOLR-1747, SOLR-1746, SOLR-1745, SOLR-1744: Streams and Readers * SOLR-1748, SOLR-1747, SOLR-1746, SOLR-1745, SOLR-1744: Streams and Readers
retrieved from ContentStreams are not closed in various places, resulting retrieved from ContentStreams are not closed in various places, resulting

30
solr/src/solrj/org/apache/solr/client/solrj/impl/StreamingUpdateSolrServer.java
View File

@ -173,12 +173,20 @@ public class StreamingUpdateSolrServer extends CommonsHttpSolrServer
} }
catch (Throwable e) { catch (Throwable e) {
handleError( e ); handleError( e );
} }
finally { finally {
// remove it from the list of running things...
// remove it from the list of running things unless we are the last runner and the queue is full...
// in which case, the next queue.put() would block and there would be no runners to handle it.
synchronized (runners) { synchronized (runners) {
runners.remove( this ); if (runners.size() == 1 && queue.remainingCapacity() == 0) {
// keep this runner alive
scheduler.execute(this);
} else {
runners.remove( this );
}
} }
log.info( "finished: {}" , this ); log.info( "finished: {}" , this );
runnerLock.unlock(); runnerLock.unlock();
} }
@ -208,7 +216,7 @@ public class StreamingUpdateSolrServer extends CommonsHttpSolrServer
return super.request( request ); return super.request( request );
} }
} }
try { try {
CountDownLatch tmpLock = lock; CountDownLatch tmpLock = lock;
if( tmpLock != null ) { if( tmpLock != null ) {
@ -216,18 +224,18 @@ public class StreamingUpdateSolrServer extends CommonsHttpSolrServer
} }
queue.put( req ); queue.put( req );
synchronized( runners ) { synchronized( runners ) {
if( runners.isEmpty() if( runners.isEmpty()
|| (queue.remainingCapacity() < queue.size() || (queue.remainingCapacity() < queue.size()
&& runners.size() < threadCount) ) && runners.size() < threadCount) )
{ {
Runner r = new Runner(); Runner r = new Runner();
scheduler.execute( r ); scheduler.execute( r );
runners.add( r ); runners.add( r );
} }
} }
} }
catch (InterruptedException e) { catch (InterruptedException e) {
log.error( "interrupted", e ); log.error( "interrupted", e );
throw new IOException( e.getLocalizedMessage() ); throw new IOException( e.getLocalizedMessage() );