Skip to content
Navigation Menu
{{ message }}
forked from lewissbaker/cppcoro
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathio_service.hpp
More file actions
321 lines (249 loc) · 8.65 KB
/
Copy pathio_service.hpp
File metadata and controls
321 lines (249 loc) · 8.65 KB
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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
///////////////////////////////////////////////////////////////////////////////
// Copyright (c) Lewis Baker
// Licenced under MIT license. See LICENSE.txt for details.
///////////////////////////////////////////////////////////////////////////////
#ifndef CPPCORO_IO_SERVICE_HPP_INCLUDED
#define CPPCORO_IO_SERVICE_HPP_INCLUDED
#include <cppcoro/config.hpp>
#include <cppcoro/cancellation_token.hpp>
#include <cppcoro/cancellation_registration.hpp>
#if CPPCORO_OS_WINNT
# include <cppcoro/detail/win32.hpp>
#endif
#include <optional>
#include <chrono>
#include <cstdint>
#include <atomic>
#include <utility>
#include <mutex>
#include <experimental/coroutine>
namespace cppcoro
{
class io_service
{
public:
class schedule_operation;
class timed_schedule_operation;
/// Initialises the io_service.
///
/// Does not set a concurrency hint. All threads that enter the
/// event loop will actively process events.
io_service();
/// Initialise the io_service with a concurrency hint.
///
/// \param concurrencyHint
/// Specifies the target maximum number of I/O threads to be
/// actively processing events.
/// Note that the number of active threads may temporarily go
/// above this number.
io_service(std::uint32_t concurrencyHint);
~io_service();
io_service(io_service&& other) = delete;
io_service(const io_service& other) = delete;
io_service& operator=(io_service&& other) = delete;
io_service& operator=(const io_service& other) = delete;
/// Returns an operation that when awaited suspends the awaiting
/// coroutine and reschedules it for resumption on an I/O thread
/// associated with this io_service.
[[nodiscard]]
schedule_operation schedule() noexcept;
/// Returns an operation that when awaited will suspend the
/// awaiting coroutine for the specified delay. Once the delay
/// has elapsed, the coroutine will resume execution on an
/// I/O thread associated with this io_service.
///
/// \param delay
/// The amount of time to delay scheduling resumption of the coroutine
/// on an I/O thread. There is no guarantee that the coroutine will
/// be resumed exactly after this delay.
///
/// \param cancellationToken [optional]
/// A cancellation token that can be used to communicate a request to
/// cancel the delayed schedule operation and schedule it for resumption
/// immediately.
/// The co_await operation will throw cppcoro::operation_cancelled if
/// cancellation was requested before the coroutine could be resumed.
template<typename REP, typename PERIOD>
[[nodiscard]]
timed_schedule_operation schedule_after(
const std::chrono::duration<REP, PERIOD>& delay,
cancellation_token cancellationToken = {}) noexcept;
/// Process events until the io_service is stopped.
///
/// \return
/// The number of events processed during this call.
std::uint64_t process_events();
/// Process events until either the io_service is stopped or
/// there are no more pending events in the queue.
///
/// \return
/// The number of events processed during this call.
std::uint64_t process_pending_events();
/// Block until either one event is processed or the io_service is stopped.
///
/// \return
/// The number of events processed during this call.
/// This will either be 0 or 1.
std::uint64_t process_one_event();
/// Process one event if there are any events pending, otherwise if there
/// are no events pending or the io_service is stopped then return immediately.
///
/// \return
/// The number of events processed during this call.
/// This will either be 0 or 1.
std::uint64_t process_one_pending_event();
/// Shut down the io_service.
///
/// This will cause any threads currently in a call to one of the process_xxx() methods
/// to return from that call once they finish processing the current event.
///
/// This call does not wait until all threads have exited the event loop so you
/// must use other synchronisation mechanisms to wait for those threads.
void stop() noexcept;
/// Reset an io_service to prepare it for resuming processing of events.
///
/// Call this after a call to stop() to allow calls to process_xxx() methods
/// to process events.
///
/// After calling stop() you should ensure that all threads have returned from
/// calls to process_xxx() methods before calling reset().
void reset();
bool is_stop_requested() const noexcept;
void notify_work_started() noexcept;
void notify_work_finished() noexcept;
#if CPPCORO_OS_WINNT
detail::win32::handle_t native_iocp_handle() noexcept;
void ensure_winsock_initialised();
#endif
private:
class timer_thread_state;
class timer_queue;
friend class schedule_operation;
friend class timed_schedule_operation;
void schedule_impl(schedule_operation* operation) noexcept;
void try_reschedule_overflow_operations() noexcept;
bool try_enter_event_loop() noexcept;
void exit_event_loop() noexcept;
bool try_process_one_event(bool waitForEvent);
void post_wake_up_event() noexcept;
timer_thread_state* ensure_timer_thread_started();
static constexpr std::uint32_t stop_requested_flag = 1;
static constexpr std::uint32_t active_thread_count_increment = 2;
// Bit 0: stop_requested_flag
// Bit 1-31: count of active threads currently running the event loop
std::atomic<std::uint32_t> m_threadState;
std::atomic<std::uint32_t> m_workCount;
#if CPPCORO_OS_WINNT
detail::win32::safe_handle m_iocpHandle;
std::atomic<bool> m_winsockInitialised;
std::mutex m_winsockInitialisationMutex;
#endif
// Head of a linked-list of schedule operations that are
// ready to run but that failed to be queued to the I/O
// completion port (eg. due to low memory).
std::atomic<schedule_operation*> m_scheduleOperations;
std::atomic<timer_thread_state*> m_timerState;
};
class io_service::schedule_operation
{
public:
schedule_operation(io_service& service) noexcept
: m_service(service)
{}
bool await_ready() const noexcept { return false; }
void await_suspend(std::experimental::coroutine_handle<> awaiter) noexcept;
void await_resume() const noexcept {}
private:
friend class io_service;
friend class io_service::timed_schedule_operation;
io_service& m_service;
std::experimental::coroutine_handle<> m_awaiter;
schedule_operation* m_next;
};
class io_service::timed_schedule_operation
{
public:
timed_schedule_operation(
io_service& service,
std::chrono::high_resolution_clock::time_point resumeTime,
cppcoro::cancellation_token cancellationToken) noexcept;
timed_schedule_operation(timed_schedule_operation&& other) noexcept;
~timed_schedule_operation();
timed_schedule_operation& operator=(timed_schedule_operation&& other) = delete;
timed_schedule_operation(const timed_schedule_operation& other) = delete;
timed_schedule_operation& operator=(const timed_schedule_operation& other) = delete;
bool await_ready() const noexcept;
void await_suspend(std::experimental::coroutine_handle<> awaiter);
void await_resume();
private:
friend class io_service::timer_queue;
friend class io_service::timer_thread_state;
io_service::schedule_operation m_scheduleOperation;
std::chrono::high_resolution_clock::time_point m_resumeTime;
cppcoro::cancellation_token m_cancellationToken;
std::optional<cppcoro::cancellation_registration> m_cancellationRegistration;
timed_schedule_operation* m_next;
std::atomic<std::uint32_t> m_refCount;
};
class io_work_scope
{
public:
explicit io_work_scope(io_service& service) noexcept
: m_service(&service)
{
service.notify_work_started();
}
io_work_scope(const io_work_scope& other) noexcept
: m_service(other.m_service)
{
if (m_service != nullptr)
{
m_service->notify_work_started();
}
}
io_work_scope(io_work_scope&& other) noexcept
: m_service(other.m_service)
{
other.m_service = nullptr;
}
~io_work_scope()
{
if (m_service != nullptr)
{
m_service->notify_work_finished();
}
}
void swap(io_work_scope& other) noexcept
{
std::swap(m_service, other.m_service);
}
io_work_scope& operator=(io_work_scope other) noexcept
{
swap(other);
return *this;
}
io_service& service() noexcept
{
return *m_service;
}
private:
io_service* m_service;
};
inline void swap(io_work_scope& a, io_work_scope& b)
{
a.swap(b);
}
}
template<typename REP, typename RATIO>
cppcoro::io_service::timed_schedule_operation
cppcoro::io_service::schedule_after(
const std::chrono::duration<REP, RATIO>& duration,
cppcoro::cancellation_token cancellationToken) noexcept
{
return timed_schedule_operation{
*this,
std::chrono::high_resolution_clock::now() + duration,
std::move(cancellationToken)
};
}
#endif
You can’t perform that action at this time.