bac08c9f9effd984c6297da04af9211d2e5c8ba3
[ccan] / ccan / io / io.c
1 /* Licensed under LGPLv2.1+ - see LICENSE file for details */
2 #include "io.h"
3 #include "backend.h"
4 #include <sys/types.h>
5 #include <sys/socket.h>
6 #include <netdb.h>
7 #include <string.h>
8 #include <errno.h>
9 #include <stdlib.h>
10 #include <assert.h>
11 #include <unistd.h>
12 #include <fcntl.h>
13
14 void *io_loop_return;
15
16 struct io_alloc io_alloc = {
17         malloc, realloc, free
18 };
19
20 #ifdef DEBUG
21 /* Set to skip the next plan. */
22 bool io_plan_nodebug;
23 /* The current connection to apply plan to. */
24 struct io_conn *current;
25 /* User-defined function to select which connection(s) to debug. */
26 bool (*io_debug_conn)(struct io_conn *conn);
27
28 struct io_plan io_debug(struct io_plan plan)
29 {
30         struct io_conn *ready = NULL;
31
32         if (io_plan_nodebug) {
33                 io_plan_nodebug = false;
34                 return plan;
35         }
36
37         if (!current || !doing_debug_on(current))
38                 return plan;
39
40         current->plan = plan;
41         backend_plan_changed(current);
42
43         /* Call back into the loop immediately. */
44         io_loop_return = do_io_loop(&ready);
45
46         if (ready) {
47                 set_current(ready);
48                 if (!ready->plan.next) {
49                         /* Call finish function immediately. */
50                         if (ready->finish) {
51                                 errno = ready->plan.u1.s;
52                                 ready->finish(ready, ready->finish_arg);
53                                 ready->finish = NULL;
54                         }
55                         backend_del_conn(ready);
56                 } else {
57                         /* Calls back in itself, via io_debug_io(). */
58                         if (ready->plan.io(ready->fd.fd, &ready->plan) != 2)
59                                 abort();
60                 }
61                 set_current(NULL);
62         }
63
64         /* Return a do-nothing plan, so backend_plan_changed in
65          * io_ready doesn't do anything (it's already been called). */
66         return io_wait_(NULL, (void *)1, NULL);
67 }
68
69 int io_debug_io(int ret)
70 {
71         /* Cache it for debugging; current changes. */
72         struct io_conn *conn = current;
73         int saved_errno = errno;
74
75         if (!doing_debug_on(conn))
76                 return ret;
77
78         /* These will all go linearly through the io_debug() path above. */
79         switch (ret) {
80         case -1:
81                 /* This will call io_debug above. */
82                 errno = saved_errno;
83                 io_close();
84                 break;
85         case 0: /* Keep going with plan. */
86                 io_debug(conn->plan);
87                 break;
88         case 1: /* Done: get next plan. */
89                 if (timeout_active(conn))
90                         backend_del_timeout(conn);
91                 /* In case they call io_duplex, clear our poll flags so
92                  * both sides don't seem to be both doing read or write
93                  * (See assert(!mask || pfd->events != mask) in poll.c) */
94                 conn->plan.pollflag = 0;
95                 conn->plan.next(conn, conn->plan.next_arg);
96                 break;
97         default:
98                 abort();
99         }
100
101         /* Normally-invalid value, used for sanity check. */
102         return 2;
103 }
104
105 /* Counterpart to io_plan_no_debug(), called in macros in io.h */
106 static void io_plan_debug_again(void)
107 {
108         io_plan_nodebug = false;
109 }
110 #else
111 static void io_plan_debug_again(void)
112 {
113 }
114 #endif
115
116 struct io_listener *io_new_listener_(int fd,
117                                      void (*init)(int fd, void *arg),
118                                      void *arg)
119 {
120         struct io_listener *l = io_alloc.alloc(sizeof(*l));
121
122         if (!l)
123                 return NULL;
124
125         l->fd.listener = true;
126         l->fd.fd = fd;
127         l->init = init;
128         l->arg = arg;
129         if (!add_listener(l)) {
130                 io_alloc.free(l);
131                 return NULL;
132         }
133         return l;
134 }
135
136 void io_close_listener(struct io_listener *l)
137 {
138         close(l->fd.fd);
139         del_listener(l);
140         io_alloc.free(l);
141 }
142
143 struct io_conn *io_new_conn_(int fd, struct io_plan plan)
144 {
145         struct io_conn *conn = io_alloc.alloc(sizeof(*conn));
146
147         io_plan_debug_again();
148
149         if (!conn)
150                 return NULL;
151
152         conn->fd.listener = false;
153         conn->fd.fd = fd;
154         conn->plan = plan;
155         conn->finish = NULL;
156         conn->finish_arg = NULL;
157         conn->duplex = NULL;
158         conn->timeout = NULL;
159         if (!add_conn(conn)) {
160                 io_alloc.free(conn);
161                 return NULL;
162         }
163         return conn;
164 }
165
166 void io_set_finish_(struct io_conn *conn,
167                     void (*finish)(struct io_conn *, void *),
168                     void *arg)
169 {
170         conn->finish = finish;
171         conn->finish_arg = arg;
172 }
173
174 struct io_conn *io_duplex_(struct io_conn *old, struct io_plan plan)
175 {
176         struct io_conn *conn;
177
178         io_plan_debug_again();
179
180         assert(!old->duplex);
181
182         conn = io_alloc.alloc(sizeof(*conn));
183         if (!conn)
184                 return NULL;
185
186         conn->fd.listener = false;
187         conn->fd.fd = old->fd.fd;
188         conn->plan = plan;
189         conn->duplex = old;
190         conn->finish = NULL;
191         conn->finish_arg = NULL;
192         conn->timeout = NULL;
193         if (!add_duplex(conn)) {
194                 io_alloc.free(conn);
195                 return NULL;
196         }
197         old->duplex = conn;
198         return conn;
199 }
200
201 bool io_timeout_(struct io_conn *conn, struct timerel t,
202                  struct io_plan (*cb)(struct io_conn *, void *), void *arg)
203 {
204         assert(cb);
205
206         if (!conn->timeout) {
207                 conn->timeout = io_alloc.alloc(sizeof(*conn->timeout));
208                 if (!conn->timeout)
209                         return false;
210         } else
211                 assert(!timeout_active(conn));
212
213         conn->timeout->next = cb;
214         conn->timeout->next_arg = arg;
215         backend_add_timeout(conn, t);
216         return true;
217 }
218
219 /* Always done: call the next thing. */
220 static int do_always(int fd, struct io_plan *plan)
221 {
222         return 1;
223 }
224
225 struct io_plan io_always_(struct io_plan (*cb)(struct io_conn *, void *),
226                           void *arg)
227 {
228         struct io_plan plan;
229
230         assert(cb);
231         plan.io = do_always;
232         plan.next = cb;
233         plan.next_arg = arg;
234         plan.pollflag = POLLALWAYS;
235
236         return plan;
237 }
238
239 /* Returns true if we're finished. */
240 static int do_write(int fd, struct io_plan *plan)
241 {
242         ssize_t ret = write(fd, plan->u1.cp, plan->u2.s);
243         if (ret < 0)
244                 return io_debug_io(-1);
245
246         plan->u1.cp += ret;
247         plan->u2.s -= ret;
248         return io_debug_io(plan->u2.s == 0);
249 }
250
251 /* Queue some data to be written. */
252 struct io_plan io_write_(const void *data, size_t len,
253                          struct io_plan (*cb)(struct io_conn *, void *),
254                          void *arg)
255 {
256         struct io_plan plan;
257
258         assert(cb);
259
260         if (len == 0)
261                 return io_always_(cb, arg);
262
263         plan.u1.const_vp = data;
264         plan.u2.s = len;
265         plan.io = do_write;
266         plan.next = cb;
267         plan.next_arg = arg;
268         plan.pollflag = POLLOUT;
269
270         return plan;
271 }
272
273 static int do_read(int fd, struct io_plan *plan)
274 {
275         ssize_t ret = read(fd, plan->u1.cp, plan->u2.s);
276         if (ret <= 0)
277                 return io_debug_io(-1);
278
279         plan->u1.cp += ret;
280         plan->u2.s -= ret;
281         return io_debug_io(plan->u2.s == 0);
282 }
283
284 /* Queue a request to read into a buffer. */
285 struct io_plan io_read_(void *data, size_t len,
286                         struct io_plan (*cb)(struct io_conn *, void *),
287                         void *arg)
288 {
289         struct io_plan plan;
290
291         assert(cb);
292
293         if (len == 0)
294                 return io_always_(cb, arg);
295
296         plan.u1.cp = data;
297         plan.u2.s = len;
298         plan.io = do_read;
299         plan.next = cb;
300         plan.next_arg = arg;
301
302         plan.pollflag = POLLIN;
303
304         return plan;
305 }
306
307 static int do_read_partial(int fd, struct io_plan *plan)
308 {
309         ssize_t ret = read(fd, plan->u1.cp, *(size_t *)plan->u2.vp);
310         if (ret <= 0)
311                 return io_debug_io(-1);
312
313         *(size_t *)plan->u2.vp = ret;
314         return io_debug_io(1);
315 }
316
317 /* Queue a partial request to read into a buffer. */
318 struct io_plan io_read_partial_(void *data, size_t *len,
319                                 struct io_plan (*cb)(struct io_conn *, void *),
320                                 void *arg)
321 {
322         struct io_plan plan;
323
324         assert(cb);
325
326         if (*len == 0)
327                 return io_always_(cb, arg);
328
329         plan.u1.cp = data;
330         plan.u2.vp = len;
331         plan.io = do_read_partial;
332         plan.next = cb;
333         plan.next_arg = arg;
334         plan.pollflag = POLLIN;
335
336         return plan;
337 }
338
339 static int do_write_partial(int fd, struct io_plan *plan)
340 {
341         ssize_t ret = write(fd, plan->u1.cp, *(size_t *)plan->u2.vp);
342         if (ret < 0)
343                 return io_debug_io(-1);
344
345         *(size_t *)plan->u2.vp = ret;
346         return io_debug_io(1);
347 }
348
349 /* Queue a partial write request. */
350 struct io_plan io_write_partial_(const void *data, size_t *len,
351                                  struct io_plan (*cb)(struct io_conn*, void *),
352                                  void *arg)
353 {
354         struct io_plan plan;
355
356         assert(cb);
357
358         if (*len == 0)
359                 return io_always_(cb, arg);
360
361         plan.u1.const_vp = data;
362         plan.u2.vp = len;
363         plan.io = do_write_partial;
364         plan.next = cb;
365         plan.next_arg = arg;
366         plan.pollflag = POLLOUT;
367
368         return plan;
369 }
370
371 static int already_connected(int fd, struct io_plan *plan)
372 {
373         return io_debug_io(1);
374 }
375
376 static int do_connect(int fd, struct io_plan *plan)
377 {
378         int err, ret;
379         socklen_t len = sizeof(err);
380
381         /* Has async connect finished? */
382         ret = getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len);
383         if (ret < 0)
384                 return -1;
385
386         if (err == 0) {
387                 /* Restore blocking if it was initially. */
388                 fcntl(fd, F_SETFD, plan->u1.s);
389                 return 1;
390         }
391         return 0;
392 }
393
394 struct io_plan io_connect_(int fd, const struct addrinfo *addr,
395                            struct io_plan (*cb)(struct io_conn*, void *),
396                            void *arg)
397 {
398         struct io_plan plan;
399
400         assert(cb);
401
402         plan.next = cb;
403         plan.next_arg = arg;
404
405         /* Save old flags, set nonblock if not already. */
406         plan.u1.s = fcntl(fd, F_GETFL);
407         fcntl(fd, F_SETFL, plan.u1.s | O_NONBLOCK);
408
409         /* Immediate connect can happen. */
410         if (connect(fd, addr->ai_addr, addr->ai_addrlen) == 0) {
411                 /* Dummy will be called immediately. */
412                 plan.pollflag = POLLOUT;
413                 plan.io = already_connected;
414         } else {
415                 if (errno != EINPROGRESS)
416                         return io_close_();
417
418                 plan.pollflag = POLLIN;
419                 plan.io = do_connect;
420         }
421         return plan;
422 }
423
424 struct io_plan io_wait_(const void *wait,
425                         struct io_plan (*cb)(struct io_conn *, void*),
426                         void *arg)
427 {
428         struct io_plan plan;
429
430         assert(cb);
431         plan.pollflag = 0;
432         plan.io = NULL;
433         plan.next = cb;
434         plan.next_arg = arg;
435
436         plan.u1.const_vp = wait;
437
438         return plan;
439 }
440
441 void io_wake(const void *wait)
442 {
443         backend_wait_changed(wait);
444 }
445
446 void io_ready(struct io_conn *conn)
447 {
448         /* Beware io_close_other! */
449         if (!conn->plan.next)
450                 return;
451
452         set_current(conn);
453         switch (conn->plan.io(conn->fd.fd, &conn->plan)) {
454         case -1: /* Failure means a new plan: close up. */
455                 conn->plan = io_close();
456                 backend_plan_changed(conn);
457                 break;
458         case 0: /* Keep going with plan. */
459                 break;
460         case 1: /* Done: get next plan. */
461                 if (timeout_active(conn))
462                         backend_del_timeout(conn);
463                 /* In case they call io_duplex, clear our poll flags so
464                  * both sides don't seem to be both doing read or write
465                  * (See assert(!mask || pfd->events != mask) in poll.c) */
466                 conn->plan.pollflag = 0;
467                 conn->plan = conn->plan.next(conn, conn->plan.next_arg);
468                 backend_plan_changed(conn);
469         }
470         set_current(NULL);
471 }
472
473 /* Close the connection, we're done. */
474 struct io_plan io_close_(void)
475 {
476         struct io_plan plan;
477
478         plan.pollflag = 0;
479         /* This means we're closing. */
480         plan.next = NULL;
481         plan.u1.s = errno;
482
483         return plan;
484 }
485
486 struct io_plan io_close_cb(struct io_conn *conn, void *arg)
487 {
488         return io_close();
489 }
490
491 void io_close_other(struct io_conn *conn)
492 {
493         /* Don't close if already closing! */
494         if (conn->plan.next) {
495                 conn->plan = io_close_();
496                 backend_plan_changed(conn);
497         }
498 }
499
500 /* Exit the loop, returning this (non-NULL) arg. */
501 struct io_plan io_break_(void *ret, struct io_plan plan)
502 {
503         io_plan_debug_again();
504
505         assert(ret);
506         io_loop_return = ret;
507
508         return plan;
509 }
510
511 static struct io_plan io_never_called(struct io_conn *conn, void *arg)
512 {
513         abort();
514 }
515
516 struct io_plan io_never(void)
517 {
518         return io_always_(io_never_called, NULL);
519 }
520
521 int io_conn_fd(const struct io_conn *conn)
522 {
523         return conn->fd.fd;
524 }
525
526 void io_set_alloc(void *(*allocfn)(size_t size),
527                   void *(*reallocfn)(void *ptr, size_t size),
528                   void (*freefn)(void *ptr))
529 {
530         io_alloc.alloc = allocfn;
531         io_alloc.realloc = reallocfn;
532         io_alloc.free = freefn;
533 }