boxtec
/
yacy_search_server
mirror of https://github.com/yacy/yacy_search_server
4
1
Fork 0
Code Issues Projects Releases Wiki Activity

more concurrency for postprocessing

Browse Source
pull/1/head
orbiter 11 years ago
parent a83cf26c38
commit 2bc6199408
1 changed files with 27 additions and 27 deletions
Show Stats Download Patch File Download Diff File

54
source/net/yacy/cora/federate/solr/connector/AbstractSolrConnector.java
Unescape View File

@ -163,38 +163,38 @@ public abstract class AbstractSolrConnector implements SolrConnector {
final String ... fields) { final String ... fields) {
assert buffersize > 0; assert buffersize > 0;
if (!prefetchIDs) return concurrentDocumentsByQueryNoPrefetch(querystring, sort, offset, maxcount, maxtime, buffersize, concurrency, fields); if (!prefetchIDs) return concurrentDocumentsByQueryNoPrefetch(querystring, sort, offset, maxcount, maxtime, buffersize, concurrency, fields);
final BlockingQueue<String> idQueue = concurrentIDsByQuery(querystring, sort, offset, maxcount, maxtime, Math.min(maxcount, 10000000), 1); final BlockingQueue<String> idQueue = concurrentIDsByQuery(querystring, sort, offset, maxcount, maxtime, Math.min(maxcount, 10000000), concurrency);
final BlockingQueue<SolrDocument> queue = buffersize <= 0 ? new LinkedBlockingQueue<SolrDocument>() : new ArrayBlockingQueue<SolrDocument>(buffersize); final BlockingQueue<SolrDocument> queue = buffersize <= 0 ? new LinkedBlockingQueue<SolrDocument>() : new ArrayBlockingQueue<SolrDocument>(Math.max(buffersize, concurrency));
final long endtime = maxtime == Long.MAX_VALUE ? Long.MAX_VALUE : System.currentTimeMillis() + maxtime; // we know infinity! final long endtime = maxtime == Long.MAX_VALUE ? Long.MAX_VALUE : System.currentTimeMillis() + maxtime; // we know infinity!
final Thread t = new Thread() { final Thread[] t = new Thread[concurrency];
@Override for (int i = 0; i < concurrency; i++) {
public void run() { t[i] = new Thread() {
this.setName("AbstractSolrConnector:concurrentDocumentsByQueryWithPrefetch(" + querystring + ")"); @Override
String nextID; public void run() {
try { this.setName("AbstractSolrConnector:concurrentDocumentsByQueryWithPrefetch(" + querystring + ")");
while (System.currentTimeMillis() < endtime && (nextID = idQueue.take()) != AbstractSolrConnector.POISON_ID) { String nextID;
try {
try { while (System.currentTimeMillis() < endtime && (nextID = idQueue.take()) != AbstractSolrConnector.POISON_ID) {
SolrDocument d = getDocumentById(nextID, fields); try {
// document may be null if another process has deleted the document meanwhile SolrDocument d = getDocumentById(nextID, fields);
// in case that the document is absent then, we silently ignore that case // document may be null if another process has deleted the document meanwhile
if (d != null) try {queue.put(d);} catch (final InterruptedException e) {} // in case that the document is absent then, we silently ignore that case
} catch (final SolrException | IOException e) { if (d != null) try {queue.put(d);} catch (final InterruptedException e) {}
ConcurrentLog.logException(e); } catch (final SolrException | IOException e) {
// fail ConcurrentLog.logException(e);
ConcurrentLog.severe("AbstractSolrConnector", "aborted concurrentDocumentsByQuery: " + e.getMessage()); // fail
break; ConcurrentLog.severe("AbstractSolrConnector", "aborted concurrentDocumentsByQuery: " + e.getMessage());
break;
}
} }
} catch (InterruptedException e) {
ConcurrentLog.severe("AbstractSolrConnector", "interrupted concurrentDocumentsByQuery: " + e.getMessage());
} }
} catch (InterruptedException e) {
ConcurrentLog.severe("AbstractSolrConnector", "interrupted concurrentDocumentsByQuery: " + e.getMessage());
}
for (int i = 0; i < Math.max(1, concurrency); i++) {
try {queue.put(AbstractSolrConnector.POISON_DOCUMENT);} catch (final InterruptedException e1) {} try {queue.put(AbstractSolrConnector.POISON_DOCUMENT);} catch (final InterruptedException e1) {}
} }
} };
}; t[i].start();
t.start(); }
return queue; return queue;
} }

Write Preview
Loading…
Cancel
Save