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
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#include "threadpoolimpl.h"
#include "threadimpl.h"
#include <vespa/vespalib/util/exceptions.h>
#include <cassert>
#include <thread>
#include <vespa/log/log.h>
LOG_SETUP(".storageframework.thread_pool_impl");
using namespace std::chrono_literals;
using vespalib::IllegalStateException;
namespace storage::framework::defaultimplementation {
ThreadPoolImpl::ThreadPoolImpl(Clock& clock)
: _clock(clock),
_stopping(false)
{ }
ThreadPoolImpl::~ThreadPoolImpl()
{
{
std::lock_guard lock(_threadVectorLock);
_stopping = true;
for (ThreadImpl * thread : _threads) {
thread->interrupt();
}
for (ThreadImpl * thread : _threads) {
thread->join();
}
}
for (uint32_t i=0; true; i+=10) {
{
std::lock_guard lock(_threadVectorLock);
if (_threads.empty()) break;
}
if (i > 1000) {
fprintf(stderr, "Failed to kill thread pool. Threads won't die. (And if allowing thread pool object"
" to be deleted this will create a segfault later)\n");
LOG_ABORT("should not be reached");
}
std::this_thread::sleep_for(10ms);
}
}
Thread::UP
ThreadPoolImpl::startThread(Runnable& runnable, vespalib::stringref id, vespalib::duration waitTime,
vespalib::duration maxProcessTime, int ticksBeforeWait,
std::optional<vespalib::CpuUsage::Category> cpu_category)
{
std::lock_guard lock(_threadVectorLock);
assert(!_stopping);
auto thread = std::make_unique<ThreadImpl>(*this, runnable, id, waitTime, maxProcessTime, ticksBeforeWait, cpu_category);
_threads.push_back(thread.get());
return thread;
}
void
ThreadPoolImpl::visitThreads(ThreadVisitor& visitor) const
{
std::lock_guard lock(_threadVectorLock);
for (const ThreadImpl * thread : _threads) {
visitor.visitThread(*thread);
}
}
void
ThreadPoolImpl::unregisterThread(ThreadImpl& t)
{
std::lock_guard lock(_threadVectorLock);
std::vector<ThreadImpl*> threads;
threads.reserve(_threads.size());
for (ThreadImpl * thread : _threads) {
if (thread != &t) {
threads.push_back(thread);
}
}
_threads.swap(threads);
}
}
|
