ccan/io: io_duplex.
[ccan] / ccan / io / io.c
1 /* Licensed under BSD-MIT - 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
12 void *io_loop_return;
13
14 struct io_listener *io_new_listener_(int fd,
15                                      struct io_op *(*start)(struct io_conn *,
16                                                             void *arg),
17                                      void (*finish)(struct io_conn *, void *),
18                                      void *arg)
19 {
20         struct io_listener *l = malloc(sizeof(*l));
21
22         if (!l)
23                 return NULL;
24
25         l->fd.listener = true;
26         l->fd.fd = fd;
27         l->fd.next = start;
28         l->fd.finish = finish;
29         l->fd.finish_arg = l->fd.next_arg = arg;
30         if (!add_listener(l)) {
31                 free(l);
32                 return NULL;
33         }
34         return l;
35 }
36
37 void io_close_listener(struct io_listener *l)
38 {
39         close(l->fd.fd);
40         del_listener(l);
41         free(l);
42 }
43
44 struct io_conn *io_new_conn_(int fd,
45                              struct io_op *(*start)(struct io_conn *, void *),
46                              void (*finish)(struct io_conn *, void *),
47                              void *arg)
48 {
49         struct io_conn *conn = malloc(sizeof(*conn));
50
51         if (!conn)
52                 return NULL;
53
54         conn->fd.listener = false;
55         conn->fd.fd = fd;
56         conn->fd.next = start;
57         conn->fd.finish = finish;
58         conn->fd.finish_arg = conn->fd.next_arg = arg;
59         conn->state = NEXT;
60         conn->duplex = NULL;
61         if (!add_conn(conn)) {
62                 free(conn);
63                 return NULL;
64         }
65         return conn;
66 }
67
68 struct io_conn *io_duplex_(struct io_conn *old,
69                              struct io_op *(*start)(struct io_conn *, void *),
70                              void (*finish)(struct io_conn *, void *),
71                              void *arg)
72 {
73         struct io_conn *conn;
74
75         assert(!old->duplex);
76
77         conn = malloc(sizeof(*conn));
78         if (!conn)
79                 return NULL;
80
81         conn->fd.listener = false;
82         conn->fd.fd = old->fd.fd;
83         conn->fd.next = start;
84         conn->fd.finish = finish;
85         conn->fd.finish_arg = conn->fd.next_arg = arg;
86         conn->state = NEXT;
87         conn->duplex = old;
88         if (!add_duplex(conn)) {
89                 free(conn);
90                 return NULL;
91         }
92         old->duplex = conn;
93         return conn;
94 }
95
96 /* Convenient token which only we can produce. */
97 static inline struct io_next *to_ionext(struct io_conn *conn)
98 {
99         return (struct io_next *)conn;
100 }
101
102 static inline struct io_op *to_ioop(enum io_state state)
103 {
104         return (struct io_op *)(long)state;
105 }
106
107 static inline struct io_conn *from_ionext(struct io_next *next)
108 {
109         return (struct io_conn *)next;
110 }
111
112 struct io_next *io_next_(struct io_conn *conn,
113                          struct io_op *(*next)(struct io_conn *, void *),
114                          void *arg)
115 {
116         conn->fd.next = next;
117         conn->fd.next_arg = arg;
118
119         return to_ionext(conn);
120 }
121
122 /* Queue some data to be written. */
123 struct io_op *io_write(const void *data, size_t len, struct io_next *next)
124 {
125         struct io_conn *conn = from_ionext(next);
126         conn->u.write.buf = data;
127         conn->u.write.len = len;
128         return to_ioop(WRITE);
129 }
130
131 /* Queue a request to read into a buffer. */
132 struct io_op *io_read(void *data, size_t len, struct io_next *next)
133 {
134         struct io_conn *conn = from_ionext(next);
135         conn->u.read.buf = data;
136         conn->u.read.len = len;
137         return to_ioop(READ);
138 }
139
140 /* Queue a partial request to read into a buffer. */
141 struct io_op *io_read_partial(void *data, size_t *len, struct io_next *next)
142 {
143         struct io_conn *conn = from_ionext(next);
144         conn->u.readpart.buf = data;
145         conn->u.readpart.lenp = len;
146         return to_ioop(READPART);
147 }
148
149 /* Queue a partial write request. */
150 struct io_op *io_write_partial(const void *data, size_t *len, struct io_next *next)
151 {
152         struct io_conn *conn = from_ionext(next);
153         conn->u.writepart.buf = data;
154         conn->u.writepart.lenp = len;
155         return to_ioop(WRITEPART);
156 }
157
158 struct io_op *io_idle(struct io_conn *conn)
159 {
160         return to_ioop(IDLE);
161 }
162
163 void io_wake_(struct io_conn *conn,
164               struct io_op *(*next)(struct io_conn *, void *), void *arg)
165
166 {
167         /* It might have finished, but we haven't called its finish() yet. */
168         if (conn->state == FINISHED)
169                 return;
170         assert(conn->state == IDLE);
171         conn->fd.next = next;
172         conn->fd.next_arg = arg;
173         backend_set_state(conn, to_ioop(NEXT));
174 }
175
176 static struct io_op *do_next(struct io_conn *conn)
177 {
178         return conn->fd.next(conn, conn->fd.next_arg);
179 }
180
181 struct io_op *do_ready(struct io_conn *conn)
182 {
183         ssize_t ret;
184         bool finished;
185
186         switch (conn->state) {
187         case WRITE:
188                 ret = write(conn->fd.fd, conn->u.write.buf, conn->u.write.len);
189                 if (ret < 0)
190                         return io_close(conn, NULL);
191                 conn->u.write.buf += ret;
192                 conn->u.write.len -= ret;
193                 finished = (conn->u.write.len == 0);
194                 break;
195         case WRITEPART:
196                 ret = write(conn->fd.fd, conn->u.writepart.buf,
197                             *conn->u.writepart.lenp);
198                 if (ret < 0)
199                         return io_close(conn, NULL);
200                 *conn->u.writepart.lenp = ret;
201                 finished = true;
202                 break;
203         case READ:
204                 ret = read(conn->fd.fd, conn->u.read.buf, conn->u.read.len);
205                 if (ret <= 0)
206                         return io_close(conn, NULL);
207                 conn->u.read.buf += ret;
208                 conn->u.read.len -= ret;
209                 finished = (conn->u.read.len == 0);
210                 break;
211         case READPART:
212                 ret = read(conn->fd.fd, conn->u.readpart.buf,
213                             *conn->u.readpart.lenp);
214                 if (ret <= 0)
215                         return io_close(conn, NULL);
216                 *conn->u.readpart.lenp = ret;
217                 finished = true;
218                 break;
219         default:
220                 /* Shouldn't happen. */
221                 abort();
222         }
223
224         if (finished)
225                 return do_next(conn);
226         return to_ioop(conn->state);
227 }
228
229 /* Useful next functions. */
230 /* Close the connection, we're done. */
231 struct io_op *io_close(struct io_conn *conn, void *arg)
232 {
233         return to_ioop(FINISHED);
234 }
235
236 /* Exit the loop, returning this (non-NULL) arg. */
237 struct io_op *io_break(void *arg, struct io_next *next)
238 {
239         io_loop_return = arg;
240
241         return to_ioop(NEXT);
242 }