maph 1.0.0 High-Performance Memory-Mapped Key-Value Database with O(1) Lookups

Loading...
Searching...
No Matches
asio.cpp
Go to the documentation of this file.
1/*
2 * Authored by Alex Hultman, 2018-2021.
3 * Intellectual property of third-party.
4
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8
9 * http://www.apache.org/licenses/LICENSE-2.0
10
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 */
17
18 extern "C" {
19 #include "libusockets.h"
20 #include "internal/internal.h"
21 #include <stdlib.h>
22}
23
24#ifdef LIBUS_USE_ASIO
25
26#include <boost/asio.hpp>
27#include <iostream>
28#include <mutex>
29#include <memory>
30#include <boost/version.hpp>
31
32// New interfaces require boost 1.66.0
33#if BOOST_VERSION < 106600
34#define LIBUS_USE_OLD_ASIO
35#define LIBUS_ASIO_DESCRIPTOR boost::asio::posix::stream_descriptor
36#define LIBUS_ASIO_LOOP boost::asio::io_service
37#else
38#define LIBUS_ASIO_DESCRIPTOR boost::asio::posix::descriptor
39#define LIBUS_ASIO_LOOP boost::asio::io_context
40#endif
41
42// setting polls to 1 disables fallthrough
43 int polls = 0; // temporary solution keeping track of outstanding work
44
45// define a timer internally as something that inherits from callback_t
46// us_timer_t is convertible to this one
47 struct boost_timer : us_internal_callback_t {
48 boost::asio::deadline_timer timer;
49 std::shared_ptr<boost_timer> isValid;
50
51 unsigned char nr = 0;
52
53 boost_timer(LIBUS_ASIO_LOOP *io) : timer(*io) {
54 isValid.reset(this, [](boost_timer *t) {});
55 }
56};
57
58 struct boost_block_poll_t : LIBUS_ASIO_DESCRIPTOR {
59
60 boost_block_poll_t(LIBUS_ASIO_LOOP *io, us_poll_t *p) : LIBUS_ASIO_DESCRIPTOR(*io), p(p) {
61 isValid.reset(this, [](boost_block_poll_t *t) {});
62 }
63
64 std::shared_ptr<boost_block_poll_t> isValid;
65
66 unsigned char nr = 0;
67 struct us_poll_t *p;
68};
69
70 extern "C" {
71
72// poll
73 void us_poll_init(struct us_poll_t *p, LIBUS_SOCKET_DESCRIPTOR fd, int poll_type) {
74 struct boost_block_poll_t *boost_block = (struct boost_block_poll_t *) p->boost_block;
75 boost_block->assign(fd);
76 p->poll_type = poll_type;
77 p->events = 0;
78
79 p->fd = fd; //apparently we access fd after close
80}
81
82 void us_poll_free(struct us_poll_t *p, struct us_loop_t *loop) {
83 struct boost_block_poll_t *boost_block = (struct boost_block_poll_t *) p->boost_block;
84
85 delete boost_block;
86 free(p);
87}
88
89 void poll_for_error(struct boost_block_poll_t *boost_block) {
90 /* There is no such thing as polling for error in old asio */
91#ifndef LIBUS_USE_OLD_ASIO
92 polls++;
93 boost_block->async_wait(boost::asio::posix::descriptor::wait_type::wait_error, [nr = boost_block->nr, weakBoostBlock = std::weak_ptr<boost_block_poll_t>(boost_block->isValid)](boost::system::error_code ec) {
94 polls--;
95
96 if (ec != boost::asio::error::operation_aborted) {
97
98 // post mortem check
99 struct boost_block_poll_t *boost_block;
100 if (auto observe = weakBoostBlock.lock()) {
101 boost_block = observe.get();
102 } else {
103 return;
104 }
105
106 // get boost_block from weakptr
107 if (nr != boost_block->nr) {
108 return;
109 }
110
111 poll_for_error(boost_block); // ska man verkligen polla for error igen
112 us_internal_dispatch_ready_poll(boost_block->p, 1, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE);
113 }
114 });
115#endif
116}
117
118 void poll_for_read(struct boost_block_poll_t *boost_block);
119
120 inline void handle_read(const std::weak_ptr<boost_block_poll_t> &weakBoostBlock, unsigned char nr, boost::system::error_code ec) {
121 if (ec != boost::asio::error::operation_aborted) {
122
123 // post mortem check
124 struct boost_block_poll_t *boost_block;
125 if (auto observe = weakBoostBlock.lock()) {
126 boost_block = observe.get();
127 } else {
128 return;
129 }
130
131 // get boost_block from weakptr
132 if (nr != boost_block->nr) {
133 return;
134 }
135
136 poll_for_read(boost_block);
137 us_internal_dispatch_ready_poll(boost_block->p, ec ? -1 : 0, LIBUS_SOCKET_READABLE);
138 }
139}
140
141 void poll_for_read(struct boost_block_poll_t *boost_block) {
142 polls++;
143#ifndef LIBUS_USE_OLD_ASIO
144 boost_block->async_wait(boost::asio::posix::descriptor::wait_type::wait_read, [nr = boost_block->nr, weakBoostBlock = std::weak_ptr<boost_block_poll_t>(boost_block->isValid)](boost::system::error_code ec) {
145 polls--;
146 handle_read(weakBoostBlock, nr, ec);
147 });
148#else
149 boost_block->async_read_some(boost::asio::null_buffers(), [nr = boost_block->nr, weakBoostBlock = std::weak_ptr<boost_block_poll_t>(boost_block->isValid)](boost::system::error_code ec, std::size_t) {
150 polls--;
151 handle_read(weakBoostBlock, nr, ec);
152 });
153#endif
154}
155
156 void poll_for_write(struct boost_block_poll_t *boost_block);
157
158 inline void handle_write(const std::weak_ptr<boost_block_poll_t> &weakBoostBlock, unsigned char nr, boost::system::error_code ec) {
159 if (ec != boost::asio::error::operation_aborted) {
160
161 // post mortem check
162 struct boost_block_poll_t *boost_block;
163 if (auto observe = weakBoostBlock.lock()) {
164 boost_block = observe.get();
165 } else {
166 return;
167 }
168
169 // get boost_block from weakptr
170 if (nr != boost_block->nr) {
171 return;
172 }
173 poll_for_write(boost_block);
174 us_internal_dispatch_ready_poll(boost_block->p, ec ? -1 : 0, LIBUS_SOCKET_WRITABLE);
175 }
176}
177
178 void poll_for_write(struct boost_block_poll_t *boost_block) {
179 polls++;
180#ifndef LIBUS_USE_OLD_ASIO
181 boost_block->async_wait(boost::asio::posix::descriptor::wait_type::wait_write, [nr = boost_block->nr, weakBoostBlock = std::weak_ptr<boost_block_poll_t>(boost_block->isValid)](boost::system::error_code ec) {
182 polls--;
183 handle_write(weakBoostBlock, nr, ec);
184 });
185#else
186 boost_block->async_write_some(boost::asio::null_buffers(), [nr = boost_block->nr, weakBoostBlock = std::weak_ptr<boost_block_poll_t>(boost_block->isValid)](boost::system::error_code ec, std::size_t) {
187 polls--;
188 handle_write(weakBoostBlock, nr, ec);
189 });
190#endif
191}
192
193 void us_poll_start(struct us_poll_t *p, struct us_loop_t *loop, int events) {
194 struct boost_block_poll_t *boost_block = (struct boost_block_poll_t *) p->boost_block;
195
196 p->events = events;
197 poll_for_error(boost_block);
198
199 if (events & LIBUS_SOCKET_READABLE) {
200 poll_for_read(boost_block);
201 }
202
203 if (events & LIBUS_SOCKET_WRITABLE) {
204 poll_for_write(boost_block);
205 }
206}
207
208 void us_poll_change(struct us_poll_t *p, struct us_loop_t *loop, int events) {
209 struct boost_block_poll_t *boost_block = (struct boost_block_poll_t *) p->boost_block;
210
211 boost_block->nr++;
212 boost_block->cancel();
213
214 us_poll_start(p, loop, events);
215}
216
217 void us_poll_stop(struct us_poll_t *p, struct us_loop_t *loop) {
218 struct boost_block_poll_t *boost_block = (struct boost_block_poll_t *) p->boost_block;
219
220 boost_block->nr++;
221 boost_block->release();
222}
223
224 int us_poll_events(struct us_poll_t *p) {
225 return p->events;
226}
227
228 unsigned int us_internal_accept_poll_event(struct us_poll_t *p) {
229 return 0;
230}
231
232 int us_internal_poll_type(struct us_poll_t *p) {
233 return p->poll_type;
234}
235
236 void us_internal_poll_set_type(struct us_poll_t *p, int poll_type) {
237 p->poll_type = poll_type;
238}
239
240 LIBUS_SOCKET_DESCRIPTOR us_poll_fd(struct us_poll_t *p) {
241 struct boost_block_poll_t *boost_block = (struct boost_block_poll_t *) p->boost_block;
242
243
244 return p->fd;
245
246 //return boost_block->native_handle();
247}
248
249// if we get an io_context ptr as hint, we use it
250// otherwise we create a new one for only us
251 struct us_loop_t *us_create_loop(void *hint, void (*wakeup_cb)(struct us_loop_t *loop), void (*pre_cb)(struct us_loop_t *loop), void (*post_cb)(struct us_loop_t *loop), unsigned int ext_size) {
252 struct us_loop_t *loop = (struct us_loop_t *) malloc(sizeof(struct us_loop_t) + ext_size);
253
254 loop->io = hint ? hint : new LIBUS_ASIO_LOOP();
255 loop->is_default = hint != 0;
256
257 // here we create two unreffed handles - timer and async
258 us_internal_loop_data_init(loop, wakeup_cb, pre_cb, post_cb);
259
260 // if we do not own this loop, we need to integrate and set up timer
261 if (hint) {
262 us_loop_integrate(loop);
263 }
264
265 return loop;
266}
267
268// based on if this was default loop or not
269 void us_loop_free(struct us_loop_t *loop) {
270 us_internal_loop_data_free(loop);
271
272 if (!loop->is_default) {
273 delete (LIBUS_ASIO_LOOP *) loop->io;
274 }
275
276 free(loop);
277}
278
279// we need fallthrough to correspond to our polls
280// therefore we exit when our polls are 0
281// if third party asio server wants to keep the loop running
282// they have to use a guard such as a us_timer_t
283 void us_loop_run(struct us_loop_t *loop) {
284 us_loop_integrate(loop);
285
286 // this way of running adds one extra epoll_wait per event loop iteration
287 // but does not add per-poll overhead. besides, asio is sprinkled with inefficiencies
288 // everywhere so it's negligible for what it solves (we must have pre, post callbacks)
289 while (polls) {
290 us_internal_loop_pre(loop);
291 size_t num = ((LIBUS_ASIO_LOOP *) loop->io)->run_one();
292 if (!num) {
293 break;
294 }
295
296 for (int i = 0; true; i++) {
297 num = ((LIBUS_ASIO_LOOP *) loop->io)->poll_one();
298 if (!num || i == 999) {
299 break;
300 }
301 }
302 us_internal_loop_post(loop);
303 }
304}
305
306 struct us_poll_t *us_create_poll(struct us_loop_t *loop, int fallthrough, unsigned int ext_size) {
307 struct us_poll_t *p = (struct us_poll_t *) malloc(sizeof(struct us_poll_t) + ext_size);
308 p->boost_block = new boost_block_poll_t( (LIBUS_ASIO_LOOP *)loop->io, p);
309
310 return p;
311}
312
313/* If we update our block position we have to updarte the uv_poll data to point to us */
314 struct us_poll_t *us_poll_resize(struct us_poll_t *p, struct us_loop_t *loop, unsigned int ext_size) {
315 p = (struct us_poll_t *) realloc(p, sizeof(struct us_poll_t) + ext_size);
316
317 // captures must never capture p directly, only boost_block and derive p from there
318 ((struct boost_block_poll_t *) p->boost_block)->p = p;
319
320 return p;
321}
322
323// timer
324 struct us_timer_t *us_create_timer(struct us_loop_t *loop, int fallthrough, unsigned int ext_size) {
325 struct boost_timer *cb = (struct boost_timer *) malloc(sizeof(struct boost_timer) + ext_size);
326
327 // inplace construct the timer on this callback_t
328 new (cb) boost_timer((LIBUS_ASIO_LOOP *)loop->io);
329
330 cb->loop = loop;
331 cb->cb_expects_the_loop = 0;
332 cb->p.poll_type = POLL_TYPE_CALLBACK; // this is missing from libuv flow
333
334 if (fallthrough) {
335 //uv_unref((uv_handle_t *) uv_timer);
336 }
337
338 return (struct us_timer_t *) cb;
339}
340
341 void *us_timer_ext(struct us_timer_t *timer) {
342 return ((struct boost_timer *) timer) + 1;
343}
344
345 void us_timer_close(struct us_timer_t *t) {
346 ((boost_timer *) t)->timer.cancel();
347 ((boost_timer *) t)->~boost_timer();
348 free(t);
349}
350
351 void poll_for_timeout(struct boost_timer *b_timer, int repeat_ms) {
352 b_timer->timer.async_wait([nr = b_timer->nr, repeat_ms, weakBoostBlock = std::weak_ptr<boost_timer>(b_timer->isValid)](const boost::system::error_code &ec) {
353 if (ec != boost::asio::error::operation_aborted) {
354
355 struct boost_timer *b_timer;
356 if (auto observe = weakBoostBlock.lock()) {
357 b_timer = observe.get();
358 } else {
359 return;
360 }
361
362 if (nr != b_timer->nr) {
363 return;
364 }
365
366 if (repeat_ms) {
367
368 if (!polls) {
369 // we do fallthrough if no other polling
370 // this is problematic if WE fallthrough
371 // but other parts do not
372 // that causes timeouts to stop working
373 // we should really ask the executor if there is work
374 // if there is, then continue ticking until there isn't
375 return;
376 }
377
378 b_timer->timer.expires_at(b_timer->timer.expires_at() + boost::posix_time::milliseconds(repeat_ms));
379 poll_for_timeout(b_timer, repeat_ms);
380 }
381 us_internal_dispatch_ready_poll((struct us_poll_t *)b_timer, 0, LIBUS_SOCKET_READABLE);
382 }
383 });
384}
385
386 void us_timer_set(struct us_timer_t *t, void (*cb)(struct us_timer_t *t), int ms, int repeat_ms) {
387 struct boost_timer *b_timer = (struct boost_timer *) t;
388
389 if (!ms) {
390 b_timer->nr++;
391 b_timer->timer.cancel();
392 } else {
393 b_timer->cb = (void(*)(struct us_internal_callback_t *)) cb;
394
395 b_timer->timer.expires_from_now(boost::posix_time::milliseconds(ms));
396 poll_for_timeout(b_timer, repeat_ms);
397 }
398}
399
400 struct us_loop_t *us_timer_loop(struct us_timer_t *t) {
401 struct us_internal_callback_t *internal_cb = (struct us_internal_callback_t *) t;
402
403 return internal_cb->loop;
404}
405
406// async (internal only) probably map to io_context::post
407 struct boost_async : us_internal_callback_t {
408 std::mutex m;
409 std::shared_ptr<boost_async> isValid;
410
411 boost_async() {
412 isValid.reset(this, [](boost_async *a) {});
413 }
414};
415
416 struct us_internal_async *us_internal_create_async(struct us_loop_t *loop, int fallthrough, unsigned int ext_size) {
417 struct boost_async *cb = (struct boost_async *) malloc(sizeof(struct boost_async) + ext_size);
418
419 // inplace construct
420 new (cb) boost_async();
421
422 // these properties are accessed from another thread when wakeup
423 cb->m.lock();
424 cb->loop = loop; // the only lock needed
425 cb->cb_expects_the_loop = 0;
426 cb->p.poll_type = POLL_TYPE_CALLBACK; // this is missing from libuv flow
427 cb->m.unlock();
428
429 if (fallthrough) {
430 //uv_unref((uv_handle_t *) uv_timer);
431 }
432
433 return (struct us_internal_async *) cb;
434}
435
436 void us_internal_async_close(struct us_internal_async *a) {
437 ((boost_async *) a)->~boost_async();
438 free(a);
439}
440
441 void us_internal_async_set(struct us_internal_async *a, void (*cb)(struct us_internal_async *)) {
442 struct boost_async *internal_cb = (struct boost_async *) a;
443
444 internal_cb->cb = (void(*)(struct us_internal_callback_t *)) cb;
445}
446
447 void us_internal_async_wakeup(struct us_internal_async *a) {
448 struct boost_async *cb = (struct boost_async *) a;
449
450 // this doesn't really guarantee loop.io being visible here
451 // really we should use the loops mutex, and have the loops constructor
452 // use its own mutex, then we are guaranteed to have visibility here
453 cb->m.lock();
454 LIBUS_ASIO_LOOP *io = (LIBUS_ASIO_LOOP *)cb->loop->io;
455 cb->m.unlock();
456
457 // should increase and decrease polls (again, loop mutex)
458 io->post([weakBoostBlock = std::weak_ptr<boost_async>(cb->isValid)]() {
459
460 // was the async deleted before we came here?
461 struct boost_async *cb;
462 if (auto observe = weakBoostBlock.lock()) {
463 cb = observe.get();
464 } else {
465 return;
466 }
467
468 us_internal_dispatch_ready_poll((struct us_poll_t *) cb, 0, 0);
469
470 });
471}
472
473}
474
475#endif
LIBUS_SOCKET_WRITABLE
#define LIBUS_SOCKET_WRITABLE
Definition asio.h:24
LIBUS_SOCKET_READABLE
#define LIBUS_SOCKET_READABLE
Definition asio.h:23
internal.h
us_internal_poll_set_type
void us_internal_poll_set_type(struct us_poll_t *p, int poll_type)
us_internal_async_wakeup
void us_internal_async_wakeup(struct us_internal_async *a)
us_internal_async_set
void us_internal_async_set(struct us_internal_async *a, void(*cb)(struct us_internal_async *))
us_internal_loop_post
void us_internal_loop_post(struct us_loop_t *loop)
Definition loop.c:190
us_internal_loop_data_free
void us_internal_loop_data_free(struct us_loop_t *loop)
Definition loop.c:42
us_internal_async_close
void us_internal_async_close(struct us_internal_async *a)
us_internal_create_async
struct us_internal_async * us_internal_create_async(struct us_loop_t *loop, int fallthrough, unsigned int ext_size)
us_internal_poll_type
int us_internal_poll_type(struct us_poll_t *p)
us_internal_loop_pre
void us_internal_loop_pre(struct us_loop_t *loop)
Definition loop.c:184
us_internal_dispatch_ready_poll
void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int events)
Definition loop.c:195
POLL_TYPE_CALLBACK
@ POLL_TYPE_CALLBACK
Definition internal.h:50
us_internal_accept_poll_event
unsigned int us_internal_accept_poll_event(struct us_poll_t *p)
us_internal_loop_data_init
void us_internal_loop_data_init(struct us_loop_t *loop, void(*wakeup_cb)(struct us_loop_t *loop), void(*pre_cb)(struct us_loop_t *loop), void(*post_cb)(struct us_loop_t *loop))
Definition loop.c:23
libusockets.h
us_create_timer
struct us_timer_t * us_create_timer(struct us_loop_t *loop, int fallthrough, unsigned int ext_size)
us_poll_init
void us_poll_init(struct us_poll_t *p, int fd, int poll_type)
LIBUS_SOCKET_DESCRIPTOR
#define LIBUS_SOCKET_DESCRIPTOR
Definition libusockets.h:38
us_create_poll
struct us_poll_t * us_create_poll(struct us_loop_t *loop, int fallthrough, unsigned int ext_size)
us_poll_change
void us_poll_change(struct us_poll_t *p, struct us_loop_t *loop, int events)
us_poll_resize
struct us_poll_t * us_poll_resize(struct us_poll_t *p, struct us_loop_t *loop, unsigned int ext_size)
us_poll_fd
int us_poll_fd(struct us_poll_t *p)
us_poll_events
int us_poll_events(struct us_poll_t *p)
us_poll_stop
void us_poll_stop(struct us_poll_t *p, struct us_loop_t *loop)
us_loop_run
void us_loop_run(struct us_loop_t *loop)
us_timer_loop
struct us_loop_t * us_timer_loop(struct us_timer_t *t)
us_timer_close
void us_timer_close(struct us_timer_t *timer)
us_poll_free
void us_poll_free(struct us_poll_t *p, struct us_loop_t *loop)
us_create_loop
struct us_loop_t * us_create_loop(void *hint, void(*wakeup_cb)(struct us_loop_t *loop), void(*pre_cb)(struct us_loop_t *loop), void(*post_cb)(struct us_loop_t *loop), unsigned int ext_size)
us_loop_free
void us_loop_free(struct us_loop_t *loop)
us_timer_ext
void * us_timer_ext(struct us_timer_t *timer)
us_poll_start
void us_poll_start(struct us_poll_t *p, struct us_loop_t *loop, int events)
us_timer_set
void us_timer_set(struct us_timer_t *timer, void(*cb)(struct us_timer_t *t), int ms, int repeat_ms)
us_loop_integrate
void us_loop_integrate(struct us_loop_t *loop)
Definition loop.c:356
us_poll_t::fd
LIBUS_SOCKET_DESCRIPTOR fd
Definition asio.h:40
us_poll_t::events
int events
Definition asio.h:42
us_poll_t::poll_type
unsigned char poll_type
Definition asio.h:41
us_loop_t::io
void * io
Definition asio.h:30
us_loop_t::is_default
int is_default
Definition asio.h:33
us_poll_t::boost_block
void * boost_block
Definition asio.h:38
us_poll_t
Definition asio.h:37
us_loop_t
Definition asio.h:26
us_internal_callback_t
Definition internal.h:99
us_internal_callback_t::loop
struct us_loop_t * loop
Definition internal.h:101
us_internal_callback_t::cb
void(* cb)(struct us_internal_callback_t *cb)
Definition internal.h:104