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
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#include "querylimiter.h"
#include <chrono>
namespace proton:: matching {
QueryLimiter::LimitedToken::LimitedToken(const Doom & doom, QueryLimiter & limiter) :
_limiter(limiter)
{
_limiter.grabToken(doom);
}
QueryLimiter::LimitedToken::~LimitedToken()
{
_limiter.releaseToken();
}
void
QueryLimiter::grabToken(const Doom & doom)
{
std::unique_lock<std::mutex> guard(_lock);
for (auto max_threads = get_max_threads(); (max_threads > 0) && (_activeThreads >= max_threads) && !doom.hard_doom(); max_threads = get_max_threads()) {
vespalib::duration left = doom.hard_left();
if (left > vespalib::duration::zero()) {
_cond.wait_for(guard, left);
}
}
_activeThreads++;
}
void
QueryLimiter::releaseToken()
{
std::lock_guard<std::mutex> guard(_lock);
_activeThreads--;
_cond.notify_one();
}
QueryLimiter::QueryLimiter() :
_lock(),
_cond(),
_activeThreads(0),
_maxThreads(-1),
_coverage(1.0),
_minHits(std::numeric_limits<uint32_t>::max())
{
}
void
QueryLimiter::configure(int maxThreads, double coverage, uint32_t minHits)
{
std::lock_guard<std::mutex> guard(_lock);
_maxThreads.store(maxThreads, std::memory_order_relaxed);
_coverage.store(coverage, std::memory_order_relaxed);
_minHits.store(minHits, std::memory_order_relaxed);
_cond.notify_all();
}
QueryLimiter::Token::UP
QueryLimiter::getToken(const Doom & doom, uint32_t numDocs, uint32_t numHits, bool hasSorting, bool hasGrouping)
{
if (get_max_threads() > 0) {
if (hasSorting || hasGrouping) {
if (numHits > get_min_hits()) {
if (numDocs * get_coverage() < numHits) {
return std::make_unique<LimitedToken>(doom, *this);
}
}
}
}
return std::make_unique<NoLimitToken>();
}
}
|
