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