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 #include "socket-priv.h"
  48
  49 #include "tcp-passive.h"
  50
  51 #include <string.h>
  52 #include <errno.h>
  53 #include <fcntl.h>
  54
  55 #ifndef G_OS_WIN32
  56 #include <unistd.h>
  57 #endif
  58
  59 /* FIXME: This should be defined in gio/gnetworking.h, which we should include;
  60  * but we cannot do that without refactoring.
  61  * (See: https://phabricator.freedesktop.org/D230). */
  62 #undef TCP_NODELAY
  63 #define TCP_NODELAY 1
  64
  65 static GMutex mutex;
  66
  67 typedef struct {
  68   NiceAddress remote_addr;
  69   GQueue send_queue;
  70   GMainContext *context;
  71   GSource *io_source;
  72   gboolean error;
  73   gboolean reliable;
  74   NiceSocketWritableCb writable_cb;
  75   gpointer writable_data;
  76   NiceSocket *passive_parent;
  77 } TcpPriv;
  78
  79 #define MAX_QUEUE_LENGTH 20
  80
  81 static void socket_close (NiceSocket *sock);
  82 static gint socket_recv_messages (NiceSocket *sock,
  83     NiceInputMessage *recv_messages, guint n_recv_messages);
  84 static gint socket_send_messages (NiceSocket *sock, const NiceAddress *to,
  85     const NiceOutputMessage *messages, guint n_messages);
  86 static gint socket_send_messages_reliable (NiceSocket *sock,
  87     const NiceAddress *to, const NiceOutputMessage *messages, guint n_messages);
  88 static gboolean socket_is_reliable (NiceSocket *sock);
  89 static gboolean socket_can_send (NiceSocket *sock, NiceAddress *addr);
  90 static void socket_set_writable_callback (NiceSocket *sock,
  91     NiceSocketWritableCb callback, gpointer user_data);
  92
  93 static gboolean socket_send_more (GSocket *gsocket, GIOCondition condition,
  94     gpointer data);
  95
  96 NiceSocket *
  97 nice_tcp_bsd_socket_new_from_gsock (GMainContext *ctx, GSocket *gsock,
  98     NiceAddress *local_addr, NiceAddress *remote_addr, gboolean reliable)
  99 {
 100   NiceSocket *sock;
 101   TcpPriv *priv;
 102
 103   g_return_val_if_fail (G_IS_SOCKET (gsock), NULL);
 104
 105   sock = g_slice_new0 (NiceSocket);
 106   sock->priv = priv = g_slice_new0 (TcpPriv);
 107
 108   if (ctx == NULL)
 109     ctx = g_main_context_default ();
 110   priv->context = g_main_context_ref (ctx);
 111   priv->remote_addr = *remote_addr;
 112   priv->error = FALSE;
 113   priv->reliable = reliable;
 114   priv->writable_cb = NULL;
 115   priv->writable_data = NULL;
 116
 117   sock->type = NICE_SOCKET_TYPE_TCP_BSD;
 118   sock->fileno = g_object_ref (gsock);
 119   sock->addr = *local_addr;
 120   sock->send_messages = socket_send_messages;
 121   sock->send_messages_reliable = socket_send_messages_reliable;
 122   sock->recv_messages = socket_recv_messages;
 123   sock->is_reliable = socket_is_reliable;
 124   sock->can_send = socket_can_send;
 125   sock->set_writable_callback = socket_set_writable_callback;
 126   sock->close = socket_close;
 127
 128   return sock;
 129 }
 130
 131 NiceSocket *
 132 nice_tcp_bsd_socket_new (GMainContext *ctx, NiceAddress *local_addr,
 133     NiceAddress *remote_addr, gboolean reliable)
 134 {
 135   union {
 136     struct sockaddr_storage storage;
 137     struct sockaddr addr;
 138   } name;
 139   NiceSocket *sock;
 140   GSocket *gsock = NULL;
 141   GError *gerr = NULL;
 142   gboolean gret = FALSE;
 143   GSocketAddress *gaddr;
 144
 145   if (remote_addr == NULL) {
 146     /* We can't connect a tcp socket with no destination address */
 147     return NULL;
 148   }
 149
 150   nice_address_copy_to_sockaddr (remote_addr, &name.addr);
 151
 152   if (name.storage.ss_family == AF_UNSPEC || name.storage.ss_family == AF_INET) {
 153     gsock = g_socket_new (G_SOCKET_FAMILY_IPV4, G_SOCKET_TYPE_STREAM,
 154         G_SOCKET_PROTOCOL_TCP, NULL);
 155
 156     name.storage.ss_family = AF_INET;
 157 #ifdef HAVE_SA_LEN
 158     name.storage.ss_len = sizeof (struct sockaddr_in);
 159 #endif
 160   } else if (name.storage.ss_family == AF_INET6) {
 161     gsock = g_socket_new (G_SOCKET_FAMILY_IPV6, G_SOCKET_TYPE_STREAM,
 162         G_SOCKET_PROTOCOL_TCP, NULL);
 163     name.storage.ss_family = AF_INET6;
 164 #ifdef HAVE_SA_LEN
 165     name.storage.ss_len = sizeof (struct sockaddr_in6);
 166 #endif
 167   }
 168
 169   if (gsock == NULL) {
 170     return NULL;
 171   }
 172
 173   gaddr = g_socket_address_new_from_native (&name.addr, sizeof (name));
 174   if (gaddr == NULL) {
 175     g_object_unref (gsock);
 176     return NULL;
 177   }
 178
 179   /* GSocket: All socket file descriptors are set to be close-on-exec. */
 180   g_socket_set_blocking (gsock, false);
 181
 182   /* setting TCP_NODELAY to TRUE in order to avoid packet batching */
 183   g_socket_set_option (gsock, IPPROTO_TCP, TCP_NODELAY, TRUE, NULL);
 184
 185   gret = g_socket_connect (gsock, gaddr, NULL, &gerr);
 186   g_object_unref (gaddr);
 187
 188   if (gret == FALSE) {
 189     if (g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_PENDING) == FALSE) {
 190       g_error_free (gerr);
 191       g_socket_close (gsock, NULL);
 192       g_object_unref (gsock);
 193       return NULL;
 194     }
 195     g_error_free (gerr);
 196   }
 197
 198   nice_address_copy_to_sockaddr (local_addr, &name.addr);
 199   gaddr = g_socket_address_new_from_native (&name.addr, sizeof (name));
 200   if (gaddr == NULL) {
 201     g_socket_close (gsock, NULL);
 202     g_object_unref (gsock);
 203     return NULL;
 204   }
 205   g_socket_bind (gsock, gaddr, FALSE, NULL);
 206   g_object_unref (gaddr);
 207
 208   sock = nice_tcp_bsd_socket_new_from_gsock (ctx, gsock, local_addr, remote_addr,
 209       reliable);
 210   g_object_unref (gsock);
 211
 212   return sock;
 213 }
 214
 215
 216 static void
 217 socket_close (NiceSocket *sock)
 218 {
 219   TcpPriv *priv = sock->priv;
 220
 221   g_mutex_lock (&mutex);
 222
 223   if (sock->fileno) {
 224     g_socket_close (sock->fileno, NULL);
 225     g_object_unref (sock->fileno);
 226     sock->fileno = NULL;
 227   }
 228   if (priv->io_source) {
 229     g_source_destroy (priv->io_source);
 230     g_source_unref (priv->io_source);
 231   }
 232
 233   if (priv->passive_parent) {
 234     nice_tcp_passive_socket_remove_connection (priv->passive_parent, &priv->remote_addr);
 235   }
 236
 237   nice_socket_free_send_queue (&priv->send_queue);
 238
 239   if (priv->context)
 240     g_main_context_unref (priv->context);
 241
 242   g_mutex_unlock (&mutex);
 243
 244   g_slice_free(TcpPriv, sock->priv);
 245 }
 246
 247 static gint
 248 socket_recv_messages (NiceSocket *sock,
 249     NiceInputMessage *recv_messages, guint n_recv_messages)
 250 {
 251   TcpPriv *priv = sock->priv;
 252   guint i;
 253
 254   /* Make sure socket has not been freed: */
 255   g_assert (sock->priv != NULL);
 256
 257   /* Don't try to access the socket if it had an error */
 258   if (priv->error)
 259     return -1;
 260
 261   for (i = 0; i < n_recv_messages; i++) {
 262     gint flags = G_SOCKET_MSG_NONE;
 263     GError *gerr = NULL;
 264     gssize len;
 265
 266     len = g_socket_receive_message (sock->fileno, NULL,
 267         recv_messages[i].buffers, recv_messages[i].n_buffers,
 268         NULL, NULL, &flags, NULL, &gerr);
 269
 270     recv_messages[i].length = MAX (len, 0);
 271
 272     /* recv returns 0 when the peer performed a shutdown.. we must return -1
 273      * here so that the agent destroys the g_source */
 274     if (len == 0) {
 275       priv->error = TRUE;
 276       break;
 277     }
 278
 279     if (len < 0) {
 280       if (g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
 281         len = 0;
 282
 283       g_error_free (gerr);
 284       return len;
 285     }
 286
 287     if (recv_messages[i].from)
 288       *recv_messages[i].from = priv->remote_addr;
 289   }
 290
 291   /* Was there an error processing the first message? */
 292   if (priv->error && i == 0)
 293     return -1;
 294
 295   return i;
 296 }
 297
 298 static gssize
 299 socket_send_message (NiceSocket *sock,
 300     const NiceOutputMessage *message, gboolean reliable)
 301 {
 302   TcpPriv *priv = sock->priv;
 303   gssize ret;
 304   GError *gerr = NULL;
 305   gsize message_len;
 306
 307   /* Make sure socket has not been freed: */
 308   g_assert (sock->priv != NULL);
 309
 310   /* Don't try to access the socket if it had an error, otherwise we risk a
 311    * crash with SIGPIPE (Broken pipe) */
 312   if (priv->error)
 313     return -1;
 314
 315   message_len = output_message_get_size (message);
 316
 317   /* First try to send the data, don't send it later if it can be sent now
 318    * this way we avoid allocating memory on every send */
 319   if (g_queue_is_empty (&priv->send_queue)) {
 320     ret = g_socket_send_message (sock->fileno, NULL, message->buffers,
 321         message->n_buffers, NULL, 0, G_SOCKET_MSG_NONE, NULL, &gerr);
 322
 323     if (ret < 0) {
 324       if (g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK) ||
 325           g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_FAILED)) {
 326         /* Queue the message and send it later. */
 327         nice_socket_queue_send_with_callback (&priv->send_queue,
 328             message, 0, message_len, FALSE, sock->fileno, &priv->io_source,
 329             priv->context, socket_send_more, sock);
 330         ret = message_len;
 331       }
 332
 333       g_error_free (gerr);
 334     } else if ((gsize) ret < message_len) {
 335       /* Partial send. */
 336       nice_socket_queue_send_with_callback (&priv->send_queue,
 337           message, ret, message_len, TRUE, sock->fileno, &priv->io_source,
 338           priv->context, socket_send_more, sock);
 339       ret = message_len;
 340     }
 341   } else {
 342     /* Only queue if we're sending reliably  */
 343     if (reliable) {
 344       /* Queue the message and send it later. */
 345       nice_socket_queue_send_with_callback (&priv->send_queue,
 346           message, 0, message_len, FALSE, sock->fileno, &priv->io_source,
 347           priv->context, socket_send_more, sock);
 348       ret = message_len;
 349     } else {
 350       /* non reliable send, so we shouldn't queue the message */
 351       ret = 0;
 352     }
 353   }
 354
 355   return ret;
 356 }
 357
 358 /* Data sent to this function must be a single entity because buffers can be
 359  * dropped if the bandwidth isn't fast enough. So do not send a message in
 360  * multiple chunks. */
 361 static gint
 362 socket_send_messages (NiceSocket *sock, const NiceAddress *to,
 363     const NiceOutputMessage *messages, guint n_messages)
 364 {
 365   guint i;
 366
 367   /* Make sure socket has not been freed: */
 368   g_assert (sock->priv != NULL);
 369
 370   for (i = 0; i < n_messages; i++) {
 371     const NiceOutputMessage *message = &messages[i];
 372     gssize len;
 373
 374     len = socket_send_message (sock, message, FALSE);
 375
 376     if (len < 0) {
 377       /* Error. */
 378       if (i > 0)
 379         break;
 380       return len;
 381     } else if (len == 0) {
 382       /* EWOULDBLOCK. */
 383       break;
 384     }
 385   }
 386
 387   return i;
 388 }
 389
 390 static gint
 391 socket_send_messages_reliable (NiceSocket *sock, const NiceAddress *to,
 392     const NiceOutputMessage *messages, guint n_messages)
 393 {
 394   guint i;
 395
 396   for (i = 0; i < n_messages; i++) {
 397     if (socket_send_message (sock, &messages[i], TRUE) < 0) {
 398       /* Error. */
 399       return -1;
 400     }
 401   }
 402
 403   return i;
 404 }
 405
 406 static gboolean
 407 socket_is_reliable (NiceSocket *sock)
 408 {
 409   TcpPriv *priv = sock->priv;
 410
 411   return priv->reliable;
 412 }
 413
 414 static gboolean
 415 socket_can_send (NiceSocket *sock, NiceAddress *addr)
 416 {
 417   TcpPriv *priv = sock->priv;
 418
 419   return g_queue_is_empty (&priv->send_queue);
 420 }
 421
 422 static void
 423 socket_set_writable_callback (NiceSocket *sock,
 424     NiceSocketWritableCb callback, gpointer user_data)
 425 {
 426   TcpPriv *priv = sock->priv;
 427
 428   priv->writable_cb = callback;
 429   priv->writable_data = user_data;
 430 }
 431
 432 static gboolean
 433 socket_send_more (
 434   GSocket *gsocket,
 435   GIOCondition condition,
 436   gpointer data)
 437 {
 438   NiceSocket *sock = (NiceSocket *) data;
 439   TcpPriv *priv;
 440
 441   g_mutex_lock (&mutex);
 442
 443   if (g_source_is_destroyed (g_main_current_source ())) {
 444     nice_debug ("Source was destroyed. "
 445         "Avoided race condition in tcp-bsd.c:socket_send_more");
 446     g_mutex_unlock (&mutex);
 447     return FALSE;
 448   }
 449
 450   priv = sock->priv;
 451
 452   /* connection hangs up or queue was emptied */
 453   if (condition & G_IO_HUP ||
 454       nice_socket_flush_send_queue_to_socket (sock->fileno,
 455           &priv->send_queue)) {
 456     g_source_destroy (priv->io_source);
 457     g_source_unref (priv->io_source);
 458     priv->io_source = NULL;
 459
 460     g_mutex_unlock (&mutex);
 461
 462     if (priv->writable_cb)
 463       priv->writable_cb (sock, priv->writable_data);
 464
 465     return FALSE;
 466   }
 467
 468   g_mutex_unlock (&mutex);
 469   return TRUE;
 470 }
 471
 472 void
 473 nice_tcp_bsd_socket_set_passive_parent (NiceSocket *sock, NiceSocket *passive_parent)
 474 {
 475   TcpPriv *priv = sock->priv;
 476
 477   g_assert (priv->passive_parent == NULL);
 478
 479   priv->passive_parent = passive_parent;
 480 }
 481
 482 NiceSocket *
 483 nice_tcp_bsd_socket_get_passive_parent (NiceSocket *sock)
 484 {
 485   TcpPriv *priv = sock->priv;
 486
 487   return priv->passive_parent;
 488 }