blob: 06cc37616c02b438db5ef4b3a7b1f6f438867ff0 (
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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include <vespa/vespalib/util/executor.h>
#include <vespa/vespalib/util/executor_stats.h>
#include <vespa/vespalib/stllike/string.h>
#include <vespa/vespalib/util/lambdatask.h>
#include <vector>
#include <mutex>
namespace vespalib {
/**
* Interface class to run multiple tasks in parallel, but tasks with same
* id has to be run in sequence.
*/
class ISequencedTaskExecutor
{
public:
class ExecutorId {
public:
ExecutorId() noexcept : ExecutorId(0) { }
explicit ExecutorId(uint32_t id) noexcept : _id(id) { }
uint32_t getId() const noexcept { return _id; }
bool operator != (ExecutorId rhs) const noexcept { return _id != rhs._id; }
bool operator == (ExecutorId rhs) const noexcept { return _id == rhs._id; }
bool operator < (ExecutorId rhs) const noexcept { return _id < rhs._id; }
private:
uint32_t _id;
};
ISequencedTaskExecutor(uint32_t numExecutors);
virtual ~ISequencedTaskExecutor();
/**
* Calculate which executor will handle an component.
*
* @param componentId component id
* @return executor id
*/
virtual ExecutorId getExecutorId(uint64_t componentId) const = 0;
uint32_t getNumExecutors() const { return _numExecutors; }
ExecutorId getExecutorIdFromName(vespalib::stringref componentId) const;
/**
* Schedule a task to run after all previously scheduled tasks with
* same id.
*
* @param id which internal executor to use
* @param task unique pointer to the task to be executed
*/
virtual void executeTask(ExecutorId id, vespalib::Executor::Task::UP task) = 0;
/**
* Call this one to ensure you get the attention of the workers.
*/
virtual void wakeup() { }
/**
* Wrap lambda function into a task and schedule it to be run.
* Caller must ensure that pointers and references are valid and
* call sync_all before tearing down pointed to/referenced data.
*
* @param id which internal executor to use
* @param function function to be wrapped in a task and later executed
*/
template <class FunctionType>
void executeLambda(ExecutorId id, FunctionType &&function) {
executeTask(id, vespalib::makeLambdaTask(std::forward<FunctionType>(function)));
}
/**
* Wait for all scheduled tasks to complete.
*/
virtual void sync_all() = 0;
virtual void setTaskLimit(uint32_t taskLimit) = 0;
virtual vespalib::ExecutorStats getStats() = 0;
/**
* Wrap lambda function into a task and schedule it to be run.
* Caller must ensure that pointers and references are valid and
* call sync_all before tearing down pointed to/referenced data.
*
* @param componentId component id
* @param function function to be wrapped in a task and later executed
*/
template <class FunctionType>
void execute(uint64_t componentId, FunctionType &&function) {
ExecutorId id = getExecutorId(componentId);
executeTask(id, vespalib::makeLambdaTask(std::forward<FunctionType>(function)));
}
/**
* Wrap lambda function into a task and schedule it to be run.
* Caller must ensure that pointers and references are valid and
* call sync_all before tearing down pointed to/referenced data.
*
* @param id executor id
* @param function function to be wrapped in a task and later executed
*/
template <class FunctionType>
void execute(ExecutorId id, FunctionType &&function) {
executeTask(id, vespalib::makeLambdaTask(std::forward<FunctionType>(function)));
}
private:
uint32_t _numExecutors;
};
}
|
