tcp-bsd.c 12.8 KB
Newer Older
1 2 3
/*
 * This file is part of the Nice GLib ICE library.
 *
4 5 6
 * (C) 2008-2009 Collabora Ltd.
 *  Contact: Youness Alaoui
 * (C) 2008-2009 Nokia Corporation. All rights reserved.
7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23
 *
 * The contents of this file are subject to the Mozilla Public License Version
 * 1.1 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * http://www.mozilla.org/MPL/
 *
 * Software distributed under the License is distributed on an "AS IS" basis,
 * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
 * for the specific language governing rights and limitations under the
 * License.
 *
 * The Original Code is the Nice GLib ICE library.
 *
 * The Initial Developers of the Original Code are Collabora Ltd and Nokia
 * Corporation. All Rights Reserved.
 *
 * Contributors:
24
 *   Youness Alaoui, Collabora Ltd.
25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46
 *
 * Alternatively, the contents of this file may be used under the terms of the
 * the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
 * case the provisions of LGPL are applicable instead of those above. If you
 * wish to allow use of your version of this file only under the terms of the
 * LGPL and not to allow others to use your version of this file under the
 * MPL, indicate your decision by deleting the provisions above and replace
 * them with the notice and other provisions required by the LGPL. If you do
 * not delete the provisions above, a recipient may use your version of this
 * file under either the MPL or the LGPL.
 */

/*
 * Implementation of TCP relay socket interface using TCP Berkeley sockets. (See
 * http://en.wikipedia.org/wiki/Berkeley_sockets.)
 */
#ifdef HAVE_CONFIG_H
# include "config.h"
#endif

#include "tcp-bsd.h"
#include "agent-priv.h"
47
#include "socket-priv.h"
48

49 50
#include "tcp-passive.h"

51 52 53 54 55 56 57 58
#include <string.h>
#include <errno.h>
#include <fcntl.h>

#ifndef G_OS_WIN32
#include <unistd.h>
#endif

59 60 61
/* FIXME: This should be defined in gio/gnetworking.h, which we should include;
 * but we cannot do that without refactoring.
 * (See: https://phabricator.freedesktop.org/D230). */
Nirbheek Chauhan's avatar
Nirbheek Chauhan committed
62
#undef TCP_NODELAY
63 64
#define TCP_NODELAY 1

65 66
static GMutex mutex;

67
typedef struct {
68
  NiceAddress remote_addr;
69 70 71
  GQueue send_queue;
  GMainContext *context;
  GSource *io_source;
72
  gboolean error;
73
  gboolean reliable;
74 75
  NiceSocketWritableCb writable_cb;
  gpointer writable_data;
76
  NiceSocket *passive_parent;
77 78
} TcpPriv;

79
#define MAX_QUEUE_LENGTH 20
80 81

