core/src/connect_handler.c

   1 #include "connect_handler.h"
   2 #include "request.h"
   3 #include "core.h"
   4
   5 #include <sys/fcntl.h> /* fcntl */
   6
   7 #include <memory.h>
   8 #include <malloc.h> /* calloc */
   9 #include <errno.h>  /* errno */
  10 #include <ev.h>
  11
  12 struct connect_request {
  13         request_t request;
  14         union {
  15                 ev_io file_watcher;
  16                 ev_timer time_watcher;
  17         };
  18         int fd;
  19         int domain, type, protocol;
  20         struct sockaddr *addr;
  21         socklen_t addr_len;
  22         char timeout_iterations;
  23 };
  24
  25 #define DEFAULT_TIMEOUT_ITERATIONS 10
  26 #define DEFAULT_TIMEOUT_INTERVAL (0.25)
  27
  28 connect_handler_t *new_connect_handler(core_t *c) {
  29         connect_handler_t *h = calloc(1, sizeof(connect_handler_t));
  30         if(!h) goto fail;
  31         h->core = c;
  32         return h;
  33 fail:
  34         free_connect_handler(h);
  35         return NULL;
  36 }
  37
  38 void free_connect_handler(connect_handler_t *h) {
  39         if(!h) return;
  40         free(h);
  41 }
  42
  43 connect_request_t *new_connect_request(connect_handler_t *h, const struct sockaddr *addr, socklen_t addr_len) {
  44         connect_request_t *cr = calloc(1, sizeof(connect_request_t));
  45         if(!cr) goto fail;
  46         cr->addr = malloc(addr_len);
  47         if(!cr->addr) goto fail;
  48         memcpy(cr->addr, addr, addr_len);
  49         cr->addr_len = addr_len;
  50         cr->timeout_iterations = DEFAULT_TIMEOUT_ITERATIONS;
  51         return cr;
  52 fail:
  53         free_connect_request(cr);
  54         return NULL;
  55 }
  56
  57 void free_connect_request(connect_request_t *cr) {
  58         if(!cr) return;
  59         if(cr->addr) free(cr->addr);
  60         free(cr);
  61 }
  62
  63 void connect_request_set_socket(connect_request_t *cr, int fd) {
  64         cr->fd = fd;
  65 }
  66
  67 int connect_request_get_socket(connect_request_t *cr) {
  68         return cr->fd;
  69 }
  70
  71 void connect_request_set_create_params(connect_request_t *cr, int domain, int type, int protocol) {
  72         cr->fd = -1;
  73         cr->domain = domain;
  74         cr->type = type;
  75         cr->protocol = protocol;
  76 }
  77
  78 static void connect_cb(int revents, void* arg) {
  79         connect_request_t *cr = arg;
  80         int fd = cr->fd;
  81         if(revents & EV_TIMEOUT) {
  82                 /* TODO: timeout */
  83                 return;
  84         }
  85         if(!(revents & EV_WRITE)) /* TODO: error ? */
  86                 return;
  87         /* Check for connection error */
  88         int err;
  89         socklen_t errlen = sizeof(err);
  90         getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &errlen);
  91         if(err == 0) {
  92                 cr->request.on_complete((request_t*)cr);
  93                 return;
  94         }
  95         errno = err;
  96         cr->request.on_error((request_t*)cr, IOR_ERROR_ERRNO);
  97 }
  98
  99 static void setup_connect(struct ev_loop *loop, connect_request_t *cr) {
 100         int fd = cr->fd;
 101         int flags;
 102         /* Make sure the socket is non-blocking */
 103         if(-1 == (flags = fcntl(fd, F_GETFL, 0)))
 104                 flags = 0;
 105         fcntl(fd, F_SETFL, flags | O_NONBLOCK);
 106
 107 tryagain:
 108         if(0 != connect(cr->fd, cr->addr, cr->addr_len)) {
 109                 if(errno == EINTR) goto tryagain;
 110                 if(errno == EINPROGRESS) {
 111                         ev_once(loop, cr->fd, EV_WRITE, -1., connect_cb, cr);
 112                 } else {
 113                         cr->request.on_error((request_t*)cr, IOR_ERROR_ERRNO);
 114                 }
 115         } else {
 116                 cr->request.on_complete((request_t*)cr);
 117         }
 118 }
 119
 120 static void setup_socket_cb(struct ev_loop *loop, struct ev_timer *w, int revents);
 121
 122 static void setup_socket(struct ev_loop *loop, connect_request_t *cr, int firstcall) {
 123         cr->fd = socket(cr->domain, cr->type, cr->protocol);
 124         if(cr->fd == -1) {
 125                 if(errno == ENFILE) { /* no files remaining - should retry */
 126                         if(!firstcall) return;
 127                         ev_timer_init(&cr->time_watcher, setup_socket_cb, 0., DEFAULT_TIMEOUT_INTERVAL);
 128                         cr->time_watcher.data = cr;
 129                         ev_timer_start(loop, &cr->time_watcher);
 130                 } else { /* another error occurred... */
 131                         if(!firstcall)
 132                                 ev_timer_stop(loop, &cr->time_watcher);
 133                         cr->request.on_error((request_t*)cr, IOR_ERROR_ERRNO);
 134                         return;
 135                 }
 136         }
 137         /* begin connection process.. */
 138         if(!firstcall)
 139                 ev_timer_stop(loop, &cr->time_watcher);
 140         setup_connect(loop, cr);
 141 }
 142
 143 static void setup_socket_cb(struct ev_loop *loop, struct ev_timer *w, int revents) {
 144         connect_request_t *cr = w->data;
 145         cr->timeout_iterations --;
 146         if(cr->timeout_iterations <= 0) {
 147                 ev_timer_stop(loop, w);
 148         }
 149         setup_socket(loop, cr, 0);
 150 }
 151
 152 void connect_handler_queue(connect_handler_t *h, connect_request_t *cr) {
 153         if(cr->fd == -1) {
 154                 setup_socket(h->core->loop, cr, 1);
 155                 return;
 156         }
 157         setup_connect(h->core->loop, cr);
 158 }
 159