cc2b1afa3b27720adff4e3aa0b0c1d4fd4a393a3
[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 #include <ccan/container_of/container_of.h>
14
15 void *io_loop_return;
16
17 struct io_plan io_conn_freed;
18
19 struct io_listener *io_new_listener_(const tal_t *ctx, int fd,
20                                      struct io_plan *(*init)(struct io_conn *,
21                                                              void *),
22                                      void *arg)
23 {
24         struct io_listener *l = tal(ctx, struct io_listener);
25         if (!l)
26                 return NULL;
27
28         l->fd.listener = true;
29         l->fd.fd = fd;
30         l->init = init;
31         l->arg = arg;
32         l->ctx = ctx;
33         if (!add_listener(l))
34                 return tal_free(l);
35         return l;
36 }
37
38 void io_close_listener(struct io_listener *l)
39 {
40         tal_free(l);
41 }
42
43 static struct io_plan *io_never_called(struct io_conn *conn, void *arg)
44 {
45         abort();
46 }
47
48 /* Returns false if conn was freed. */
49 static bool next_plan(struct io_conn *conn, struct io_plan *plan)
50 {
51         struct io_plan *(*next)(struct io_conn *, void *arg);
52
53         next = plan->next;
54
55         plan->status = IO_UNSET;
56         plan->io = NULL;
57         plan->next = io_never_called;
58
59         plan = next(conn, plan->next_arg);
60
61         if (plan == &io_conn_freed)
62                 return false;
63
64         /* It should have set a plan inside this conn (or duplex) */
65         assert(plan == &conn->plan[IO_IN]
66                || plan == &conn->plan[IO_OUT]
67                || plan == &conn->plan[2]);
68         assert(conn->plan[IO_IN].status != IO_UNSET
69                || conn->plan[IO_OUT].status != IO_UNSET);
70
71         backend_new_plan(conn);
72         return true;
73 }
74
75 bool io_fd_block(int fd, bool block)
76 {
77         int flags = fcntl(fd, F_GETFL);
78
79         if (flags == -1)
80                 return false;
81
82         if (block)
83                 flags &= ~O_NONBLOCK;
84         else
85                 flags |= O_NONBLOCK;
86
87         return fcntl(fd, F_SETFL, flags) != -1;
88 }
89
90 struct io_conn *io_new_conn_(const tal_t *ctx, int fd,
91                              struct io_plan *(*init)(struct io_conn *, void *),
92                              void *arg)
93 {
94         struct io_conn *conn = tal(ctx, struct io_conn);
95
96         if (!conn)
97                 return NULL;
98
99         conn->fd.listener = false;
100         conn->fd.fd = fd;
101         conn->finish = NULL;
102         conn->finish_arg = NULL;
103         list_node_init(&conn->always);
104
105         if (!add_conn(conn))
106                 return tal_free(conn);
107
108         /* Keep our I/O async. */
109         io_fd_block(fd, false);
110
111         /* We start with out doing nothing, and in doing our init. */
112         conn->plan[IO_OUT].status = IO_UNSET;
113
114         conn->plan[IO_IN].next = init;
115         conn->plan[IO_IN].next_arg = arg;
116         if (!next_plan(conn, &conn->plan[IO_IN]))
117                 return NULL;
118
119         return conn;
120 }
121
122 void io_set_finish_(struct io_conn *conn,
123                     void (*finish)(struct io_conn *, void *),
124                     void *arg)
125 {
126         conn->finish = finish;
127         conn->finish_arg = arg;
128 }
129
130 struct io_plan_arg *io_plan_arg(struct io_conn *conn, enum io_direction dir)
131 {
132         assert(conn->plan[dir].status == IO_UNSET);
133
134         conn->plan[dir].status = IO_POLLING;
135         return &conn->plan[dir].arg;
136 }
137
138 static struct io_plan *set_always(struct io_conn *conn,
139                                   enum io_direction dir,
140                                   struct io_plan *(*next)(struct io_conn *,
141                                                           void *),
142                                   void *arg)
143 {
144         struct io_plan *plan = &conn->plan[dir];
145
146         plan->status = IO_ALWAYS;
147         backend_new_always(conn);
148         return io_set_plan(conn, dir, NULL, next, arg);
149 }
150
151 static struct io_plan *io_always_dir(struct io_conn *conn,
152                                      enum io_direction dir,
153                                      struct io_plan *(*next)(struct io_conn *,
154                                                              void *),
155                                      void *arg)
156 {
157         return set_always(conn, dir, next, arg);
158 }
159
160 struct io_plan *io_always_(struct io_conn *conn,
161                            struct io_plan *(*next)(struct io_conn *, void *),
162                            void *arg)
163 {
164         return io_always_dir(conn, IO_IN, next, arg);
165 }
166
167 struct io_plan *io_out_always_(struct io_conn *conn,
168                                struct io_plan *(*next)(struct io_conn *,
169                                                        void *),
170                                void *arg)
171 {
172         return io_always_dir(conn, IO_OUT, next, arg);
173 }
174
175 static int do_write(int fd, struct io_plan_arg *arg)
176 {
177         ssize_t ret = write(fd, arg->u1.cp, arg->u2.s);
178         if (ret < 0)
179                 return -1;
180
181         arg->u1.cp += ret;
182         arg->u2.s -= ret;
183         return arg->u2.s == 0;
184 }
185
186 /* Queue some data to be written. */
187 struct io_plan *io_write_(struct io_conn *conn, const void *data, size_t len,
188                           struct io_plan *(*next)(struct io_conn *, void *),
189                           void *next_arg)
190 {
191         struct io_plan_arg *arg = io_plan_arg(conn, IO_OUT);
192
193         if (len == 0)
194                 return set_always(conn, IO_OUT, next, next_arg);
195
196         arg->u1.const_vp = data;
197         arg->u2.s = len;
198
199         return io_set_plan(conn, IO_OUT, do_write, next, next_arg);
200 }
201
202 static int do_read(int fd, struct io_plan_arg *arg)
203 {
204         ssize_t ret = read(fd, arg->u1.cp, arg->u2.s);
205         if (ret <= 0)
206                 return -1;
207
208         arg->u1.cp += ret;
209         arg->u2.s -= ret;
210         return arg->u2.s == 0;
211 }
212
213 /* Queue a request to read into a buffer. */
214 struct io_plan *io_read_(struct io_conn *conn,
215                          void *data, size_t len,
216                          struct io_plan *(*next)(struct io_conn *, void *),
217                          void *next_arg)
218 {
219         struct io_plan_arg *arg = io_plan_arg(conn, IO_IN);
220
221         if (len == 0)
222                 return set_always(conn, IO_IN, next, next_arg);
223
224         arg->u1.cp = data;
225         arg->u2.s = len;
226
227         return io_set_plan(conn, IO_IN, do_read, next, next_arg);
228 }
229
230 static int do_read_partial(int fd, struct io_plan_arg *arg)
231 {
232         ssize_t ret = read(fd, arg->u1.cp, *(size_t *)arg->u2.vp);
233         if (ret <= 0)
234                 return -1;
235
236         *(size_t *)arg->u2.vp = ret;
237         return 1;
238 }
239
240 /* Queue a partial request to read into a buffer. */
241 struct io_plan *io_read_partial_(struct io_conn *conn,
242                                  void *data, size_t maxlen, size_t *len,
243                                  struct io_plan *(*next)(struct io_conn *,
244                                                          void *),
245                                  void *next_arg)
246 {
247         struct io_plan_arg *arg = io_plan_arg(conn, IO_IN);
248
249         if (maxlen == 0)
250                 return set_always(conn, IO_IN, next, next_arg);
251
252         arg->u1.cp = data;
253         /* We store the max len in here temporarily. */
254         *len = maxlen;
255         arg->u2.vp = len;
256
257         return io_set_plan(conn, IO_IN, do_read_partial, next, next_arg);
258 }
259
260 static int do_write_partial(int fd, struct io_plan_arg *arg)
261 {
262         ssize_t ret = write(fd, arg->u1.cp, *(size_t *)arg->u2.vp);
263         if (ret < 0)
264                 return -1;
265
266         *(size_t *)arg->u2.vp = ret;
267         return 1;
268 }
269
270 /* Queue a partial write request. */
271 struct io_plan *io_write_partial_(struct io_conn *conn,
272                                   const void *data, size_t maxlen, size_t *len,
273                                   struct io_plan *(*next)(struct io_conn *,
274                                                           void*),
275                                   void *next_arg)
276 {
277         struct io_plan_arg *arg = io_plan_arg(conn, IO_OUT);
278
279         if (maxlen == 0)
280                 return set_always(conn, IO_OUT, next, next_arg);
281
282         arg->u1.const_vp = data;
283         /* We store the max len in here temporarily. */
284         *len = maxlen;
285         arg->u2.vp = len;
286
287         return io_set_plan(conn, IO_OUT, do_write_partial, next, next_arg);
288 }
289
290 static int do_connect(int fd, struct io_plan_arg *arg)
291 {
292         int err, ret;
293         socklen_t len = sizeof(err);
294
295         /* Has async connect finished? */
296         ret = getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len);
297         if (ret < 0)
298                 return -1;
299
300         if (err == 0) {
301                 return 1;
302         } else if (err == EINPROGRESS)
303                 return 0;
304
305         errno = err;
306         return -1;
307 }
308
309 struct io_plan *io_connect_(struct io_conn *conn, const struct addrinfo *addr,
310                             struct io_plan *(*next)(struct io_conn *, void *),
311                             void *next_arg)
312 {
313         int fd = io_conn_fd(conn);
314
315         /* We don't actually need the arg, but we need it polling. */
316         io_plan_arg(conn, IO_OUT);
317
318         /* Note that io_new_conn() will make fd O_NONBLOCK */
319
320         /* Immediate connect can happen. */
321         if (connect(fd, addr->ai_addr, addr->ai_addrlen) == 0)
322                 return set_always(conn, IO_OUT, next, next_arg);
323
324         if (errno != EINPROGRESS)
325                 return io_close(conn);
326
327         return io_set_plan(conn, IO_OUT, do_connect, next, next_arg);
328 }
329
330 static struct io_plan *io_wait_dir(struct io_conn *conn,
331                                    const void *wait,
332                                    enum io_direction dir,
333                                    struct io_plan *(*next)(struct io_conn *,
334                                                            void *),
335                                    void *next_arg)
336 {
337         struct io_plan_arg *arg = io_plan_arg(conn, dir);
338         arg->u1.const_vp = wait;
339
340         conn->plan[dir].status = IO_WAITING;
341
342         return io_set_plan(conn, dir, NULL, next, next_arg);
343 }
344
345 struct io_plan *io_wait_(struct io_conn *conn,
346                          const void *wait,
347                          struct io_plan *(*next)(struct io_conn *, void *),
348                          void *next_arg)
349 {
350         return io_wait_dir(conn, wait, IO_IN, next, next_arg);
351 }
352
353 struct io_plan *io_out_wait_(struct io_conn *conn,
354                              const void *wait,
355                              struct io_plan *(*next)(struct io_conn *, void *),
356                              void *next_arg)
357 {
358         return io_wait_dir(conn, wait, IO_OUT, next, next_arg);
359 }
360
361 void io_wake(const void *wait)
362 {
363         backend_wake(wait);
364 }
365
366 /* Returns false if this has been freed. */
367 static bool do_plan(struct io_conn *conn, struct io_plan *plan)
368 {
369         /* We shouldn't have polled for this event if this wasn't true! */
370         assert(plan->status == IO_POLLING);
371
372         switch (plan->io(conn->fd.fd, &plan->arg)) {
373         case -1:
374                 io_close(conn);
375                 return false;
376         case 0:
377                 return true;
378         case 1:
379                 return next_plan(conn, plan);
380         default:
381                 /* IO should only return -1, 0 or 1 */
382                 abort();
383         }
384 }
385
386 void io_ready(struct io_conn *conn, int pollflags)
387 {
388         if (pollflags & POLLIN)
389                 if (!do_plan(conn, &conn->plan[IO_IN]))
390                         return;
391
392         if (pollflags & POLLOUT)
393                 do_plan(conn, &conn->plan[IO_OUT]);
394 }
395
396 void io_do_always(struct io_conn *conn)
397 {
398         /* There's a corner case where the in next_plan wakes up the
399          * out, placing it in IO_ALWAYS and we end up processing it immediately,
400          * only to leave it in the always list.
401          *
402          * Yet we can't just process one, in case they are both supposed
403          * to be done, so grab state beforehand.
404          */
405         bool always_out = (conn->plan[IO_OUT].status == IO_ALWAYS);
406
407         if (conn->plan[IO_IN].status == IO_ALWAYS)
408                 if (!next_plan(conn, &conn->plan[IO_IN]))
409                         return;
410
411         if (always_out) {
412                 /* You can't *unalways* a conn (except by freeing, in which
413                  * case next_plan() returned false */
414                 assert(conn->plan[IO_OUT].status == IO_ALWAYS);
415                 next_plan(conn, &conn->plan[IO_OUT]);
416         }
417 }
418
419 void io_do_wakeup(struct io_conn *conn, enum io_direction dir)
420 {
421         struct io_plan *plan = &conn->plan[dir];
422
423         assert(plan->status == IO_WAITING);
424
425         set_always(conn, dir, plan->next, plan->next_arg);
426 }
427
428 /* Close the connection, we're done. */
429 struct io_plan *io_close(struct io_conn *conn)
430 {
431         tal_free(conn);
432         return &io_conn_freed;
433 }
434
435 struct io_plan *io_close_cb(struct io_conn *conn, void *next_arg)
436 {
437         return io_close(conn);
438 }
439
440 struct io_plan *io_close_taken_fd(struct io_conn *conn)
441 {
442         io_fd_block(conn->fd.fd, true);
443
444         cleanup_conn_without_close(conn);
445         return io_close(conn);
446 }
447
448 /* Exit the loop, returning this (non-NULL) arg. */
449 void io_break(const void *ret)
450 {
451         assert(ret);
452         io_loop_return = (void *)ret;
453 }
454
455 struct io_plan *io_never(struct io_conn *conn, void *unused)
456 {
457         return io_always(conn, io_never_called, NULL);
458 }
459
460 int io_conn_fd(const struct io_conn *conn)
461 {
462         return conn->fd.fd;
463 }
464
465 struct io_plan *io_duplex(struct io_conn *conn,
466                           struct io_plan *in_plan, struct io_plan *out_plan)
467 {
468         assert(conn == container_of(in_plan, struct io_conn, plan[IO_IN]));
469         /* in_plan must be conn->plan[IO_IN], out_plan must be [IO_OUT] */
470         assert(out_plan == in_plan + 1);
471         return out_plan + 1;
472 }
473
474 struct io_plan *io_halfclose(struct io_conn *conn)
475 {
476         /* Both unset?  OK. */
477         if (conn->plan[IO_IN].status == IO_UNSET
478             && conn->plan[IO_OUT].status == IO_UNSET)
479                 return io_close(conn);
480
481         /* We leave this unset then. */
482         if (conn->plan[IO_IN].status == IO_UNSET)
483                 return &conn->plan[IO_IN];
484         else
485                 return &conn->plan[IO_OUT];
486 }
487
488 struct io_plan *io_set_plan(struct io_conn *conn, enum io_direction dir,
489                             int (*io)(int fd, struct io_plan_arg *arg),
490                             struct io_plan *(*next)(struct io_conn *, void *),
491                             void *next_arg)
492 {
493         struct io_plan *plan = &conn->plan[dir];
494
495         plan->io = io;
496         plan->next = next;
497         plan->next_arg = next_arg;
498         assert(next != NULL);
499
500         return plan;
501 }
502
503 bool io_flush_sync(struct io_conn *conn)
504 {
505         struct io_plan *plan = &conn->plan[IO_OUT];
506         bool ok;
507
508         /* Not writing?  Nothing to do. */
509         if (plan->status != IO_POLLING)
510                 return true;
511
512         /* Synchronous please. */
513         io_fd_block(io_conn_fd(conn), true);
514
515 again:
516         switch (plan->io(conn->fd.fd, &plan->arg)) {
517         case -1:
518                 ok = false;
519                 break;
520         /* Incomplete, try again. */
521         case 0:
522                 goto again;
523         case 1:
524                 ok = true;
525                 /* In case they come back. */
526                 set_always(conn, IO_OUT, plan->next, plan->next_arg);
527                 break;
528         default:
529                 /* IO should only return -1, 0 or 1 */
530                 abort();
531         }
532
533         io_fd_block(io_conn_fd(conn), false);
534         return ok;
535 }