1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
|
/////////////////////////////////////////////////////////////////////////////
// Copyright (c) 2009-2014 Alan Wright. All rights reserved.
// Distributable under the terms of either the Apache License (Version 2.0)
// or the GNU Lesser General Public License.
/////////////////////////////////////////////////////////////////////////////
#include "LuceneInc.h"
#include <boost/version.hpp>
#if BOOST_VERSION >= 107300 // Boost 1.73.0+
#include <boost/bind/bind.hpp>
#else
#include <boost/bind.hpp>
#endif
#include <boost/bind/protect.hpp>
#include "ParallelMultiSearcher.h"
#include "_MultiSearcher.h"
#include "HitQueue.h"
#include "FieldDocSortedHitQueue.h"
#include "FieldDoc.h"
#include "TopFieldDocs.h"
#include "ThreadPool.h"
namespace Lucene {
ParallelMultiSearcher::ParallelMultiSearcher(Collection<SearchablePtr> searchables) : MultiSearcher(searchables) {
}
ParallelMultiSearcher::~ParallelMultiSearcher() {
}
int32_t ParallelMultiSearcher::docFreq(const TermPtr& term) {
ThreadPoolPtr threadPool(ThreadPool::getInstance());
Collection<FuturePtr> searchThreads(Collection<FuturePtr>::newInstance(searchables.size()));
for (int32_t i = 0; i < searchables.size(); ++i) {
searchThreads[i] = threadPool->scheduleTask(boost::protect(boost::bind<int32_t>(boost::mem_fn(&Searchable::docFreq), searchables[i], term)));
}
int32_t docFreq = 0;
for (int32_t i = 0; i < searchThreads.size(); ++i) {
docFreq += searchThreads[i]->get<int32_t>();
}
return docFreq;
}
TopDocsPtr ParallelMultiSearcher::search(const WeightPtr& weight, const FilterPtr& filter, int32_t n) {
HitQueuePtr hq(newLucene<HitQueue>(n, false));
SynchronizePtr lock(newInstance<Synchronize>());
ThreadPoolPtr threadPool(ThreadPool::getInstance());
Collection<FuturePtr> searchThreads(Collection<FuturePtr>::newInstance(searchables.size()));
Collection<MultiSearcherCallableNoSortPtr> multiSearcher(Collection<MultiSearcherCallableNoSortPtr>::newInstance(searchables.size()));
for (int32_t i = 0; i < searchables.size(); ++i) { // search each searchable
multiSearcher[i] = newLucene<MultiSearcherCallableNoSort>(lock, searchables[i], weight, filter, n, hq, i, starts);
searchThreads[i] = threadPool->scheduleTask(boost::protect(boost::bind<TopDocsPtr>(boost::mem_fn(&MultiSearcherCallableNoSort::call), multiSearcher[i])));
}
int32_t totalHits = 0;
double maxScore = -std::numeric_limits<double>::infinity();
for (int32_t i = 0; i < searchThreads.size(); ++i) {
TopDocsPtr topDocs(searchThreads[i]->get<TopDocsPtr>());
totalHits += topDocs->totalHits;
maxScore = std::max(maxScore, topDocs->maxScore);
}
Collection<ScoreDocPtr> scoreDocs(Collection<ScoreDocPtr>::newInstance(hq->size()));
for (int32_t i = hq->size() - 1; i >= 0; --i) { // put docs in array
scoreDocs[i] = hq->pop();
}
return newLucene<TopDocs>(totalHits, scoreDocs, maxScore);
}
TopFieldDocsPtr ParallelMultiSearcher::search(const WeightPtr& weight, const FilterPtr& filter, int32_t n, const SortPtr& sort) {
if (!sort) {
boost::throw_exception(NullPointerException(L"sort must not be null"));
}
FieldDocSortedHitQueuePtr hq(newLucene<FieldDocSortedHitQueue>(n));
SynchronizePtr lock(newInstance<Synchronize>());
ThreadPoolPtr threadPool(ThreadPool::getInstance());
Collection<FuturePtr> searchThreads(Collection<FuturePtr>::newInstance(searchables.size()));
Collection<MultiSearcherCallableWithSortPtr> multiSearcher(Collection<MultiSearcherCallableWithSortPtr>::newInstance(searchables.size()));
for (int32_t i = 0; i < searchables.size(); ++i) { // search each searchable
multiSearcher[i] = newLucene<MultiSearcherCallableWithSort>(lock, searchables[i], weight, filter, n, hq, sort, i, starts);
searchThreads[i] = threadPool->scheduleTask(boost::protect(boost::bind<TopFieldDocsPtr>(boost::mem_fn(&MultiSearcherCallableWithSort::call), multiSearcher[i])));
}
int32_t totalHits = 0;
double maxScore = -std::numeric_limits<double>::infinity();
for (int32_t i = 0; i < searchThreads.size(); ++i) {
TopFieldDocsPtr topDocs(searchThreads[i]->get<TopFieldDocsPtr>());
totalHits += topDocs->totalHits;
maxScore = std::max(maxScore, topDocs->maxScore);
}
Collection<ScoreDocPtr> scoreDocs(Collection<ScoreDocPtr>::newInstance(hq->size()));
for (int32_t i = hq->size() - 1; i >= 0; --i) { // put docs in array
scoreDocs[i] = hq->pop();
}
return newLucene<TopFieldDocs>(totalHits, scoreDocs, hq->getFields(), maxScore);
}
}
|
