git.ozlabs.org Git - ccan/blob - 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 <poll.h>
  12
  13 void *io_loop_return;
  14
  15 struct io_listener *io_new_listener_(int fd,
  16                                      struct io_plan (*start)(struct io_conn *,
  17                                                              void *arg),
  18                                      void (*finish)(struct io_conn *, void *),
  19                                      void *arg)
  20 {
  21         struct io_listener *l = malloc(sizeof(*l));
  22
  23         if (!l)
  24                 return NULL;
  25
  26         l->fd.listener = true;
  27         l->fd.fd = fd;
  28         l->next = start;
  29         l->finish = finish;
  30         l->conn_arg = arg;
  31         if (!add_listener(l)) {
  32                 free(l);
  33                 return NULL;
  34         }
  35         return l;
  36 }
  37
  38 void io_close_listener(struct io_listener *l)
  39 {
  40         close(l->fd.fd);
  41         del_listener(l);
  42         free(l);
  43 }
  44
  45 struct io_conn *io_new_conn_(int fd,
  46                              struct io_plan (*start)(struct io_conn *, void *),
  47                              void (*finish)(struct io_conn *, void *),
  48                              void *arg)
  49 {
  50         struct io_conn *conn = malloc(sizeof(*conn));
  51
  52         if (!conn)
  53                 return NULL;
  54
  55         conn->fd.listener = false;
  56         conn->fd.fd = fd;
  57         conn->plan.next = start;
  58         conn->finish = finish;
  59         conn->finish_arg = conn->plan.next_arg = arg;
  60         conn->plan.pollflag = 0;
  61         conn->plan.state = IO_NEXT;
  62         conn->duplex = NULL;
  63         conn->timeout = NULL;
  64         if (!add_conn(conn)) {
  65                 free(conn);
  66                 return NULL;
  67         }
  68         return conn;
  69 }
  70
  71 struct io_conn *io_duplex_(struct io_conn *old,
  72                              struct io_plan (*start)(struct io_conn *, void *),
  73                              void (*finish)(struct io_conn *, void *),
  74                              void *arg)
  75 {
  76         struct io_conn *conn;
  77
  78         assert(!old->duplex);
  79
  80         conn = malloc(sizeof(*conn));
  81         if (!conn)
  82                 return NULL;
  83
  84         conn->fd.listener = false;
  85         conn->fd.fd = old->fd.fd;
  86         conn->plan.next = start;
  87         conn->finish = finish;
  88         conn->finish_arg = conn->plan.next_arg = arg;
  89         conn->plan.pollflag = 0;
  90         conn->plan.state = IO_NEXT;
  91         conn->duplex = old;
  92         conn->timeout = NULL;
  93         if (!add_duplex(conn)) {
  94                 free(conn);
  95                 return NULL;
  96         }
  97         old->duplex = conn;
  98         return conn;
  99 }
 100
 101 bool io_timeout_(struct io_conn *conn, struct timespec ts,
 102                  struct io_plan (*cb)(struct io_conn *, void *), void *arg)
 103 {
 104         if (!conn->timeout) {
 105                 conn->timeout = malloc(sizeof(*conn->timeout));
 106                 if (!conn->timeout)
 107                         return false;
 108         } else
 109                 assert(!timeout_active(conn));
 110
 111         conn->timeout->next = cb;
 112         conn->timeout->next_arg = arg;
 113         backend_add_timeout(conn, ts);
 114         return true;
 115 }
 116
 117 static enum io_result do_write(struct io_conn *conn)
 118 {
 119         ssize_t ret = write(conn->fd.fd, conn->plan.u.write.buf, conn->plan.u.write.len);
 120         if (ret < 0)
 121                 return RESULT_CLOSE;
 122
 123         conn->plan.u.write.buf += ret;
 124         conn->plan.u.write.len -= ret;
 125         if (conn->plan.u.write.len == 0)
 126                 return RESULT_FINISHED;
 127         else
 128                 return RESULT_AGAIN;
 129 }
 130
 131 /* Queue some data to be written. */
 132 struct io_plan io_write_(const void *data, size_t len,
 133                          struct io_plan (*cb)(struct io_conn *, void *),
 134                          void *arg)
 135 {
 136         struct io_plan plan;
 137
 138         plan.u.write.buf = data;
 139         plan.u.write.len = len;
 140         plan.io = do_write;
 141         plan.next = cb;
 142         plan.next_arg = arg;
 143         plan.pollflag = POLLOUT;
 144         plan.state = IO_IO;
 145         return plan;
 146 }
 147
 148 static enum io_result do_read(struct io_conn *conn)
 149 {
 150         ssize_t ret = read(conn->fd.fd, conn->plan.u.read.buf,
 151                            conn->plan.u.read.len);
 152         if (ret <= 0)
 153                 return RESULT_CLOSE;
 154         conn->plan.u.read.buf += ret;
 155         conn->plan.u.read.len -= ret;
 156         if (conn->plan.u.read.len == 0)
 157                 return RESULT_FINISHED;
 158         else
 159                 return RESULT_AGAIN;
 160 }
 161
 162 /* Queue a request to read into a buffer. */
 163 struct io_plan io_read_(void *data, size_t len,
 164                         struct io_plan (*cb)(struct io_conn *, void *),
 165                         void *arg)
 166 {
 167         struct io_plan plan;
 168
 169         plan.u.read.buf = data;
 170         plan.u.read.len = len;
 171         plan.io = do_read;
 172         plan.next = cb;
 173         plan.next_arg = arg;
 174         plan.pollflag = POLLIN;
 175         plan.state = IO_IO;
 176         return plan;
 177 }
 178
 179 static enum io_result do_read_partial(struct io_conn *conn)
 180 {
 181         ssize_t ret = read(conn->fd.fd, conn->plan.u.readpart.buf,
 182                            *conn->plan.u.readpart.lenp);
 183         if (ret <= 0)
 184                 return RESULT_CLOSE;
 185         *conn->plan.u.readpart.lenp = ret;
 186         return RESULT_FINISHED;
 187 }
 188
 189 /* Queue a partial request to read into a buffer. */
 190 struct io_plan io_read_partial_(void *data, size_t *len,
 191                                 struct io_plan (*cb)(struct io_conn *, void *),
 192                                 void *arg)
 193 {
 194         struct io_plan plan;
 195
 196         plan.u.readpart.buf = data;
 197         plan.u.readpart.lenp = len;
 198         plan.io = do_read_partial;
 199         plan.next = cb;
 200         plan.next_arg = arg;
 201         plan.pollflag = POLLIN;
 202         plan.state = IO_IO;
 203
 204         return plan;
 205 }
 206
 207 static enum io_result do_write_partial(struct io_conn *conn)
 208 {
 209         ssize_t ret = write(conn->fd.fd, conn->plan.u.writepart.buf,
 210                             *conn->plan.u.writepart.lenp);
 211         if (ret < 0)
 212                 return RESULT_CLOSE;
 213         *conn->plan.u.writepart.lenp = ret;
 214         return RESULT_FINISHED;
 215 }
 216
 217 /* Queue a partial write request. */
 218 struct io_plan io_write_partial_(const void *data, size_t *len,
 219                                  struct io_plan (*cb)(struct io_conn*, void *),
 220                                  void *arg)
 221 {
 222         struct io_plan plan;
 223
 224         plan.u.writepart.buf = data;
 225         plan.u.writepart.lenp = len;
 226         plan.io = do_write_partial;
 227         plan.next = cb;
 228         plan.next_arg = arg;
 229         plan.pollflag = POLLOUT;
 230         plan.state = IO_IO;
 231
 232         return plan;
 233 }
 234
 235 struct io_plan io_idle(void)
 236 {
 237         struct io_plan plan;
 238
 239         plan.pollflag = 0;
 240         plan.state = IO_IDLE;
 241
 242         return plan;
 243 }
 244
 245 void io_wake_(struct io_conn *conn,
 246               struct io_plan (*fn)(struct io_conn *, void *), void *arg)
 247
 248 {
 249         /* It might have finished, but we haven't called its finish() yet. */
 250         if (conn->plan.state == IO_FINISHED)
 251                 return;
 252         assert(conn->plan.state == IO_IDLE);
 253         conn->plan.next = fn;
 254         conn->plan.next_arg = arg;
 255         conn->plan.pollflag = 0;
 256         conn->plan.state = IO_NEXT;
 257         backend_wakeup(conn);
 258 }
 259
 260 static struct io_plan do_next(struct io_conn *conn)
 261 {
 262         if (timeout_active(conn))
 263                 backend_del_timeout(conn);
 264         return conn->plan.next(conn, conn->plan.next_arg);
 265 }
 266
 267 struct io_plan do_ready(struct io_conn *conn)
 268 {
 269         assert(conn->plan.state == IO_IO);
 270         switch (conn->plan.io(conn)) {
 271         case RESULT_CLOSE:
 272                 return io_close(conn, NULL);
 273         case RESULT_FINISHED:
 274                 return do_next(conn);
 275         case RESULT_AGAIN:
 276                 return conn->plan;
 277         default:
 278                 abort();
 279         }
 280 }
 281
 282 /* Useful next functions. */
 283 /* Close the connection, we're done. */
 284 struct io_plan io_close(struct io_conn *conn, void *arg)
 285 {
 286         struct io_plan plan;
 287
 288         plan.state = IO_FINISHED;
 289         plan.pollflag = 0;
 290
 291         return plan;
 292 }
 293
 294 /* Exit the loop, returning this (non-NULL) arg. */
 295 struct io_plan io_break_(void *ret,
 296                          struct io_plan (*fn)(struct io_conn *, void *),
 297                          void *arg)
 298 {
 299         struct io_plan plan;
 300
 301         io_loop_return = ret;
 302
 303         plan.state = IO_NEXT;
 304         plan.pollflag = 0;
 305         plan.next = fn;
 306         plan.next_arg = arg;
 307
 308         return plan;
 309 }