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 *
nice_tcp_bsd_socket_new_from_gsock(GMainContext * ctx,GSocket * gsock,NiceAddress * local_addr,NiceAddress * remote_addr,gboolean reliable)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 *
nice_tcp_bsd_socket_new(GMainContext * ctx,NiceAddress * local_addr,NiceAddress * remote_addr,gboolean reliable)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
socket_close(NiceSocket * sock)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
socket_recv_messages(NiceSocket * sock,NiceInputMessage * recv_messages,guint n_recv_messages)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
socket_send_message(NiceSocket * sock,const NiceOutputMessage * message,gboolean reliable)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
socket_send_messages(NiceSocket * sock,const NiceAddress * to,const NiceOutputMessage * messages,guint n_messages)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
socket_send_messages_reliable(NiceSocket * sock,const NiceAddress * to,const NiceOutputMessage * messages,guint n_messages)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
socket_is_reliable(NiceSocket * sock)407 socket_is_reliable (NiceSocket *sock)
408 {
409 TcpPriv *priv = sock->priv;
410
411 return priv->reliable;
412 }
413
414 static gboolean
socket_can_send(NiceSocket * sock,NiceAddress * addr)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
socket_set_writable_callback(NiceSocket * sock,NiceSocketWritableCb callback,gpointer user_data)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
socket_send_more(GSocket * gsocket,GIOCondition condition,gpointer data)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
nice_tcp_bsd_socket_set_passive_parent(NiceSocket * sock,NiceSocket * passive_parent)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 *
nice_tcp_bsd_socket_get_passive_parent(NiceSocket * sock)483 nice_tcp_bsd_socket_get_passive_parent (NiceSocket *sock)
484 {
485 TcpPriv *priv = sock->priv;
486
487 return priv->passive_parent;
488 }
489