static void socket_close (NiceSocket *sock);
82 83
static gint socket_recv_messages (NiceSocket *sock,
    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 87
static gint socket_send_messages_reliable (NiceSocket *sock,
    const NiceAddress *to, const NiceOutputMessage *messages, guint n_messages);
88
static gboolean socket_is_reliable (NiceSocket *sock);
89 90 91
static gboolean socket_can_send (NiceSocket *sock, NiceAddress *addr);
static void socket_set_writable_callback (NiceSocket *sock,
    NiceSocketWritableCb callback, gpointer user_data);
92

93
static gboolean socket_send_more (GSocket *gsocket, GIOCondition condition,
94 95
    gpointer data);

96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113
NiceSocket *
nice_tcp_bsd_socket_new_from_gsock (GMainContext *ctx, GSocket *gsock,
    NiceAddress *local_addr, NiceAddress *remote_addr, gboolean reliable)
{
  NiceSocket *sock;
  TcpPriv *priv;

  g_return_val_if_fail (G_IS_SOCKET (gsock), NULL);

  sock = g_slice_new0 (NiceSocket);
  sock->priv = priv = g_slice_new0 (TcpPriv);

  if (ctx == NULL)
    ctx = g_main_context_default ();
  priv->context = g_main_context_ref (ctx);
  priv->remote_addr = *remote_addr;
  priv->error = FALSE;
  priv->reliable = reliable;
114 115
  priv->writable_cb = NULL;
  priv->writable_data = NULL;
116 117 118 119 120 121 122 123

  sock->type = NICE_SOCKET_TYPE_TCP_BSD;
  sock->fileno = g_object_ref (gsock);
  sock->addr = *local_addr;
  sock->send_messages = socket_send_messages;
  sock->send_messages_reliable = socket_send_messages_reliable;
  sock->recv_messages = socket_recv_messages;
  sock->is_reliable = socket_is_reliable;
124 125
  sock->can_send = socket_can_send;
  sock->set_writable_callback = socket_set_writable_callback;
126 127 128 129 130
  sock->close = socket_close;

  return sock;
}

131
NiceSocket *
132 133
nice_tcp_bsd_socket_new (GMainContext *ctx, NiceAddress *local_addr,
    NiceAddress *remote_addr, gboolean reliable)
134
{
135 136 137 138
  union {
    struct sockaddr_storage storage;
    struct sockaddr addr;
  } name;
139
  NiceSocket *sock;
140 141
  GSocket *gsock = NULL;
  GError *gerr = NULL;
142
  gboolean gret = FALSE;
143
  GSocketAddress *gaddr;
144

145
  if (remote_addr == NULL) {
146 147
    /* We can't connect a tcp socket with no destination address */
    return NULL;
148 149
  }

150
  nice_address_copy_to_sockaddr (remote_addr, &name.addr);
151

152 153 154
  if (name.storage.ss_family == AF_UNSPEC || name.storage.ss_family == AF_INET) {
    gsock = g_socket_new (G_SOCKET_FAMILY_IPV4, G_SOCKET_TYPE_STREAM,
        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 161 162 163
  } else if (name.storage.ss_family == AF_INET6) {
    gsock = g_socket_new (G_SOCKET_FAMILY_IPV6, G_SOCKET_TYPE_STREAM,
        G_SOCKET_PROTOCOL_TCP, NULL);
    name.storage.ss_family = AF_INET6;
164
#ifdef HAVE_SA_LEN
165
    name.storage.ss_len = sizeof (struct sockaddr_in6);
166 167 168
#endif
  }

Youness Alaoui's avatar
Youness Alaoui committed
169
  if (gsock == NULL) {
170
    return NULL;
Youness Alaoui's avatar
Youness Alaoui committed
171
  }
172

173 174 175 176 177 178
  gaddr = g_socket_address_new_from_native (&name.addr, sizeof (name));
  if (gaddr == NULL) {
    g_object_unref (gsock);
    return NULL;
  }

179
  /* GSocket: All socket file descriptors are set to be close-on-exec. */
180 181
  g_socket_set_blocking (gsock, false);

182 183 184
  /* setting TCP_NODELAY to TRUE in order to avoid packet batching */
  g_socket_set_option (gsock, IPPROTO_TCP, TCP_NODELAY, TRUE, NULL);

185 186
  gret = g_socket_connect (gsock, gaddr, NULL, &gerr);
  g_object_unref (gaddr);
187

188 189
  if (gret == FALSE) {
    if (g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_PENDING) == FALSE) {
190
      g_error_free (gerr);
191 192 193 194
      g_socket_close (gsock, NULL);
      g_object_unref (gsock);
      return NULL;
    }
195
    g_error_free (gerr);
196
  }
197

198 199 200
  nice_address_copy_to_sockaddr (local_addr, &name.addr);
  gaddr = g_socket_address_new_from_native (&name.addr, sizeof (name));
  if (gaddr == NULL) {
Youness Alaoui's avatar
Youness Alaoui committed
201 202
    g_socket_close (gsock, NULL);
    g_object_unref (gsock);
203 204
    return NULL;
  }
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 210
      reliable);
  g_object_unref (gsock);
211 212 213 214 215 216 217 218 219 220

  return sock;
}


