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

Loading...
Searching...
No Matches
libuv.c
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#include "libusockets.h"
19#include "internal/internal.h"
20#include <stdlib.h>
21
22#ifdef LIBUS_USE_LIBUV
23
24/* uv_poll_t->data always (except for most times after calling us_poll_stop) points to the us_poll_t */
25 static void poll_cb(uv_poll_t *p, int status, int events) {
26 us_internal_dispatch_ready_poll((struct us_poll_t *) p->data, status < 0, events);
27}
28
29 static void prepare_cb(uv_prepare_t *p) {
30 struct us_loop_t *loop = p->data;
31 us_internal_loop_pre(loop);
32}
33
34/* Note: libuv timers execute AFTER the post callback */
35 static void check_cb(uv_check_t *p) {
36 struct us_loop_t *loop = p->data;
37 us_internal_loop_post(loop);
38}
39
40/* Not used for polls, since polls need two frees */
41 static void close_cb_free(uv_handle_t *h) {
42 free(h->data);
43}
44
45/* This one is different for polls, since we need two frees here */
46 static void close_cb_free_poll(uv_handle_t *h) {
47 /* It is only in case we called us_poll_stop then quickly us_poll_free that we enter this.
48 * Most of the time, actual freeing is done by us_poll_free. */
49 if (h->data) {
50 free(h->data);
51 free(h);
52 }
53}
54
55 static void timer_cb(uv_timer_t *t) {
56 struct us_internal_callback_t *cb = t->data;
57 cb->cb(cb);
58}
59
60 static void async_cb(uv_async_t *a) {
61 struct us_internal_callback_t *cb = a->data;
62 // internal asyncs give their loop, not themselves
63 cb->cb((struct us_internal_callback_t *) cb->loop);
64}
65
66// poll
67 void us_poll_init(struct us_poll_t *p, LIBUS_SOCKET_DESCRIPTOR fd, int poll_type) {
68 p->poll_type = poll_type;
69 p->fd = fd;
70}
71
72 void us_poll_free(struct us_poll_t *p, struct us_loop_t *loop) {
73 /* The idea here is like so; in us_poll_stop we call uv_close after setting data of uv-poll to 0.
74 * This means that in close_cb_free we call free on 0 with does nothing, since us_poll_stop should
75 * not really free the poll. HOWEVER, if we then call us_poll_free while still closing the uv-poll,
76 * we simply change back the data to point to our structure so that we actually do free it like we should. */
77 if (uv_is_closing((uv_handle_t *) p->uv_p)) {
78 p->uv_p->data = p;
79 } else {
80 free(p->uv_p);
81 free(p);
82 }
83}
84
85 void us_poll_start(struct us_poll_t *p, struct us_loop_t *loop, int events) {
86 p->poll_type = us_internal_poll_type(p) | ((events & LIBUS_SOCKET_READABLE) ? POLL_TYPE_POLLING_IN : 0) | ((events & LIBUS_SOCKET_WRITABLE) ? POLL_TYPE_POLLING_OUT : 0);
87
88 uv_poll_init_socket(loop->uv_loop, p->uv_p, p->fd);
89 uv_poll_start(p->uv_p, events, poll_cb);
90}
91
92 void us_poll_change(struct us_poll_t *p, struct us_loop_t *loop, int events) {
93 if (us_poll_events(p) != events) {
94 p->poll_type = us_internal_poll_type(p) | ((events & LIBUS_SOCKET_READABLE) ? POLL_TYPE_POLLING_IN : 0) | ((events & LIBUS_SOCKET_WRITABLE) ? POLL_TYPE_POLLING_OUT : 0);
95
96 uv_poll_start(p->uv_p, events, poll_cb);
97 }
98}
99
100 void us_poll_stop(struct us_poll_t *p, struct us_loop_t *loop) {
101 uv_poll_stop(p->uv_p);
102
103 /* We normally only want to close the poll here, not free it. But if we stop it, then quickly "free" it with
104 * us_poll_free, we postpone the actual freeing to close_cb_free_poll whenever it triggers.
105 * That's why we set data to null here, so that us_poll_free can reset it if needed */
106 p->uv_p->data = 0;
107 uv_close((uv_handle_t *) p->uv_p, close_cb_free_poll);
108}
109
110 int us_poll_events(struct us_poll_t *p) {
111 return ((p->poll_type & POLL_TYPE_POLLING_IN) ? LIBUS_SOCKET_READABLE : 0) | ((p->poll_type & POLL_TYPE_POLLING_OUT) ? LIBUS_SOCKET_WRITABLE : 0);
112}
113
114 unsigned int us_internal_accept_poll_event(struct us_poll_t *p) {
115 return 0;
116}
117
118 int us_internal_poll_type(struct us_poll_t *p) {
119 return p->poll_type & 3;
120}
121
122 void us_internal_poll_set_type(struct us_poll_t *p, int poll_type) {
123 p->poll_type = poll_type | (p->poll_type & 12);
124}
125
126 LIBUS_SOCKET_DESCRIPTOR us_poll_fd(struct us_poll_t *p) {
127 return p->fd;
128}
129
130 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) {
131 struct us_loop_t *loop = (struct us_loop_t *) malloc(sizeof(struct us_loop_t) + ext_size);
132
133 loop->uv_loop = hint ? hint : uv_loop_new();
134 loop->is_default = hint != 0;
135
136 loop->uv_pre = malloc(sizeof(uv_prepare_t));
137 uv_prepare_init(loop->uv_loop, loop->uv_pre);
138 uv_prepare_start(loop->uv_pre, prepare_cb);
139 uv_unref((uv_handle_t *) loop->uv_pre);
140 loop->uv_pre->data = loop;
141
142 loop->uv_check = malloc(sizeof(uv_check_t));
143 uv_check_init(loop->uv_loop, loop->uv_check);
144 uv_unref((uv_handle_t *) loop->uv_check);
145 uv_check_start(loop->uv_check, check_cb);
146 loop->uv_check->data = loop;
147
148 // here we create two unreffed handles - timer and async
149 us_internal_loop_data_init(loop, wakeup_cb, pre_cb, post_cb);
150
151 // if we do not own this loop, we need to integrate and set up timer
152 if (hint) {
153 us_loop_integrate(loop);
154 }
155
156 return loop;
157}
158
159// based on if this was default loop or not
160 void us_loop_free(struct us_loop_t *loop) {
161 // ref and close down prepare and check
162 uv_ref((uv_handle_t *) loop->uv_pre);
163 uv_prepare_stop(loop->uv_pre);
164 loop->uv_pre->data = loop->uv_pre;
165 uv_close((uv_handle_t *) loop->uv_pre, close_cb_free);
166
167 uv_ref((uv_handle_t *) loop->uv_check);
168 uv_check_stop(loop->uv_check);
169 loop->uv_check->data = loop->uv_check;
170 uv_close((uv_handle_t *) loop->uv_check, close_cb_free);
171
172 us_internal_loop_data_free(loop);
173
174 // we need to run the loop one last round to call all close callbacks
175 // we cannot do this if we do not own the loop, default
176 if (!loop->is_default) {
177 uv_run(loop->uv_loop, UV_RUN_NOWAIT);
178 uv_loop_delete(loop->uv_loop);
179 }
180
181 // now we can free our part
182 free(loop);
183}
184
185 void us_loop_run(struct us_loop_t *loop) {
186 us_loop_integrate(loop);
187
188 uv_run(loop->uv_loop, UV_RUN_DEFAULT);
189}
190
191 struct us_poll_t *us_create_poll(struct us_loop_t *loop, int fallthrough, unsigned int ext_size) {
192 struct us_poll_t *p = (struct us_poll_t *) malloc(sizeof(struct us_poll_t) + ext_size);
193 p->uv_p = malloc(sizeof(uv_poll_t));
194 p->uv_p->data = p;
195 return p;
196}
197
198/* If we update our block position we have to updarte the uv_poll data to point to us */
199 struct us_poll_t *us_poll_resize(struct us_poll_t *p, struct us_loop_t *loop, unsigned int ext_size) {
200
201 struct us_poll_t *new_p = realloc(p, sizeof(struct us_poll_t) + ext_size);
202 new_p->uv_p->data = new_p;
203
204 return new_p;
205}
206
207// timer
208 struct us_timer_t *us_create_timer(struct us_loop_t *loop, int fallthrough, unsigned int ext_size) {
209 struct us_internal_callback_t *cb = malloc(sizeof(struct us_internal_callback_t) + sizeof(uv_timer_t) + ext_size);
210
211 cb->loop = loop;
212 cb->cb_expects_the_loop = 0; // never read?
213 cb->leave_poll_ready = 0; // never read?
214
215 uv_timer_t *uv_timer = (uv_timer_t *) (cb + 1);
216 uv_timer_init(loop->uv_loop, uv_timer);
217 uv_timer->data = cb;
218
219 if (fallthrough) {
220 uv_unref((uv_handle_t *) uv_timer);
221 }
222
223 return (struct us_timer_t *) cb;
224}
225
226 void *us_timer_ext(struct us_timer_t *timer) {
227 return ((char *) timer) + sizeof(struct us_internal_callback_t) + sizeof(uv_timer_t);
228}
229
230 void us_timer_close(struct us_timer_t *t) {
231 struct us_internal_callback_t *cb = (struct us_internal_callback_t *) t;
232
233 uv_timer_t *uv_timer = (uv_timer_t *) (cb + 1);
234
235 // always ref the timer before closing it
236 uv_ref((uv_handle_t *) uv_timer);
237
238 uv_timer_stop(uv_timer);
239
240 uv_timer->data = cb;
241 uv_close((uv_handle_t *) uv_timer, close_cb_free);
242}
243
244 void us_timer_set(struct us_timer_t *t, void (*cb)(struct us_timer_t *t), int ms, int repeat_ms) {
245 struct us_internal_callback_t *internal_cb = (struct us_internal_callback_t *) t;
246
247 internal_cb->cb = (void(*)(struct us_internal_callback_t *)) cb;
248
249 uv_timer_t *uv_timer = (uv_timer_t *) (internal_cb + 1);
250 if (!ms) {
251 uv_timer_stop(uv_timer);
252 } else {
253 uv_timer_start(uv_timer, timer_cb, ms, repeat_ms);
254 }
255}
256
257 struct us_loop_t *us_timer_loop(struct us_timer_t *t) {
258 struct us_internal_callback_t *internal_cb = (struct us_internal_callback_t *) t;
259
260 return internal_cb->loop;
261}
262
263// async (internal only)
264 struct us_internal_async *us_internal_create_async(struct us_loop_t *loop, int fallthrough, unsigned int ext_size) {
265 struct us_internal_callback_t *cb = malloc(sizeof(struct us_internal_callback_t) + sizeof(uv_async_t) + ext_size);
266
267 cb->loop = loop;
268 return (struct us_internal_async *) cb;
269}
270
271 void us_internal_async_close(struct us_internal_async *a) {
272 struct us_internal_callback_t *cb = (struct us_internal_callback_t *) a;
273
274 uv_async_t *uv_async = (uv_async_t *) (cb + 1);
275
276 // always ref the async before closing it
277 uv_ref((uv_handle_t *) uv_async);
278
279 uv_async->data = cb;
280 uv_close((uv_handle_t *) uv_async, close_cb_free);
281}
282
283 void us_internal_async_set(struct us_internal_async *a, void (*cb)(struct us_internal_async *)) {
284 struct us_internal_callback_t *internal_cb = (struct us_internal_callback_t *) a;
285
286 internal_cb->cb = (void (*)(struct us_internal_callback_t *)) cb;
287
288 uv_async_t *uv_async = (uv_async_t *) (internal_cb + 1);
289 uv_async_init(internal_cb->loop->uv_loop, uv_async, async_cb);
290 uv_unref((uv_handle_t *) uv_async);
291 uv_async->data = internal_cb;
292}
293
294 void us_internal_async_wakeup(struct us_internal_async *a) {
295 struct us_internal_callback_t *internal_cb = (struct us_internal_callback_t *) a;
296
297 uv_async_t *uv_async = (uv_async_t *) (internal_cb + 1);
298 uv_async_send(uv_async);
299}
300
301#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_POLLING_IN
@ POLL_TYPE_POLLING_IN
Definition internal.h:54
POLL_TYPE_POLLING_OUT
@ POLL_TYPE_POLLING_OUT
Definition internal.h:53
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_loop_t::uv_check
uv_check_t * uv_check
Definition libuv.h:34
us_poll_t::fd
LIBUS_SOCKET_DESCRIPTOR fd
Definition asio.h:40
us_loop_t::data
struct us_internal_loop_data_t data
Definition asio.h:27
us_loop_t::uv_pre
uv_prepare_t * uv_pre
Definition libuv.h:33
us_poll_t::poll_type
unsigned char poll_type
Definition asio.h:41
us_poll_t::uv_p
uv_poll_t * uv_p
Definition libuv.h:40
us_loop_t::uv_loop
uv_loop_t * uv_loop
Definition libuv.h:30
us_loop_t::is_default
int is_default
Definition asio.h:33
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
us_internal_callback_t::p
struct us_poll_t p
Definition internal.h:100