72204135e8388589bfc7094ca2ec2f9bf47e8eb0
[srvx.git] / src / ioset.c
1 /* ioset.h - srvx event loop
2  * Copyright 2002-2004, 2006 srvx Development Team
3  *
4  * This file is part of srvx.
5  *
6  * srvx is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License as published by
8  * the Free Software Foundation; either version 2 of the License, or
9  * (at your option) any later version.
10  *
11  * This program is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14  * GNU General Public License for more details.
15  *
16  * You should have received a copy of the GNU General Public License
17  * along with srvx; if not, write to the Free Software Foundation,
18  * Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
19  */
20
21 #include "ioset-impl.h"
22 #include "log.h"
23 #include "timeq.h"
24 #include "saxdb.h"
25 #include "conf.h"
26
27 #ifdef HAVE_FCNTL_H
28 #include <fcntl.h>
29 #endif
30 #ifdef HAVE_SYS_SOCKET_H
31 #include <sys/socket.h>
32 #endif
33
34 #define IS_EOL(CH) ((CH) == '\n')
35
36 extern int uplink_connect(void);
37 int clock_skew;
38 int do_write_dbs;
39 int do_reopen;
40 static struct io_engine *engine;
41 static struct io_fd *active_fd;
42
43 static void
44 ioq_init(struct ioq *ioq, int size) {
45     ioq->buf = malloc(size);
46     ioq->get = ioq->put = 0;
47     ioq->size = size;
48 }
49
50 static unsigned int
51 ioq_put_avail(const struct ioq *ioq) {
52     /* Subtract 1 from ioq->get to be sure we don't fill the buffer
53      * and make it look empty even when there's data in it. */
54     if (ioq->put < ioq->get)
55         return ioq->get - ioq->put - 1;
56     else if (ioq->get == 0)
57         return ioq->size - ioq->put - 1;
58     else
59         return ioq->size - ioq->put;
60 }
61
62 static unsigned int
63 ioq_get_avail(const struct ioq *ioq) {
64     return ((ioq->put < ioq->get) ? ioq->size : ioq->put) - ioq->get;
65 }
66
67 static unsigned int
68 ioq_used(const struct ioq *ioq) {
69     return ((ioq->put < ioq->get) ? ioq->size : 0) + ioq->put - ioq->get;
70 }
71
72 static unsigned int
73 ioq_grow(struct ioq *ioq) {
74     int new_size = ioq->size << 1;
75     char *new_buf = malloc(new_size);
76     int get_avail = ioq_get_avail(ioq);
77     memcpy(new_buf, ioq->buf + ioq->get, get_avail);
78     if (ioq->put < ioq->get)
79         memcpy(new_buf + get_avail, ioq->buf, ioq->put);
80     free(ioq->buf);
81     ioq->put = ioq_used(ioq);
82     ioq->get = 0;
83     ioq->buf = new_buf;
84     ioq->size = new_size;
85     return new_size - ioq->put;
86 }
87
88 extern struct io_engine io_engine_kqueue;
89 extern struct io_engine io_engine_epoll;
90 extern struct io_engine io_engine_win32;
91 extern struct io_engine io_engine_select;
92
93 void
94 ioset_init(void)
95 {
96     if (engine) /* someone beat us to it */
97         return;
98
99 #if WITH_IOSET_KQUEUE
100     if (!engine && io_engine_kqueue.init())
101         engine = &io_engine_kqueue;
102 #endif
103
104 #if WITH_IOSET_EPOLL
105     if (!engine && io_engine_epoll.init())
106         engine = &io_engine_epoll;
107 #endif
108
109 #if WITH_IOSET_WIN32
110     if (!engine && io_engine_win32.init())
111         engine = &io_engine_win32;
112 #endif
113
114     if (engine) {
115         /* we found one that works */
116     } else if (io_engine_select.init())
117         engine = &io_engine_select;
118     else
119         log_module(MAIN_LOG, LOG_FATAL, "No usable I/O engine found.");
120     log_module(MAIN_LOG, LOG_DEBUG, "Using %s I/O engine.", engine->name);
121 }
122
123 void
124 ioset_cleanup(void) {
125     engine->cleanup();
126 }
127
128 struct io_fd *
129 ioset_add(int fd) {
130     struct io_fd *res;
131     int flags;
132
133     if (fd < 0) {
134         log_module(MAIN_LOG, LOG_ERROR, "Somebody called ioset_add(%d) on a negative fd!", fd);
135         return 0;
136     }
137     if (!engine)
138         ioset_init();
139     res = calloc(1, sizeof(*res));
140     if (!res)
141         return 0;
142     res->fd = fd;
143     ioq_init(&res->send, 1024);
144     ioq_init(&res->recv, 1024);
145     engine->add(res);
146     flags = fcntl(fd, F_GETFL);
147     fcntl(fd, F_SETFL, flags|O_NONBLOCK);
148     return res;
149 }
150
151 struct io_fd *ioset_listen(struct sockaddr *local, unsigned int sa_size, void *data, void (*accept_cb)(struct io_fd *listener, struct io_fd *new_connect))
152 {
153     struct io_fd *io_fd;
154     unsigned int opt;
155     int res;
156     int fd;
157
158     fd = socket(local ? local->sa_family : PF_INET, SOCK_STREAM, 0);
159     if (fd < 0) {
160         log_module(MAIN_LOG, LOG_ERROR, "Unable to create listening socket: %s", strerror(errno));
161         return NULL;
162     }
163
164     if (local && sa_size) {
165         res = bind(fd, local, sa_size);
166         if (res < 0) {
167             log_module(MAIN_LOG, LOG_ERROR, "Unable to bind listening socket %d: %s", fd, strerror(errno));
168             close(fd);
169             return NULL;
170         }
171
172         opt = 1;
173         res = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
174         if (res < 0) {
175             log_module(MAIN_LOG, LOG_WARNING, "Unable to mark listener address as re-usable: %s", strerror(errno));
176         }
177     }
178
179     res = listen(fd, 1);
180     if (res < 0) {
181         log_module(MAIN_LOG, LOG_ERROR, "Unable to listen on socket %d: %s", fd, strerror(errno));
182         close(fd);
183         return NULL;
184     }
185
186     io_fd = ioset_add(fd);
187     if (!io_fd) {
188         close(fd);
189         return NULL;
190     }
191     io_fd->state = IO_LISTENING;
192     io_fd->data = data;
193     io_fd->accept_cb = accept_cb;
194     engine->update(io_fd);
195     return io_fd;
196 }
197
198 struct io_fd *
199 ioset_connect(struct sockaddr *local, unsigned int sa_size, const char *peer, unsigned int port, int blocking, void *data, void (*connect_cb)(struct io_fd *fd, int error)) {
200     struct addrinfo hints;
201     struct addrinfo *ai;
202     struct io_fd *io_fd;
203     struct io_fd *old_active;
204     int res;
205     int fd;
206     char portnum[10];
207
208     memset(&hints, 0, sizeof(hints));
209     hints.ai_family = local ? local->sa_family : 0;
210     hints.ai_socktype = SOCK_STREAM;
211     snprintf(portnum, sizeof(portnum), "%u", port);
212     if (getaddrinfo(peer, portnum, &hints, &ai)) {
213         log_module(MAIN_LOG, LOG_ERROR, "getaddrinfo(%s, %s) failed.", peer, portnum);
214         return NULL;
215     }
216
217     if (local) {
218         if ((fd = socket(local->sa_family, SOCK_STREAM, 0)) < 0) {
219             log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s)", peer, errno, strerror(errno));
220             freeaddrinfo(ai);
221             return NULL;
222         }
223         if (bind(fd, local, sa_size) < 0) {
224             log_module(MAIN_LOG, LOG_ERROR, "bind() of socket for %s (fd %d) returned errno %d (%s).  Will let operating system choose.", peer, fd, errno, strerror(errno));
225         }
226     } else {
227         if ((fd = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
228             log_module(MAIN_LOG, LOG_ERROR, "socket() for %s returned errno %d (%s).", peer, errno, strerror(errno));
229             freeaddrinfo(ai);
230             return NULL;
231         }
232     }
233
234     if (blocking) {
235         res = connect(fd, ai->ai_addr, ai->ai_addrlen);
236         io_fd = ioset_add(fd);
237     } else {
238         io_fd = ioset_add(fd);
239         res = connect(fd, ai->ai_addr, ai->ai_addrlen);
240     }
241     freeaddrinfo(ai);
242     if (!io_fd) {
243         close(fd);
244         return NULL;
245     }
246     io_fd->state = IO_CONNECTING;
247     io_fd->data = data;
248     io_fd->connect_cb = connect_cb;
249     if (res < 0) {
250         switch (errno) {
251         case EINPROGRESS: /* only if !blocking */
252             engine->update(io_fd);
253             return io_fd;
254         default:
255             log_module(MAIN_LOG, LOG_ERROR, "connect(%s:%d) (fd %d) returned errno %d (%s).", peer, port, io_fd->fd, errno, strerror(errno));
256             /* then fall through */
257         case EHOSTUNREACH:
258         case ECONNREFUSED:
259             ioset_close(io_fd, 1);
260             engine->update(io_fd);
261             return NULL;
262         }
263     }
264     io_fd->state = IO_CONNECTED;
265     old_active = active_fd;
266     if (connect_cb)
267         connect_cb(io_fd, ((res < 0) ? errno : 0));
268     if (active_fd)
269         engine->update(io_fd);
270     if (old_active != io_fd)
271         active_fd = old_active;
272     return io_fd;
273 }
274
275 void ioset_update(struct io_fd *fd) {
276     engine->update(fd);
277 }
278
279 static void
280 ioset_try_write(struct io_fd *fd) {
281     int res;
282     unsigned int req = ioq_get_avail(&fd->send);
283     res = write(fd->fd, fd->send.buf+fd->send.get, req);
284     if (res < 0) {
285         switch (errno) {
286         case EAGAIN:
287             break;
288         default:
289             log_module(MAIN_LOG, LOG_ERROR, "write() on fd %d error %d: %s", fd->fd, errno, strerror(errno));
290         }
291     } else {
292         fd->send.get += res;
293         if (fd->send.get == fd->send.size)
294             fd->send.get = 0;
295         engine->update(fd);
296     }
297 }
298
299 void
300 ioset_close(struct io_fd *fdp, int os_close) {
301     if (!fdp)
302         return;
303     if (active_fd == fdp)
304         active_fd = NULL;
305     if (fdp->destroy_cb)
306         fdp->destroy_cb(fdp);
307     if (fdp->send.get != fdp->send.put) {
308         int flags;
309
310         flags = fcntl(fdp->fd, F_GETFL);
311         fcntl(fdp->fd, F_SETFL, flags&~O_NONBLOCK);
312         ioset_try_write(fdp);
313         /* it may need to send the beginning of the buffer now.. */
314         if (fdp->send.get != fdp->send.put)
315             ioset_try_write(fdp);
316     }
317     free(fdp->send.buf);
318     free(fdp->recv.buf);
319     if (os_close)
320         close(fdp->fd);
321     engine->remove(fdp);
322     free(fdp);
323 }
324
325 static void
326 ioset_accept(struct io_fd *listener)
327 {
328     struct io_fd *old_active;
329     struct io_fd *new_fd;
330     int fd;
331
332     fd = accept(listener->fd, NULL, 0);
333     if (fd < 0) {
334         log_module(MAIN_LOG, LOG_ERROR, "Unable to accept new connection on listener %d: %s", listener->fd, strerror(errno));
335         return;
336     }
337
338     new_fd = ioset_add(fd);
339     new_fd->state = IO_CONNECTED;
340     old_active = active_fd;
341     active_fd = new_fd;
342     listener->accept_cb(listener, new_fd);
343     assert(active_fd == NULL || active_fd == new_fd);
344     if (active_fd == new_fd) {
345         if (new_fd->send.get != new_fd->send.put)
346             ioset_try_write(new_fd);
347         else
348             engine->update(new_fd);
349     }
350     active_fd = old_active;
351 }
352
353 static int
354 ioset_find_line_length(struct io_fd *fd) {
355     unsigned int pos, max, len;
356     len = 0;
357     max = (fd->recv.put < fd->recv.get) ? fd->recv.size : fd->recv.put;
358     for (pos = fd->recv.get; pos < max; ++pos, ++len)
359         if (IS_EOL(fd->recv.buf[pos]))
360             return fd->line_len = len + 1;
361     if (fd->recv.put < fd->recv.get)
362         for (pos = 0; pos < fd->recv.put; ++pos, ++len)
363             if (IS_EOL(fd->recv.buf[pos]))
364                 return fd->line_len = len + 1;
365     return fd->line_len = 0;
366 }
367
368 static void
369 ioset_buffered_read(struct io_fd *fd) {
370     int put_avail, nbr, fdnum;
371
372     if (!(put_avail = ioq_put_avail(&fd->recv)))
373         put_avail = ioq_grow(&fd->recv);
374     nbr = read(fd->fd, fd->recv.buf + fd->recv.put, put_avail);
375     if (nbr < 0) {
376         switch (errno) {
377         case EAGAIN:
378             break;
379         default:
380             log_module(MAIN_LOG, LOG_ERROR, "Unexpected read() error %d on fd %d: %s", errno, fd->fd, strerror(errno));
381             /* Just flag it as EOF and call readable_cb() to notify the fd's owner. */
382             fd->state = IO_CLOSED;
383             fd->readable_cb(fd);
384             if (active_fd == fd)
385                 engine->update(fd);
386         }
387     } else if (nbr == 0) {
388         fd->state = IO_CLOSED;
389         fd->readable_cb(fd);
390         if (active_fd == fd)
391             engine->update(fd);
392     } else {
393         if (fd->line_len == 0) {
394             unsigned int pos;
395             for (pos = fd->recv.put; pos < fd->recv.put + nbr; ++pos) {
396                 if (IS_EOL(fd->recv.buf[pos])) {
397                     if (fd->recv.put < fd->recv.get)
398                         fd->line_len = fd->recv.size + pos + 1 - fd->recv.get;
399                     else
400                         fd->line_len = pos + 1 - fd->recv.get;
401                     break;
402                 }
403             }
404         }
405         fd->recv.put += nbr;
406         if (fd->recv.put == fd->recv.size)
407             fd->recv.put = 0;
408         fdnum = fd->fd;
409         while (fd->wants_reads && (fd->line_len > 0)) {
410             struct io_fd *old_active;
411             int died = 0;
412
413             old_active = active_fd;
414             active_fd = fd;
415             fd->readable_cb(fd);
416             if (active_fd)
417                 ioset_find_line_length(fd);
418             else
419                 died = 1;
420             if (old_active != fd)
421                 active_fd = old_active;
422             if (died)
423                 break;
424         }
425     }
426 }
427
428 int
429 ioset_line_read(struct io_fd *fd, char *dest, int max) {
430     int avail, done;
431     if ((fd->state == IO_CLOSED) && (!ioq_get_avail(&fd->recv) ||  (fd->line_len < 0)))
432         return 0;
433     if (fd->line_len < 0)
434         return -1;
435     if (fd->line_len < max)
436         max = fd->line_len;
437     avail = ioq_get_avail(&fd->recv);
438     if (max > avail) {
439         memcpy(dest, fd->recv.buf + fd->recv.get, avail);
440         fd->recv.get += avail;
441         assert(fd->recv.get == fd->recv.size);
442         fd->recv.get = 0;
443         done = avail;
444     } else {
445         done = 0;
446     }
447     memcpy(dest + done, fd->recv.buf + fd->recv.get, max - done);
448     fd->recv.get += max - done;
449     if (fd->recv.get == fd->recv.size)
450         fd->recv.get = 0;
451     dest[max] = 0;
452     ioset_find_line_length(fd);
453     return max;
454 }
455
456 void
457 ioset_events(struct io_fd *fd, int readable, int writable)
458 {
459     if (!fd || (!readable && !writable))
460         return;
461     active_fd = fd;
462     switch (fd->state) {
463     case IO_CLOSED:
464         break;
465     case IO_LISTENING:
466         if (active_fd && readable)
467             ioset_accept(fd);
468         break;
469     case IO_CONNECTING:
470         assert(active_fd == NULL || active_fd == fd);
471         if (active_fd && readable) {
472             socklen_t arglen;
473             int rc;
474             arglen = sizeof(rc);
475             if (getsockopt(fd->fd, SOL_SOCKET, SO_ERROR, &rc, &arglen) < 0)
476                 rc = errno;
477             fd->state = IO_CLOSED;
478             if (fd->connect_cb)
479                 fd->connect_cb(fd, rc);
480         } else if (active_fd && writable) {
481             fd->state = IO_CONNECTED;
482             if (fd->connect_cb)
483                 fd->connect_cb(fd, 0);
484         }
485         if (active_fd != fd)
486             break;
487         engine->update(fd);
488         /* and fall through */
489     case IO_CONNECTED:
490         assert(active_fd == NULL || active_fd == fd);
491         if (active_fd && readable) {
492             if (fd->line_reads)
493                 ioset_buffered_read(fd);
494             else
495                 fd->readable_cb(fd);
496         }
497
498         assert(active_fd == NULL || active_fd == fd);
499         if (active_fd && writable)
500             ioset_try_write(fd);
501         break;
502     }
503 }
504
505 void
506 ioset_run(void) {
507     extern struct io_fd *socket_io_fd;
508     struct timeval timeout;
509     time_t wakey;
510
511     while (!quit_services) {
512         while (!socket_io_fd)
513             uplink_connect();
514
515         /* How long to sleep? (fill in select_timeout) */
516         wakey = timeq_next();
517         if ((wakey - now) < 0)
518             timeout.tv_sec = 0;
519         else
520             timeout.tv_sec = wakey - now;
521         timeout.tv_usec = 0;
522
523         if (engine->loop(&timeout))
524             continue;
525
526         /* Call any timeq events we need to call. */
527         timeq_run();
528         if (do_write_dbs) {
529             saxdb_write_all();
530             do_write_dbs = 0;
531         }
532         if (do_reopen) {
533             extern char *services_config;
534             conf_read(services_config);
535             do_reopen = 0;
536         }
537     }
538 }
539
540 void
541 ioset_write(struct io_fd *fd, const char *buf, unsigned int nbw) {
542     unsigned int avail;
543     while (ioq_used(&fd->send) + nbw >= fd->send.size)
544         ioq_grow(&fd->send);
545     avail = ioq_put_avail(&fd->send);
546     if (nbw > avail) {
547         memcpy(fd->send.buf + fd->send.put, buf, avail);
548         buf += avail;
549         nbw -= avail;
550         fd->send.put = 0;
551     }
552     memcpy(fd->send.buf + fd->send.put, buf, nbw);
553     fd->send.put += nbw;
554     if (fd->send.put == fd->send.size)
555         fd->send.put = 0;
556     engine->update(fd);
557 }
558
559 int
560 ioset_printf(struct io_fd *fd, const char *fmt, ...) {
561     char tmpbuf[MAXLEN];
562     va_list ap;
563     int res;
564
565     va_start(ap, fmt);
566     res = vsnprintf(tmpbuf, sizeof(tmpbuf), fmt, ap);
567     va_end(ap);
568     if (res > 0 && (size_t)res <= sizeof(tmpbuf))
569         ioset_write(fd, tmpbuf, res);
570     return res;
571 }
572
573 void
574 ioset_set_time(unsigned long new_now) {
575     clock_skew = new_now - time(NULL);
576     now = new_now;
577 }