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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#include "executorthreadingservice.h"
#include "threading_service_config.h"
#include <vespa/searchcore/proton/metrics/executor_threading_service_stats.h>
#include <vespa/vespalib/util/blockingthreadstackexecutor.h>
#include <vespa/vespalib/util/cpu_usage.h>
#include <vespa/vespalib/util/sequencedtaskexecutor.h>
#include <vespa/vespalib/util/singleexecutor.h>
using vespalib::BlockingThreadStackExecutor;
using vespalib::ThreadStackExecutor;
using vespalib::CpuUsage;
using vespalib::SequencedTaskExecutor;
using vespalib::SingleExecutor;
using vespalib::SyncableThreadExecutor;
using vespalib::steady_time;
using OptimizeFor = vespalib::Executor::OptimizeFor;
namespace proton {
namespace {
std::unique_ptr<SyncableThreadExecutor>
createExecutorWithOneThread(const ThreadingServiceConfig & cfg, vespalib::Runnable::init_fun_t init_function) {
uint32_t taskLimit = cfg.defaultTaskLimit();
if (cfg.optimize() == OptimizeFor::THROUGHPUT) {
uint32_t watermark = (cfg.kindOfwatermark() == 0) ? taskLimit / 10 : cfg.kindOfwatermark();
return std::make_unique<SingleExecutor>(std::move(init_function), taskLimit, cfg.is_task_limit_hard(), watermark, 100ms);
} else {
if (cfg.is_task_limit_hard()) {
return std::make_unique<BlockingThreadStackExecutor>(1, taskLimit, std::move(init_function));
} else {
return std::make_unique<ThreadStackExecutor>(1, std::move(init_function));
}
}
}
VESPA_THREAD_STACK_TAG(master_executor)
VESPA_THREAD_STACK_TAG(index_executor)
VESPA_THREAD_STACK_TAG(summary_executor)
}
ExecutorThreadingService::ExecutorThreadingService(vespalib::Executor& sharedExecutor,
FNET_Transport& transport,
const vespalib::Clock& clock,
vespalib::ISequencedTaskExecutor& field_writer)
: ExecutorThreadingService(sharedExecutor, transport, clock, field_writer, nullptr, ThreadingServiceConfig::make())
{}
ExecutorThreadingService::ExecutorThreadingService(vespalib::Executor & sharedExecutor,
FNET_Transport & transport,
const vespalib::Clock & clock,
vespalib::ISequencedTaskExecutor& field_writer,
vespalib::InvokeService * invokerService,
const ThreadingServiceConfig & cfg)
: _sharedExecutor(sharedExecutor),
_transport(transport),
_clock(clock),
_masterExecutor(1, CpuUsage::wrap(master_executor, CpuUsage::Category::WRITE)),
_master_task_limit(cfg.master_task_limit()),
_indexExecutor(createExecutorWithOneThread(cfg, CpuUsage::wrap(index_executor, CpuUsage::Category::WRITE))),
_summaryExecutor(createExecutorWithOneThread(cfg, CpuUsage::wrap(summary_executor, CpuUsage::Category::WRITE))),
_masterService(_masterExecutor),
_indexService(*_indexExecutor),
_field_writer(field_writer),
_invokeRegistrations()
{
if (cfg.optimize() == vespalib::Executor::OptimizeFor::THROUGHPUT && invokerService) {
_invokeRegistrations.push_back(invokerService->registerInvoke([executor=_indexExecutor.get()](){ executor->wakeup();}));
_invokeRegistrations.push_back(invokerService->registerInvoke([executor=_summaryExecutor.get()](){ executor->wakeup();}));
}
}
ExecutorThreadingService::~ExecutorThreadingService() = default;
void
ExecutorThreadingService::blocking_master_execute(vespalib::Executor::Task::UP task)
{
uint32_t limit = master_task_limit();
if (limit > 0) {
_masterExecutor.wait_for_task_count(limit);
}
_masterExecutor.execute(std::move(task));
}
void
ExecutorThreadingService::shutdown()
{
_masterExecutor.shutdown().sync();
_field_writer.sync_all();
_summaryExecutor->shutdown().sync();
_indexExecutor->shutdown().sync();
_field_writer.sync_all();
_field_writer.sync_all();
}
void
ExecutorThreadingService::set_task_limits(uint32_t master_task_limit,
uint32_t field_task_limit,
uint32_t summary_task_limit)
{
_master_task_limit.store(master_task_limit, std::memory_order_release);
_indexExecutor->setTaskLimit(field_task_limit);
_summaryExecutor->setTaskLimit(summary_task_limit);
_field_writer.setTaskLimit(field_task_limit);
}
ExecutorThreadingServiceStats
ExecutorThreadingService::getStats()
{
auto master_stats = _masterExecutor.getStats();
auto index_stats = _indexExecutor->getStats();
auto summary_stats = _summaryExecutor->getStats();
return ExecutorThreadingServiceStats(master_stats, index_stats, summary_stats);
}
vespalib::ISequencedTaskExecutor &
ExecutorThreadingService::field_writer() {
return _field_writer;
}
}
|
