socket/tcp-bsd.c

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