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

Loading...
Searching...
No Matches
gcd.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_GCD
23
24/* Loops */
25 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) {
26 struct us_loop_t *loop = (struct us_loop_t *) malloc(sizeof(struct us_loop_t) + ext_size);
27
28 // init the queue from hint
29
30 us_internal_loop_data_init(loop, wakeup_cb, pre_cb, post_cb);
31
32 // why not call integrate here? instead of in run
33
34 return loop;
35}
36
37 void us_loop_free(struct us_loop_t *loop) {
38 us_internal_loop_data_free(loop);
39
40 // free queue if different from main
41
42 free(loop);
43}
44
45/* We don't actually need to include CoreFoundation as we only need one single function,
46 * It will be up to the user to link to CoreFoundation, however that should be automatic in most use cases */
47 extern void CFRunLoopRun();
48
49 void us_loop_run(struct us_loop_t *loop) {
50 us_loop_integrate(loop);
51
52 /* We are absolutely not compatible with dispatch_main,
53 * However every real application should run with CoreFoundation,
54 * Foundation or Cocoa as the main loop, driving dispatch anyways */
55 CFRunLoopRun();
56
57 /* I guess "fallthrough" polls should be added to another run mode than the default one to fall through */
58}
59
60 void gcd_read_handler(void *p) {
61 us_internal_dispatch_ready_poll((struct us_poll_t *) p, 0, LIBUS_SOCKET_READABLE);
62}
63
64 void gcd_write_handler(void *p) {
65 us_internal_dispatch_ready_poll((struct us_poll_t *) p, 0, LIBUS_SOCKET_WRITABLE);
66}
67
68/* Polls */
69 void us_poll_init(struct us_poll_t *p, LIBUS_SOCKET_DESCRIPTOR fd, int poll_type) {
70 p->poll_type = poll_type;
71 p->fd = fd;
72
73 /* I guess these are already activated? */
74 p->gcd_read = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, p->fd, 0, dispatch_get_main_queue());
75 dispatch_set_context(p->gcd_read, p);
76 dispatch_source_set_event_handler_f(p->gcd_read, gcd_read_handler);
77 dispatch_source_set_cancel_handler_f(p->gcd_read, gcd_read_handler);
78
79 p->gcd_write = dispatch_source_create(DISPATCH_SOURCE_TYPE_WRITE, p->fd, 0, dispatch_get_main_queue());
80 dispatch_set_context(p->gcd_write, p);
81 dispatch_source_set_event_handler_f(p->gcd_write, gcd_write_handler);
82 dispatch_source_set_cancel_handler_f(p->gcd_write, gcd_write_handler);
83}
84
85 void us_poll_free(struct us_poll_t *p, struct us_loop_t *loop) {
86 /* It is program error to release suspended filters */
87 us_poll_change(p, loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE);
88 dispatch_release(p->gcd_read);
89 dispatch_release(p->gcd_write);
90 free(p);
91}
92
93 void us_poll_start(struct us_poll_t *p, struct us_loop_t *loop, int events) {
94 p->events = events;
95
96 if (events & LIBUS_SOCKET_READABLE) {
97 dispatch_resume(p->gcd_read);
98 }
99
100 if (events & LIBUS_SOCKET_WRITABLE) {
101 dispatch_resume(p->gcd_write);
102 }
103}
104
105 void us_poll_change(struct us_poll_t *p, struct us_loop_t *loop, int events) {
106 int old_events = p->events;
107
108 if ((old_events & LIBUS_SOCKET_READABLE) != (events & LIBUS_SOCKET_READABLE)) {
109 if (old_events & LIBUS_SOCKET_READABLE) {
110 dispatch_suspend(p->gcd_read);
111 } else {
112 dispatch_resume(p->gcd_read);
113 }
114 }
115
116 if ((old_events & LIBUS_SOCKET_WRITABLE) != (events & LIBUS_SOCKET_WRITABLE)) {
117 if (old_events & LIBUS_SOCKET_WRITABLE) {
118 dispatch_suspend(p->gcd_write);
119 } else {
120 dispatch_resume(p->gcd_write);
121 }
122 }
123
124 p->events = events;
125}
126
127 void us_poll_stop(struct us_poll_t *p, struct us_loop_t *loop) {
128 if (p->events & LIBUS_SOCKET_READABLE) {
129 dispatch_suspend(p->gcd_read);
130 }
131
132 if (p->events & LIBUS_SOCKET_WRITABLE) {
133 dispatch_suspend(p->gcd_write);
134 }
135
136 p->events = 0;
137}
138
139 int us_poll_events(struct us_poll_t *p) {
140 return p->events;
141}
142
143 void *us_poll_ext(struct us_poll_t *p) {
144 return p + 1;
145}
146
147 unsigned int us_internal_accept_poll_event(struct us_poll_t *p) {
148 //printf("us_internal_accept_poll_event\n");
149 return 0;
150}
151
152 int us_internal_poll_type(struct us_poll_t *p) {
153 return p->poll_type & 3;
154}
155
156 void us_internal_poll_set_type(struct us_poll_t *p, int poll_type) {
157 p->poll_type = poll_type;
158}
159
160 LIBUS_SOCKET_DESCRIPTOR us_poll_fd(struct us_poll_t *p) {
161 return p->fd;
162}
163
164 struct us_poll_t *us_create_poll(struct us_loop_t *loop, int fallthrough, unsigned int ext_size) {
165 struct us_poll_t *poll = (struct us_poll_t *) malloc(sizeof(struct us_poll_t) + ext_size);
166
167 return poll;
168}
169
170 struct us_poll_t *us_poll_resize(struct us_poll_t *p, struct us_loop_t *loop, unsigned int ext_size) {
171 int events = us_poll_events(p);
172
173 struct us_poll_t *new_p = realloc(p, sizeof(struct us_poll_t) + ext_size + 1024);
174 if (p != new_p) {
175 /* It is a program error to release suspended filters */
176 us_poll_change(new_p, loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE);
177 dispatch_release(new_p->gcd_read);
178 dispatch_release(new_p->gcd_write);
179
180 /* Create and start new filters */
181 us_poll_init(new_p, us_poll_fd(new_p), us_internal_poll_type(new_p));
182 us_poll_start(new_p, loop, events);
183 }
184
185 return new_p;
186}
187
188/* Timers */
189 void gcd_timer_handler(void *t) {
190 struct us_internal_callback_t *internal_cb = (struct us_internal_callback_t *) t;
191
192 internal_cb->cb(t);
193}
194
195 struct us_timer_t *us_create_timer(struct us_loop_t *loop, int fallthrough, unsigned int ext_size) {
196 struct us_internal_callback_t *cb = malloc(sizeof(struct us_internal_callback_t) + sizeof(dispatch_source_t) + ext_size);
197
198 cb->loop = loop;
199 cb->cb_expects_the_loop = 0;
200 cb->leave_poll_ready = 0;
201
202 dispatch_source_t *gcd_timer = (dispatch_source_t *) (cb + 1);
203
204 *gcd_timer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, dispatch_get_main_queue());
205 dispatch_source_set_event_handler_f(*gcd_timer, gcd_timer_handler);
206 dispatch_set_context(*gcd_timer, cb);
207
208 if (fallthrough) {
209 //uv_unref((uv_handle_t *) uv_timer);
210 }
211
212 return (struct us_timer_t *) cb;
213}
214
215 void *us_timer_ext(struct us_timer_t *timer) {
216 struct us_internal_callback_t *cb = (struct us_internal_callback_t *) timer;
217
218 return (cb + 1);
219}
220
221 void us_timer_close(struct us_timer_t *t) {
222
223}
224
225 void us_timer_set(struct us_timer_t *t, void (*cb)(struct us_timer_t *t), int ms, int repeat_ms) {
226 struct us_internal_callback_t *internal_cb = (struct us_internal_callback_t *) t;
227
228 internal_cb->cb = (void(*)(struct us_internal_callback_t *)) cb;
229
230 dispatch_source_t *gcd_timer = (dispatch_source_t *) (internal_cb + 1);
231 uint64_t nanos = (uint64_t)ms * 1000000;
232 dispatch_source_set_timer(*gcd_timer, 0, nanos, 0);
233 dispatch_activate(*gcd_timer);
234}
235
236 struct us_loop_t *us_timer_loop(struct us_timer_t *t) {
237 struct us_internal_callback_t *internal_cb = (struct us_internal_callback_t *) t;
238
239 return internal_cb->loop;
240}
241
242/* Asyncs */
243 void async_handler(void *c) {
244 struct us_internal_callback_t *internal_cb = (struct us_internal_callback_t *) c;
245
246 internal_cb->cb((struct us_internal_callback_t *) internal_cb->loop);
247}
248
249 struct us_internal_async *us_internal_create_async(struct us_loop_t *loop, int fallthrough, unsigned int ext_size) {
250 struct us_internal_callback_t *cb = malloc(sizeof(struct us_internal_callback_t) + ext_size);
251
252 cb->loop = loop;
253 cb->cb_expects_the_loop = 1;
254 cb->leave_poll_ready = 0;
255
256 if (fallthrough) {
257 //uv_unref((uv_handle_t *) uv_timer);
258 }
259
260 return (struct us_internal_async *) cb;
261}
262
263 void us_internal_async_close(struct us_internal_async *a) {
264 // cancel? free?
265}
266
267 void us_internal_async_set(struct us_internal_async *a, void (*cb)(struct us_internal_async *)) {
268 struct us_internal_callback_t *internal_cb = (struct us_internal_callback_t *) a;
269
270 internal_cb->cb = (void (*)(struct us_internal_callback_t *)) cb;
271}
272
273 void us_internal_async_wakeup(struct us_internal_async *a) {
274 // will probably need to track in-flight work item and cancel in close
275 dispatch_async_f(dispatch_get_main_queue(), a, async_handler);
276}
277
278#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_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_dispatch_ready_poll
void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int events)
Definition loop.c:195
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_ext
void * us_poll_ext(struct us_poll_t *p)
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::fd
int fd
Definition epoll_kqueue.h:50
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::gcd_read
dispatch_source_t gcd_read
Definition gcd.h:35
us_poll_t::gcd_write
dispatch_source_t gcd_write
Definition gcd.h:35
us_poll_t::poll_type
unsigned char poll_type
Definition asio.h:41
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