blob: 3b4f12b7c85ded24bcf19f9d767961a75889cd8a (
plain) (
blame)
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
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#include "executor_thread_service.h"
#include <vespa/vespalib/util/lambdatask.h>
#include <vespa/vespalib/util/gate.h>
#include <vespa/fastos/thread.h>
using vespalib::makeLambdaTask;
using vespalib::Executor;
using vespalib::Gate;
using vespalib::Runnable;
using vespalib::SyncableThreadExecutor;
namespace proton {
namespace internal {
struct ThreadId {
FastOS_ThreadId _id;
};
}
namespace {
void
sampleThreadId(FastOS_ThreadId *threadId)
{
*threadId = FastOS_Thread::GetCurrentThreadId();
}
std::unique_ptr<internal::ThreadId>
getThreadId(SyncableThreadExecutor &executor)
{
std::unique_ptr<internal::ThreadId> id = std::make_unique<internal::ThreadId>();
executor.execute(makeLambdaTask([threadId=&id->_id] { sampleThreadId(threadId);}));
executor.sync();
return id;
}
void
runRunnable(Runnable *runnable, Gate *gate)
{
runnable->run();
gate->countDown();
}
} // namespace
ExecutorThreadService::ExecutorThreadService(SyncableThreadExecutor &executor)
: _executor(executor),
_threadId(getThreadId(executor))
{
}
ExecutorThreadService::~ExecutorThreadService() = default;
void
ExecutorThreadService::run(Runnable &runnable)
{
if (isCurrentThread()) {
runnable.run();
} else {
Gate gate;
_executor.execute(makeLambdaTask([runnablePtr=&runnable, gatePtr=&gate] { runRunnable(runnablePtr, gatePtr); }));
gate.await();
}
}
bool
ExecutorThreadService::isCurrentThread() const
{
FastOS_ThreadId currentThreadId = FastOS_Thread::GetCurrentThreadId();
return FastOS_Thread::CompareThreadIds(_threadId->_id, currentThreadId);
}
vespalib::ThreadExecutor::Stats ExecutorThreadService::getStats() {
return _executor.getStats();
}
void ExecutorThreadService::setTaskLimit(uint32_t taskLimit) {
_executor.setTaskLimit(taskLimit);
}
uint32_t ExecutorThreadService::getTaskLimit() const {
return _executor.getTaskLimit();
}
void
ExecutorThreadService::wakeup() {
_executor.wakeup();
}
} // namespace proton
|
