socket/tcp-bsd.c

   1 /*
   2  * This file is part of the Nice GLib ICE library.
   3  *
   4  * (C) 2006-2008 Collabora Ltd.
   5  *  Contact: Dafydd Harries
   6  *  Contact: Olivier Crete
   7  * (C) 2006, 2007 Nokia Corporation. All rights reserved.
   8  *  Contact: Kai Vehmanen
   9  *
  10  * The contents of this file are subject to the Mozilla Public License Version
  11  * 1.1 (the "License"); you may not use this file except in compliance with
  12  * the License. You may obtain a copy of the License at
  13  * http://www.mozilla.org/MPL/
  14  *
  15  * Software distributed under the License is distributed on an "AS IS" basis,
  16  * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
  17  * for the specific language governing rights and limitations under the
  18  * License.
  19  *
  20  * The Original Code is the Nice GLib ICE library.
  21  *
  22  * The Initial Developers of the Original Code are Collabora Ltd and Nokia
  23  * Corporation. All Rights Reserved.
  24  *
  25  * Contributors:
  26  *   Dafydd Harries, Collabora Ltd.
  27  *   Olivier Crete, Collabora Ltd.
  28  *   Rémi Denis-Courmont, Nokia
  29  *   Kai Vehmanen
  30  *
  31  * Alternatively, the contents of this file may be used under the terms of the
  32  * the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
  33  * case the provisions of LGPL are applicable instead of those above. If you
  34  * wish to allow use of your version of this file only under the terms of the
  35  * LGPL and not to allow others to use your version of this file under the
  36  * MPL, indicate your decision by deleting the provisions above and replace
  37  * them with the notice and other provisions required by the LGPL. If you do
  38  * not delete the provisions above, a recipient may use your version of this
  39  * file under either the MPL or the LGPL.
  40  */
  41
  42 /*
  43  * Implementation of TCP relay socket interface using TCP Berkeley sockets. (See
  44  * http://en.wikipedia.org/wiki/Berkeley_sockets.)
  45  */
  46 #ifdef HAVE_CONFIG_H
  47 # include "config.h"
  48 #endif
  49
  50 #include "tcp-bsd.h"
  51 #include "agent-priv.h"
  52
  53 #include <string.h>
  54 #include <errno.h>
  55 #include <fcntl.h>
  56
  57 #ifndef G_OS_WIN32
  58 #include <unistd.h>
  59 #endif
  60
  61 typedef struct {
  62   NiceAgent *agent;
  63   NiceAddress server_addr;
  64   GQueue send_queue;
  65   GMainContext *context;
  66   GIOChannel *io_channel;
  67   GSource *io_source;
  68 } TcpPriv;
  69
  70 struct to_be_sent {
  71   guint length;
  72   gchar *buf;
  73   gboolean can_drop;
  74 };
  75
  76 #define MAX_QUEUE_LENGTH 20
  77
  78 static void socket_close (NiceSocket *sock);
  79 static gint socket_recv (NiceSocket *sock, NiceAddress *from,
  80     guint len, gchar *buf);
  81 static gboolean socket_send (NiceSocket *sock, const NiceAddress *to,
  82     guint len, const gchar *buf);
  83 static gboolean socket_is_reliable (NiceSocket *sock);
  84
  85
  86 static void add_to_be_sent (NiceSocket *sock, const gchar *buf, guint len,
  87     gboolean head);
  88 static void free_to_be_sent (struct to_be_sent *tbs);
  89 static gboolean socket_send_more (GIOChannel *source, GIOCondition condition,
  90     gpointer data);
  91
  92 NiceSocket *
  93 nice_tcp_bsd_socket_new (NiceAgent *agent, GMainContext *ctx, NiceAddress *addr)
  94 {
  95   int sockfd = -1;
  96   int ret;
  97   struct sockaddr_storage name;
  98   guint name_len = sizeof (name);
  99   NiceSocket *sock = g_slice_new0 (NiceSocket);
 100   TcpPriv *priv;
 101
 102   if (addr != NULL) {
 103     nice_address_copy_to_sockaddr(addr, (struct sockaddr *)&name);
 104   } else {
 105     memset (&name, 0, sizeof (name));
 106     name.ss_family = AF_UNSPEC;
 107   }
 108
 109   if ((sockfd == -1) &&
 110       ((name.ss_family == AF_UNSPEC) ||
 111           (name.ss_family == AF_INET))) {
 112     sockfd = socket (PF_INET, SOCK_STREAM, 0);
 113     name.ss_family = AF_INET;
 114 #ifdef HAVE_SA_LEN
 115     name.ss_len = sizeof (struct sockaddr_in);
 116 #endif
 117   }
 118
 119   if (sockfd == -1) {
 120     g_slice_free (NiceSocket, sock);
 121     return NULL;
 122   }
 123
 124 #ifdef FD_CLOEXEC
 125   fcntl (sockfd, F_SETFD, fcntl (sockfd, F_GETFD) | FD_CLOEXEC);
 126 #endif
 127 #ifdef O_NONBLOCK
 128   fcntl (sockfd, F_SETFL, fcntl (sockfd, F_GETFL) | O_NONBLOCK);
 129 #endif
 130
 131   name_len = name.ss_family == AF_INET? sizeof (struct sockaddr_in) :
 132       sizeof(struct sockaddr_in6);
 133   ret = connect (sockfd, (const struct sockaddr *)&name, name_len);
 134
 135 #ifdef G_OS_WIN32
 136   if (ret < 0 && WSAGetLastError () != WSAEINPROGRESS) {
 137     closesocket (sockfd);
 138 #else
 139   if (ret < 0 && errno != EINPROGRESS) {
 140     close (sockfd);
 141 #endif
 142     g_slice_free (NiceSocket, sock);
 143     return NULL;
 144   }
 145
 146   name_len = name.ss_family == AF_INET? sizeof (struct sockaddr_in) :
 147       sizeof(struct sockaddr_in6);
 148   if (getsockname (sockfd, (struct sockaddr *) &name, &name_len) < 0) {
 149     g_slice_free (NiceSocket, sock);
 150 #ifdef G_OS_WIN32
 151     closesocket(sockfd);
 152 #else
 153     close (sockfd);
 154 #endif
 155     return NULL;
 156   }
 157
 158   nice_address_set_from_sockaddr (&sock->addr, (struct sockaddr *)&name);
 159
 160   sock->priv = priv = g_slice_new0 (TcpPriv);
 161
 162   priv->agent = agent;
 163   priv->context = ctx;
 164   priv->server_addr = *addr;
 165
 166   sock->fileno = sockfd;
 167   sock->send = socket_send;
 168   sock->recv = socket_recv;
 169   sock->is_reliable = socket_is_reliable;
 170   sock->close = socket_close;
 171
 172   return sock;
 173 }
 174
 175
 176 static void
 177 socket_close (NiceSocket *sock)
 178 {
 179   TcpPriv *priv = sock->priv;
 180
 181 #ifdef G_OS_WIN32
 182   closesocket(sock->fileno);
 183 #else
 184   close (sock->fileno);
 185 #endif
 186   if (priv->io_source) {
 187     g_source_destroy (priv->io_source);
 188     g_source_unref (priv->io_source);
 189   }
 190   if (priv->io_channel)
 191     g_io_channel_unref (priv->io_channel);
 192   g_queue_foreach (&priv->send_queue, (GFunc) free_to_be_sent, NULL);
 193   g_queue_clear (&priv->send_queue);
 194
 195   g_slice_free(TcpPriv, sock->priv);
 196 }
 197
 198 static gint
 199 socket_recv (NiceSocket *sock, NiceAddress *from, guint len, gchar *buf)
 200 {
 201   TcpPriv *priv = sock->priv;
 202   int ret;
 203
 204   ret = recv (sock->fileno, buf, len, 0);
 205
 206   /* recv returns 0 when the peer performed a shutdown.. we must return -1 here
 207    * so that the agent destroys the g_source */
 208   if (ret == 0)
 209     return -1;
 210
 211   if (ret < 0) {
 212 #ifdef G_OS_WIN32
 213     if (WSAGetLastError () == WSAEWOULDBLOCK)
 214 #else
 215     if (errno == EAGAIN)
 216 #endif
 217       return 0;
 218     else
 219       return ret;
 220   }
 221
 222   if (from)
 223     *from = priv->server_addr;
 224   return ret;
 225 }
 226
 227 /* Data sent to this function must be a single entity because buffers can be
 228  * dropped if the bandwidth isn't fast enough. So do not send a message in
 229  * multiple chunks. */
 230 static gboolean
 231 socket_send (NiceSocket *sock, const NiceAddress *to,
 232     guint len, const gchar *buf)
 233 {
 234   TcpPriv *priv = sock->priv;
 235   int ret;
 236
 237   /* First try to send the data, don't send it later if it can be sent now
 238      this way we avoid allocating memory on every send */
 239   if (g_queue_is_empty (&priv->send_queue)) {
 240     ret = send (sock->fileno, buf, len, 0);
 241
 242     if (ret < 0) {
 243 #ifdef G_OS_WIN32
 244       if (WSAGetLastError () == WSAEWOULDBLOCK) {
 245 #else
 246       if (errno == EAGAIN) {
 247 #endif
 248         add_to_be_sent (sock, buf, len, FALSE);
 249         return TRUE;
 250       } else {
 251         return FALSE;
 252       }
 253     } else if ((guint)ret < len) {
 254       add_to_be_sent (sock, buf + ret, len - ret, TRUE);
 255       return TRUE;
 256     }
 257   } else {
 258     if (g_queue_get_length(&priv->send_queue) >= MAX_QUEUE_LENGTH) {
 259       int peek_idx = 0;
 260       struct to_be_sent *tbs = NULL;
 261       while ((tbs = g_queue_peek_nth (&priv->send_queue, peek_idx)) != NULL) {
 262         if (tbs->can_drop) {
 263           tbs = g_queue_pop_nth (&priv->send_queue, peek_idx);
 264           g_free (tbs->buf);
 265           g_slice_free (struct to_be_sent, tbs);
 266           break;
 267         } else {
 268           peek_idx++;
 269         }
 270       }
 271     }
 272     add_to_be_sent (sock, buf, len, FALSE);
 273   }
 274
 275   return TRUE;
 276 }
 277
 278 static gboolean
 279 socket_is_reliable (NiceSocket *sock)
 280 {
 281   return TRUE;
 282 }
 283
 284
 285 /*
 286  * Returns:
 287  * -1 = error
 288  * 0 = have more to send
 289  * 1 = sent everything
 290  */
 291
 292 static gboolean
 293 socket_send_more (
 294   GIOChannel *source,
 295   GIOCondition condition,
 296   gpointer data)
 297 {
 298   NiceSocket *sock = (NiceSocket *) data;
 299   TcpPriv *priv = sock->priv;
 300   struct to_be_sent *tbs = NULL;
 301
 302   agent_lock ();
 303
 304   if (g_source_is_destroyed (g_main_current_source ())) {
 305     nice_debug ("Source was destroyed. "
 306         "Avoided race condition in tcp-bsd.c:socket_send_more");
 307     agent_unlock ();
 308     return FALSE;
 309   }
 310
 311   while ((tbs = g_queue_pop_head (&priv->send_queue)) != NULL) {
 312     int ret;
 313
 314     ret = send (sock->fileno, tbs->buf, tbs->length, 0);
 315
 316     if (ret < 0) {
 317 #ifdef G_OS_WIN32
 318       if (WSAGetLastError () == WSAEWOULDBLOCK) {
 319 #else
 320       if (errno == EAGAIN) {
 321 #endif
 322         add_to_be_sent (sock, tbs->buf, tbs->length, TRUE);
 323         g_free (tbs->buf);
 324         g_slice_free (struct to_be_sent, tbs);
 325         break;
 326       }
 327     } else if (ret < (int) tbs->length) {
 328       add_to_be_sent (sock, tbs->buf + ret, tbs->length - ret, TRUE);
 329       g_free (tbs->buf);
 330       g_slice_free (struct to_be_sent, tbs);
 331       break;
 332     }
 333
 334     g_free (tbs->buf);
 335     g_slice_free (struct to_be_sent, tbs);
 336   }
 337
 338   if (g_queue_is_empty (&priv->send_queue)) {
 339     g_io_channel_unref (priv->io_channel);
 340     priv->io_channel = NULL;
 341     g_source_destroy (priv->io_source);
 342     g_source_unref (priv->io_source);
 343     priv->io_source = NULL;
 344
 345     agent_unlock ();
 346     return FALSE;
 347   }
 348
 349   agent_unlock ();
 350   return TRUE;
 351 }
 352
 353
 354 static void
 355 add_to_be_sent (NiceSocket *sock, const gchar *buf, guint len, gboolean head)
 356 {
 357   TcpPriv *priv = sock->priv;
 358   struct to_be_sent *tbs = NULL;
 359
 360   if (len <= 0)
 361     return;
 362
 363   tbs = g_slice_new0 (struct to_be_sent);
 364   tbs->buf = g_memdup (buf, len);
 365   tbs->length = len;
 366   tbs->can_drop = !head;
 367   if (head)
 368     g_queue_push_head (&priv->send_queue, tbs);
 369   else
 370     g_queue_push_tail (&priv->send_queue, tbs);
 371
 372   if (priv->io_channel == NULL) {
 373     priv->io_channel = g_io_channel_unix_new (sock->fileno);
 374     priv->io_source = g_io_create_watch (priv->io_channel, G_IO_OUT);
 375     g_source_set_callback (priv->io_source, (GSourceFunc) socket_send_more,
 376         sock, NULL);
 377     g_source_attach (priv->io_source, priv->context);
 378   }
 379 }
 380
 381
 382
 383 static void
 384 free_to_be_sent (struct to_be_sent *tbs)
 385 {
 386   g_free (tbs->buf);
 387   g_slice_free (struct to_be_sent, tbs);
 388 }
 389