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
127
128
129
|
#pragma once
#include <mbgl/actor/scheduler.hpp>
#include <mbgl/actor/mailbox.hpp>
#include <mbgl/util/noncopyable.hpp>
#include <mbgl/util/util.hpp>
#include <mbgl/util/work_task.hpp>
#include <mbgl/util/work_request.hpp>
#include <atomic>
#include <functional>
#include <utility>
#include <queue>
#include <mutex>
namespace mbgl {
namespace util {
using LOOP_HANDLE = void *;
class RunLoop : public Scheduler,
private util::noncopyable {
public:
enum class Type : uint8_t {
Default,
New,
};
enum class Priority : bool {
Default = false,
High = true,
};
enum class Event : uint8_t {
None = 0,
Read = 1,
Write = 2,
ReadWrite = Read | Write,
};
RunLoop(Type type = Type::Default);
~RunLoop() override;
static RunLoop* Get();
static LOOP_HANDLE getLoopHandle();
void run();
void runOnce();
void stop();
// So far only needed by the libcurl backend.
void addWatch(int fd, Event, std::function<void(int, Event)>&& callback);
void removeWatch(int fd);
// Invoke fn(args...) on this RunLoop.
template <class Fn, class... Args>
void invoke(Priority priority, Fn&& fn, Args&&... args) {
push(priority, WorkTask::make(std::forward<Fn>(fn), std::forward<Args>(args)...));
}
// Invoke fn(args...) on this RunLoop.
template <class Fn, class... Args>
void invoke(Fn&& fn, Args&&... args) {
invoke(Priority::Default, std::forward<Fn>(fn), std::forward<Args>(args)...);
}
// Post the cancellable work fn(args...) to this RunLoop.
template <class Fn, class... Args>
std::unique_ptr<AsyncRequest>
invokeCancellable(Fn&& fn, Args&&... args) {
std::shared_ptr<WorkTask> task = WorkTask::make(std::forward<Fn>(fn), std::forward<Args>(args)...);
push(Priority::Default, task);
return std::make_unique<WorkRequest>(task);
}
void schedule(std::function<void()> fn) override { invoke(std::move(fn)); }
class Impl;
private:
MBGL_STORE_THREAD(tid)
using Queue = std::queue<std::shared_ptr<WorkTask>>;
// Wakes up the RunLoop so that it starts processing items in the queue.
void wake();
// Adds a WorkTask to the queue, and wakes it up.
void push(Priority priority, std::shared_ptr<WorkTask> task) {
std::lock_guard<std::mutex> lock(mutex);
if (priority == Priority::High) {
highPriorityQueue.emplace(std::move(task));
} else {
defaultQueue.emplace(std::move(task));
}
wake();
}
void process() {
std::shared_ptr<WorkTask> task;
std::unique_lock<std::mutex> lock(mutex);
while (true) {
if (!highPriorityQueue.empty()) {
task = std::move(highPriorityQueue.front());
highPriorityQueue.pop();
} else if (!defaultQueue.empty()) {
task = std::move(defaultQueue.front());
defaultQueue.pop();
} else {
break;
}
lock.unlock();
(*task)();
task.reset();
lock.lock();
}
}
Queue defaultQueue;
Queue highPriorityQueue;
std::mutex mutex;
std::unique_ptr<Impl> impl;
};
} // namespace util
} // namespace mbgl
#include <mbgl/util/work_task_impl.hpp>
|
