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   NiceAgent *agent;
  58   NiceAddress server_addr;
  59   GQueue send_queue;
  60   GMainContext *context;
  61   GIOChannel *io_channel;
  62   GSource *io_source;
  63 } TcpPriv;
  64
  65 struct to_be_sent {
  66   guint length;
  67   gchar *buf;
  68   gboolean can_drop;
  69 };
  70
  71 #define MAX_QUEUE_LENGTH 20
  72
  73 static void socket_close (NiceSocket *sock);
  74 static gint socket_recv (NiceSocket *sock, NiceAddress *from,
  75     guint len, gchar *buf);
  76 static gboolean socket_send (NiceSocket *sock, const NiceAddress *to,
  77     guint len, const gchar *buf);
  78 static gboolean socket_is_reliable (NiceSocket *sock);
  79
  80
  81 static void add_to_be_sent (NiceSocket *sock, const gchar *buf, guint len,
  82     gboolean head);
  83 static void free_to_be_sent (struct to_be_sent *tbs);
  84 static gboolean socket_send_more (GIOChannel *source, GIOCondition condition,
  85     gpointer data);
  86
  87 NiceSocket *
  88 nice_tcp_bsd_socket_new (NiceAgent *agent, GMainContext *ctx, NiceAddress *addr)
  89 {
  90   int sockfd = -1;
  91   int ret;
  92   struct sockaddr_storage name;
  93   guint name_len = sizeof (name);
  94   NiceSocket *sock;
  95   TcpPriv *priv;
  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 (sockfd == -1) {
 107     if (name.ss_family == AF_UNSPEC || name.ss_family == AF_INET) {
 108       sockfd = socket (PF_INET, SOCK_STREAM, 0);
 109       name.ss_family = AF_INET;
 110 #ifdef HAVE_SA_LEN
 111       name.ss_len = sizeof (struct sockaddr_in);
 112 #endif
 113     } else if (name.ss_family == AF_INET6) {
 114       sockfd = socket (PF_INET6, SOCK_STREAM, 0);
 115       name.ss_family = AF_INET6;
 116 #ifdef HAVE_SA_LEN
 117       name.ss_len = sizeof (struct sockaddr_in6);
 118 #endif
 119     }
 120   }
 121
 122   if (sockfd == -1) {
 123     g_slice_free (NiceSocket, sock);
 124     return NULL;
 125   }
 126
 127 #ifdef FD_CLOEXEC
 128   fcntl (sockfd, F_SETFD, fcntl (sockfd, F_GETFD) | FD_CLOEXEC);
 129 #endif
 130 #ifdef O_NONBLOCK
 131   fcntl (sockfd, F_SETFL, fcntl (sockfd, F_GETFL) | O_NONBLOCK);
 132 #endif
 133
 134   name_len = name.ss_family == AF_INET? sizeof (struct sockaddr_in) :
 135       sizeof(struct sockaddr_in6);
 136   ret = connect (sockfd, (const struct sockaddr *)&name, name_len);
 137
 138 #ifdef G_OS_WIN32
 139   if (ret < 0 && WSAGetLastError () != WSAEINPROGRESS) {
 140     closesocket (sockfd);
 141 #else
 142   if (ret < 0 && errno != EINPROGRESS) {
 143     close (sockfd);
 144 #endif
 145     g_slice_free (NiceSocket, sock);
 146     return NULL;
 147   }
 148
 149   name_len = name.ss_family == AF_INET? sizeof (struct sockaddr_in) :
 150       sizeof(struct sockaddr_in6);
 151   if (getsockname (sockfd, (struct sockaddr *) &name, &name_len) < 0) {
 152     g_slice_free (NiceSocket, sock);
 153 #ifdef G_OS_WIN32
 154     closesocket(sockfd);
 155 #else
 156     close (sockfd);
 157 #endif
 158     return NULL;
 159   }
 160
 161   nice_address_set_from_sockaddr (&sock->addr, (struct sockaddr *)&name);
 162
 163   sock->priv = priv = g_slice_new0 (TcpPriv);
 164
 165   priv->agent = agent;
 166   priv->context = ctx;
 167   priv->server_addr = *addr;
 168
 169   sock->fileno = sockfd;
 170   sock->send = socket_send;
 171   sock->recv = socket_recv;
 172   sock->is_reliable = socket_is_reliable;
 173   sock->close = socket_close;
 174
 175   return sock;
 176 }
 177
 178
 179 static void
 180 socket_close (NiceSocket *sock)
 181 {
 182   TcpPriv *priv = sock->priv;
 183
 184 #ifdef G_OS_WIN32
 185   closesocket(sock->fileno);
 186 #else
 187   close (sock->fileno);
 188 #endif
 189   if (priv->io_source) {
 190     g_source_destroy (priv->io_source);
 191     g_source_unref (priv->io_source);
 192   }
 193   if (priv->io_channel)
 194     g_io_channel_unref (priv->io_channel);
 195   g_queue_foreach (&priv->send_queue, (GFunc) free_to_be_sent, NULL);
 196   g_queue_clear (&priv->send_queue);
 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
 207   ret = recv (sock->fileno, buf, len, 0);
 208
 209   /* recv returns 0 when the peer performed a shutdown.. we must return -1 here
 210    * so that the agent destroys the g_source */
 211   if (ret == 0)
 212     return -1;
 213
 214   if (ret < 0) {
 215 #ifdef G_OS_WIN32
 216     if (WSAGetLastError () == WSAEWOULDBLOCK)
 217 #else
 218     if (errno == EAGAIN)
 219 #endif
 220       return 0;
 221     else
 222       return ret;
 223   }
 224
 225   if (from)
 226     *from = priv->server_addr;
 227   return ret;
 228 }
 229
 230 /* Data sent to this function must be a single entity because buffers can be
 231  * dropped if the bandwidth isn't fast enough. So do not send a message in
 232  * multiple chunks. */
 233 static gboolean
 234 socket_send (NiceSocket *sock, const NiceAddress *to,
 235     guint len, const gchar *buf)
 236 {
 237   TcpPriv *priv = sock->priv;
 238   int ret;
 239
 240   /* First try to send the data, don't send it later if it can be sent now
 241      this way we avoid allocating memory on every send */
 242   if (g_queue_is_empty (&priv->send_queue)) {
 243     ret = send (sock->fileno, buf, len, 0);
 244
 245     if (ret < 0) {
 246 #ifdef G_OS_WIN32
 247       if (WSAGetLastError () == WSAEWOULDBLOCK) {
 248 #else
 249       if (errno == EAGAIN) {
 250 #endif
 251         add_to_be_sent (sock, buf, len, FALSE);
 252         return TRUE;
 253       } else {
 254         return FALSE;
 255       }
 256     } else if ((guint)ret < len) {
 257       add_to_be_sent (sock, buf + ret, len - ret, TRUE);
 258       return TRUE;
 259     }
 260   } else {
 261     if (g_queue_get_length(&priv->send_queue) >= MAX_QUEUE_LENGTH) {
 262       int peek_idx = 0;
 263       struct to_be_sent *tbs = NULL;
 264       while ((tbs = g_queue_peek_nth (&priv->send_queue, peek_idx)) != NULL) {
 265         if (tbs->can_drop) {
 266           tbs = g_queue_pop_nth (&priv->send_queue, peek_idx);
 267           g_free (tbs->buf);
 268           g_slice_free (struct to_be_sent, tbs);
 269           break;
 270         } else {
 271           peek_idx++;
 272         }
 273       }
 274     }
 275     add_to_be_sent (sock, buf, len, FALSE);
 276   }
 277
 278   return TRUE;
 279 }
 280
 281 static gboolean
 282 socket_is_reliable (NiceSocket *sock)
 283 {
 284   return TRUE;
 285 }
 286
 287
 288 /*
 289  * Returns:
 290  * -1 = error
 291  * 0 = have more to send
 292  * 1 = sent everything
 293  */
 294
 295 static gboolean
 296 socket_send_more (
 297   GIOChannel *source,
 298   GIOCondition condition,
 299   gpointer data)
 300 {
 301   NiceSocket *sock = (NiceSocket *) data;
 302   TcpPriv *priv = sock->priv;
 303   struct to_be_sent *tbs = NULL;
 304
 305   agent_lock ();
 306
 307   if (g_source_is_destroyed (g_main_current_source ())) {
 308     nice_debug ("Source was destroyed. "
 309         "Avoided race condition in tcp-bsd.c:socket_send_more");
 310     agent_unlock ();
 311     return FALSE;
 312   }
 313
 314   while ((tbs = g_queue_pop_head (&priv->send_queue)) != NULL) {
 315     int ret;
 316
 317     ret = send (sock->fileno, tbs->buf, tbs->length, 0);
 318
 319     if (ret < 0) {
 320 #ifdef G_OS_WIN32
 321       if (WSAGetLastError () == WSAEWOULDBLOCK) {
 322 #else
 323       if (errno == EAGAIN) {
 324 #endif
 325         add_to_be_sent (sock, tbs->buf, tbs->length, TRUE);
 326         g_free (tbs->buf);
 327         g_slice_free (struct to_be_sent, tbs);
 328         break;
 329       }
 330     } else if (ret < (int) tbs->length) {
 331       add_to_be_sent (sock, tbs->buf + ret, tbs->length - ret, TRUE);
 332       g_free (tbs->buf);
 333       g_slice_free (struct to_be_sent, tbs);
 334       break;
 335     }
 336
 337     g_free (tbs->buf);
 338     g_slice_free (struct to_be_sent, tbs);
 339   }
 340
 341   if (g_queue_is_empty (&priv->send_queue)) {
 342     g_io_channel_unref (priv->io_channel);
 343     priv->io_channel = NULL;
 344     g_source_destroy (priv->io_source);
 345     g_source_unref (priv->io_source);
 346     priv->io_source = NULL;
 347
 348     agent_unlock ();
 349     return FALSE;
 350   }
 351
 352   agent_unlock ();
 353   return TRUE;
 354 }
 355
 356
 357 static void
 358 add_to_be_sent (NiceSocket *sock, const gchar *buf, guint len, gboolean head)
 359 {
 360   TcpPriv *priv = sock->priv;
 361   struct to_be_sent *tbs = NULL;
 362
 363   if (len <= 0)
 364     return;
 365
 366   tbs = g_slice_new0 (struct to_be_sent);
 367   tbs->buf = g_memdup (buf, len);
 368   tbs->length = len;
 369   tbs->can_drop = !head;
 370   if (head)
 371     g_queue_push_head (&priv->send_queue, tbs);
 372   else
 373     g_queue_push_tail (&priv->send_queue, tbs);
 374
 375   if (priv->io_channel == NULL) {
 376     priv->io_channel = g_io_channel_unix_new (sock->fileno);
 377     priv->io_source = g_io_create_watch (priv->io_channel, G_IO_OUT);
 378     g_source_set_callback (priv->io_source, (GSourceFunc) socket_send_more,
 379         sock, NULL);
 380     g_source_attach (priv->io_source, priv->context);
 381   }
 382 }
 383
 384
 385
 386 static void
 387 free_to_be_sent (struct to_be_sent *tbs)
 388 {
 389   g_free (tbs->buf);
 390   g_slice_free (struct to_be_sent, tbs);
 391 }
 392