tcp.c

   1 /* Evented TCP Server
   2  * Copyright (c) 2007 Ry Dahl <ry.d4hl@gmail.com>
   3  * All rights reserved.
   4  */
   5
   6 /* TODO: add timeouts for clients */
   7 #include <unistd.h>
   8 #include <fcntl.h>
   9 #include <sys/types.h>
  10 #include <arpa/inet.h>
  11 #include <netdb.h>
  12
  13 #include <stdio.h>
  14 #include <string.h>
  15 #include <stdlib.h>
  16 #include <errno.h>
  17
  18 #include <ev.h>
  19 #include <glib.h>
  20
  21 #include <assert.h>
  22
  23 #include "tcp.h"
  24
  25 #define TCP_CHUNKSIZE (16*1024)
  26
  27 /* Private function */
  28 void tcp_client_free(tcp_client *client);
  29
  30 /* Returns the number of bytes remaining to write */
  31 int tcp_client_write(tcp_client *client, const char *data, int length)
  32 {
  33   assert(client->open);
  34   int sent = send(client->fd, data, length, 0);
  35   if(sent < 0) {
  36     tcp_error(strerror(errno));
  37     tcp_client_close(client);
  38     return 0;
  39   }
  40   return sent;
  41 }
  42
  43 void tcp_client_on_readable( struct ev_loop *loop
  44                            , struct ev_io *watcher
  45                            , int revents
  46                            )
  47 {
  48   tcp_client *client = (tcp_client*)(watcher->data);
  49   int length;
  50
  51   // check for error in revents
  52   if(EV_ERROR & revents) {
  53     tcp_error("tcp_client_on_readable() got error event, closing client");
  54     tcp_client_free(client);
  55     return;
  56   }
  57
  58   assert(client->open);
  59
  60   if(client->read_cb == NULL) return;
  61
  62   length = recv(client->fd, client->read_buffer, TCP_CHUNKSIZE, 0);
  63
  64   if(length == 0) {
  65     tcp_client_free(client);
  66     g_debug("zero length read? what to do?");
  67     return;
  68   } else if(length < 0) {
  69     if(errno == EBADF || errno == ECONNRESET) {
  70       g_debug("errno says Connection reset by peer");
  71     } else {
  72       tcp_error("Error recving data: %s", strerror(errno));
  73     }
  74     tcp_client_free(client);
  75     return;
  76   }
  77
  78   g_debug("Read %d bytes", length);
  79
  80   /* User needs to copy the data out of the buffer or process it before
  81    * leaving this function.
  82    */
  83   client->read_cb(client->read_buffer, length, client->read_cb_data);
  84 }
  85
  86 tcp_client* tcp_client_new(tcp_server *server)
  87 {
  88   socklen_t len;
  89   tcp_client *client;
  90
  91   client = g_new0(tcp_client, 1);
  92   client->parent = server;
  93   client->fd = accept(server->fd, (struct sockaddr*)&(client->sockaddr), &len);
  94   if(client->fd < 0) {
  95     tcp_error("Could not get client socket");
  96     tcp_client_free(client);
  97     return NULL;
  98   }
  99   client->open = TRUE;
 100
 101   int r = fcntl(client->fd, F_SETFL, O_NONBLOCK);
 102   if(r < 0) {
 103     tcp_error("Setting nonblock mode on socket failed");
 104     tcp_client_free(client);
 105     return NULL;
 106   }
 107
 108   client->read_buffer = (char*)malloc(sizeof(char)*TCP_CHUNKSIZE);
 109
 110   client->read_watcher = g_new0(struct ev_io, 1);
 111   client->read_watcher->data = client;
 112   ev_init (client->read_watcher, tcp_client_on_readable);
 113   ev_io_set (client->read_watcher, client->fd, EV_READ | EV_ERROR);
 114   ev_io_start (server->loop, client->read_watcher);
 115
 116   return client;
 117 }
 118
 119 void tcp_client_free(tcp_client *client)
 120 {
 121   if(client->open)
 122     tcp_client_close(client);
 123   free(client->read_buffer);
 124   free(client);
 125   g_debug("tcp client freed");
 126
 127 }
 128
 129 void tcp_client_close(tcp_client *client)
 130 {
 131   assert(client->open);
 132
 133   if(client->read_watcher) {
 134     g_debug("killing read watcher");
 135     ev_io_stop(client->parent->loop, client->read_watcher);
 136     free(client->read_watcher);
 137     client->read_watcher = NULL;
 138   }
 139   close(client->fd);
 140   client->open = FALSE;
 141 }
 142
 143 tcp_server* tcp_server_new()
 144 {
 145   int r;
 146   int flags = 1;
 147
 148   tcp_server *server = g_new0(tcp_server, 1);
 149
 150   server->fd = socket(PF_INET, SOCK_STREAM, 0);
 151   r = fcntl(server->fd, F_SETFL, O_NONBLOCK);
 152   if(r < 0) {
 153     tcp_error("Setting nonblock mode on socket failed");
 154     tcp_server_free(server);
 155     return NULL;
 156   }
 157
 158   r = setsockopt(server->fd, SOL_SOCKET, SO_REUSEADDR, &flags, sizeof(flags));
 159   if(r < 0) {
 160     tcp_error("failed to set setsock to reuseaddr");
 161     tcp_server_free(server);
 162     return NULL;
 163   }
 164   /*
 165   r = setsockopt(server->fd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(flags));
 166   if(r < 0) {
 167     tcp_error("failed to set socket to nodelay");
 168     tcp_server_free(server);
 169     return NULL;
 170   }
 171   */
 172
 173   server->loop = ev_default_loop(0);
 174   server->clients = g_queue_new();
 175   server->open = FALSE;
 176   return server;
 177 }
 178
 179 void tcp_server_free(tcp_server *server)
 180 {
 181   tcp_server_close(server);
 182   g_queue_free(server->clients);
 183   free(server);
 184 }
 185
 186 void tcp_server_close(tcp_server *server)
 187 {
 188   assert(server->open);
 189
 190   tcp_client *client;
 191   while((client = g_queue_pop_head(server->clients)))
 192     tcp_client_close(client);
 193
 194   if(server->port_s) {
 195     free(server->port_s);
 196     server->port_s = NULL;
 197   }
 198   if(server->dns_info) {
 199     free(server->dns_info);
 200     server->dns_info = NULL;
 201   }
 202   if(server->accept_watcher) {
 203     printf("killing accept watcher\n");
 204     ev_io_stop(server->loop, server->accept_watcher);
 205     free(server->accept_watcher);
 206     server->accept_watcher = NULL;
 207   }
 208   ev_unloop(server->loop, EVUNLOOP_ALL);
 209
 210   close(server->fd);
 211   server->open = FALSE;
 212 }
 213
 214 void tcp_server_accept( struct ev_loop *loop
 215                       , struct ev_io *watcher
 216                       , int revents
 217                       )
 218 {
 219   tcp_server *server = (tcp_server*)(watcher->data);
 220   tcp_client *client;
 221
 222   assert(server->open);
 223   assert(server->loop == loop);
 224
 225   // check for error in revents
 226   if(EV_ERROR & revents) {
 227     tcp_error("tcp_client_on_readable() got error event, closing client");
 228     tcp_server_close(server);
 229     return;
 230   }
 231
 232   client = tcp_client_new(server);
 233   g_queue_push_head(server->clients, (gpointer)client);
 234
 235   if(server->accept_cb != NULL)
 236     server->accept_cb(client, server->accept_cb_data);
 237
 238   return;
 239 }
 240
 241 void tcp_server_listen ( tcp_server *server
 242                        , char *address
 243                        , int port
 244                        , int backlog
 245                        , tcp_server_accept_cb_t accept_cb
 246                        , void *accept_cb_data
 247                        )
 248 {
 249   int r;
 250
 251   server->sockaddr.sin_family = AF_INET;
 252   server->sockaddr.sin_port = htons(port);
 253
 254   /* for easy access to the port */
 255   server->port_s = malloc(sizeof(char)*8);
 256   sprintf(server->port_s, "%d", port);
 257
 258   server->dns_info = gethostbyname(address);
 259   if (!(server->dns_info && server->dns_info->h_addr)) {
 260     tcp_error("Could not look up hostname %s", address);
 261     goto error;
 262   }
 263   memmove(&(server->sockaddr.sin_addr), server->dns_info->h_addr, sizeof(struct in_addr));
 264
 265   /* Other socket options. These could probably be fine tuned.
 266    * SO_SNDBUF       set buffer size for output
 267    * SO_RCVBUF       set buffer size for input
 268    * SO_SNDLOWAT     set minimum count for output
 269    * SO_RCVLOWAT     set minimum count for input
 270    * SO_SNDTIMEO     set timeout value for output
 271    * SO_RCVTIMEO     set timeout value for input
 272    */
 273   r = bind(server->fd, (struct sockaddr*)&(server->sockaddr), sizeof(server->sockaddr));
 274   if(r < 0) {
 275     tcp_error("Failed to bind to %s %d", address, port);
 276     goto error;
 277   }
 278
 279   r = listen(server->fd, backlog);
 280   if(r < 0) {
 281     tcp_error("listen() failed");
 282     goto error;
 283   }
 284
 285   assert(server->open == FALSE);
 286   server->open = TRUE;
 287
 288   server->accept_watcher = g_new0(struct ev_io, 1);
 289   server->accept_watcher->data = server;
 290   server->accept_cb = accept_cb;
 291   server->accept_cb_data = accept_cb_data;
 292
 293   ev_init (server->accept_watcher, tcp_server_accept);
 294   ev_io_set (server->accept_watcher, server->fd, EV_READ | EV_ERROR);
 295   ev_io_start (server->loop, server->accept_watcher);
 296   ev_loop (server->loop, 0);
 297   return;
 298
 299 error:
 300   tcp_server_close(server);
 301   return;
 302 }
 303
 304 char* tcp_server_address(tcp_server *server)
 305 {
 306   if(server->dns_info)
 307     return server->dns_info->h_name;
 308   else
 309     return NULL;
 310 }