static void
socket_close (NiceSocket *sock)
{
  TcpPriv *priv = sock->priv;

221 222
  g_mutex_lock (&mutex);

Youness Alaoui's avatar
Youness Alaoui committed
223 224 225
  if (sock->fileno) {
    g_socket_close (sock->fileno, NULL);
    g_object_unref (sock->fileno);
226 227
    sock->fileno = NULL;
  }
228 229 230 231
  if (priv->io_source) {
    g_source_destroy (priv->io_source);
    g_source_unref (priv->io_source);
  }
232

233 234 235 236
  if (priv->passive_parent) {
    nice_tcp_passive_socket_remove_connection (priv->passive_parent, &priv->remote_addr);
  }

237
  nice_socket_free_send_queue (&priv->send_queue);
238

239 240 241
  if (priv->context)
    g_main_context_unref (priv->context);

242
  g_mutex_unlock (&mutex);
243

244 245 246 247
  g_slice_free(TcpPriv, sock->priv);
}

static gint
248 249
socket_recv_messages (NiceSocket *sock,
    NiceInputMessage *recv_messages, guint n_recv_messages)
250 251
{
  TcpPriv *priv = sock->priv;
252
  guint i;
253

254 255
  /* Make sure socket has not been freed: */
  g_assert (sock->priv != NULL);
256

257 258 259 260
  /* Don't try to access the socket if it had an error */
  if (priv->error)
    return -1;

261 262 263 264
  for (i = 0; i < n_recv_messages; i++) {
    gint flags = G_SOCKET_MSG_NONE;
    GError *gerr = NULL;
    gssize len;
265

266 267 268 269 270 271 272 273 274 275 276 277
    len = g_socket_receive_message (sock->fileno, NULL,
        recv_messages[i].buffers, recv_messages[i].n_buffers,
        NULL, NULL, &flags, NULL, &gerr);

    recv_messages[i].length = MAX (len, 0);

    /* recv returns 0 when the peer performed a shutdown.. we must return -1
     * here so that the agent destroys the g_source */
    if (len == 0) {
      priv->error = TRUE;
      break;
    }
278

279 280 281
    if (len < 0) {
      if (g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
        len = 0;
282

283 284 285 286 287
      g_error_free (gerr);
      return len;
    }

    if (recv_messages[i].from)
288
      *recv_messages[i].from = priv->remote_addr;
289 290
  }

291 292 293 294 295
  /* Was there an error processing the first message? */
  if (priv->error && i == 0)
    return -1;

  return i;
296 297
}

298
static gssize
299 300
socket_send_message (NiceSocket *sock,
    const NiceOutputMessage *message, gboolean reliable)
301 302
{
  TcpPriv *priv = sock->priv;
303
  gssize ret;
304
  GError *gerr = NULL;
305
  gsize message_len;
306

307 308
  /* Make sure socket has not been freed: */
  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 313 314
  if (priv->error)
    return -1;

315 316
  message_len = output_message_get_size (message);

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 321 322
    ret = g_socket_send_message (sock->fileno, NULL, message->buffers,
        message->n_buffers, NULL, 0, G_SOCKET_MSG_NONE, NULL, &gerr);

323
    if (ret < 0) {
324 325 326
      if (g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK) ||
          g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_FAILED)) {
        /* Queue the message and send it later. */
327 328
        nice_socket_queue_send_with_callback (&priv->send_queue,
            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 337
      nice_socket_queue_send_with_callback (&priv->send_queue,
          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 343 344
    /* Only queue if we're sending reliably  */
    if (reliable) {
      /* Queue the message and send it later. */
345 346
      nice_socket_queue_send_with_callback (&priv->send_queue,
          message, 0, message_len, FALSE, sock->fileno, &priv->io_source,
347
          priv->context, socket_send_more, sock);
348 349 350 351
      ret = message_len;
    } else {
      /* non reliable send, so we shouldn't queue the message */
      ret = 0;
352
    }
353 354
  }

355 356 357 358 359 360 361
  return ret;
}

/* Data sent to this function must be a single entity because buffers can be
 * dropped if the bandwidth isn't fast enough. So do not send a message in
 * multiple chunks. */
static gint
362 363
socket_send_messages (NiceSocket *sock, const NiceAddress *to,
    const NiceOutputMessage *messages, guint n_messages)
364 365 366
{
  guint i;

367 368
  /* Make sure socket has not been freed: */
  g_assert (sock->priv != NULL);
369

370 371 372 373
  for (i = 0; i < n_messages; i++) {
    const NiceOutputMessage *message = &messages[i];
    gssize len;

374
    len = socket_send_message (sock, message, FALSE);
375 376 377

    if (len < 0) {
      /* Error. */
378 379
      if (i > 0)
        break;
380 381 382 383 384 385 386 387
      return len;
    } else if (len == 0) {
      /* EWOULDBLOCK. */
      break;
    }
  }

  return i;
388 389
}

390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405
static gint
socket_send_messages_reliable (NiceSocket *sock, const NiceAddress *to,
    const NiceOutputMessage *messages, guint n_messages)
{
  guint i;

  for (i = 0; i < n_messages; i++) {
    if (socket_send_message (sock, &messages[i], TRUE) < 0) {
      /* Error. */
      return -1;
    }
  }

  return i;
}

406 407 408
static gboolean
socket_is_reliable (NiceSocket *sock)
{
409 410 411
  TcpPriv *priv = sock->priv;

  return priv->reliable;
412 413
}

414 415 416 417
static gboolean
socket_can_send (NiceSocket *sock, NiceAddress *addr)
{
  TcpPriv *priv = sock->priv;
418

419 420 421 422 423 424 425 426 427 428 429 430
  return g_queue_is_empty (&priv->send_queue);
}

static void
socket_set_writable_callback (NiceSocket *sock,
    NiceSocketWritableCb callback, gpointer user_data)
{
  TcpPriv *priv = sock->priv;

  priv->writable_cb = callback;
  priv->writable_data = user_data;
}
431 432 433

static gboolean
socket_send_more (
434
  GSocket *gsocket,
435 436 437 438 439 440
  GIOCondition condition,
  gpointer data)
{
  NiceSocket *sock = (NiceSocket *) data;
  TcpPriv *priv = sock->priv;

441
  g_mutex_lock (&mutex);
442 443 444 445

  if (g_source_is_destroyed (g_main_current_source ())) {
    nice_debug ("Source was destroyed. "
        "Avoided race condition in tcp-bsd.c:socket_send_more");
446
    g_mutex_unlock (&mutex);
447 448
    return FALSE;
  }
449

450 451 452 453
  /* connection hangs up or queue was emptied */
  if (condition & G_IO_HUP ||
      nice_socket_flush_send_queue_to_socket (sock->fileno,
          &priv->send_queue)) {
454 455 456 457
    g_source_destroy (priv->io_source);
    g_source_unref (priv->io_source);
    priv->io_source = NULL;

458
    g_mutex_unlock (&mutex);
459 460 461 462

    if (priv->writable_cb)
      priv->writable_cb (sock, priv->writable_data);

463 464 465
    return FALSE;
  }

466
  g_mutex_unlock (&mutex);
467 468
  return TRUE;
}
469 470 471 472 473 474 475 476 477 478

void
nice_tcp_bsd_socket_set_passive_parent (NiceSocket *sock, NiceSocket *passive_parent)
{
  TcpPriv *priv = sock->priv;

  g_assert (priv->passive_parent == NULL);

  priv->passive_parent = passive_parent;
}