From 9497d78fb13d1d2648dea0211b06c8fd79435e03 Mon Sep 17 00:00:00 2001
From: Youness Alaoui <youness.alaoui@collabora.co.uk>
Date: Thu, 11 Jun 2009 14:07:54 -0400
Subject: [PATCH] Add support for droping packets when using TCP and the
 bandwidth is too slow for the data output. The queue now has a limit of 20
 packets and the tcp-turn now has to call send in a single shot to avoid
 dropping parts of the packet (dropping the length but send the data, or
 dropping only the padding at the end, etc..)

---
 TODO              |  1 -
 socket/tcp-bsd.c  | 25 ++++++++++++++++++++++---
 socket/tcp-turn.c | 25 ++++++++++++-------------
 3 files changed, 34 insertions(+), 17 deletions(-)

diff --git a/TODO b/TODO
index c3c30ed..eaab911 100644
--- a/TODO
+++ b/TODO
@@ -2,7 +2,6 @@ check for the cookie and act accordingly for incoming messages.
 channel bind have a 10 minutes expiration :@
 nice_socket_recv returns -1 means we must close the nice_socket and stop all connchecks/candidates and reelect if was eleected...
 Add HTTP Digest support
-tcp turn must drop packets if the queue fills too fast
 make sure the timers don't race condition with frees, and have some kind of refcounting (like in turn send requests) for timers
 Add support for TURN-13
 
diff --git a/socket/tcp-bsd.c b/socket/tcp-bsd.c
index 2c41d75..2b829ba 100644
--- a/socket/tcp-bsd.c
+++ b/socket/tcp-bsd.c
@@ -70,8 +70,10 @@ typedef struct {
 struct to_be_sent {
   guint length;
   gchar *buf;
+  gboolean can_drop;
 };
 
+#define MAX_QUEUE_LENGTH 20
 
 static void socket_close (NiceSocket *sock);
 static gint socket_recv (NiceSocket *sock, NiceAddress *from,
@@ -222,7 +224,9 @@ socket_recv (NiceSocket *sock, NiceAddress *from, guint len, gchar *buf)
   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 gboolean
 socket_send (NiceSocket *sock, const NiceAddress *to,
     guint len, const gchar *buf)
@@ -247,10 +251,24 @@ socket_send (NiceSocket *sock, const NiceAddress *to,
         return FALSE;
       }
     } else if ((guint)ret < len) {
-      add_to_be_sent (sock, buf + ret, len - ret, FALSE);
+      add_to_be_sent (sock, buf + ret, len - ret, TRUE);
       return TRUE;
     }
   } else {
+    if (g_queue_get_length(&priv->send_queue) >= MAX_QUEUE_LENGTH) {
+      int peek_idx = 0;
+      struct to_be_sent *tbs = NULL;
+      while ((tbs = g_queue_peek_nth (&priv->send_queue, peek_idx)) != NULL) {
+        if (tbs->can_drop) {
+          tbs = g_queue_pop_nth (&priv->send_queue, peek_idx);
+          g_free (tbs->buf);
+          g_slice_free (struct to_be_sent, tbs);
+          break;
+        } else {
+          peek_idx++;
+        }
+      }
+    }
     add_to_be_sent (sock, buf, len, FALSE);
   }
 
@@ -283,7 +301,7 @@ socket_send_more (
 
   g_static_rec_mutex_lock (&priv->agent->mutex);
 
-  while ((tbs = g_queue_pop_head (&priv->send_queue))) {
+  while ((tbs = g_queue_pop_head (&priv->send_queue)) != NULL) {
     int ret;
 
     ret = send (sock->fileno, tbs->buf, tbs->length, 0);
@@ -338,6 +356,7 @@ add_to_be_sent (NiceSocket *sock, const gchar *buf, guint len, gboolean head)
   tbs = g_slice_new0 (struct to_be_sent);
   tbs->buf = g_memdup (buf, len);
   tbs->length = len;
+  tbs->can_drop = !head;
   if (head)
     g_queue_push_head (&priv->send_queue, tbs);
   else
diff --git a/socket/tcp-turn.c b/socket/tcp-turn.c
index b0306da..0d0dc53 100644
--- a/socket/tcp-turn.c
+++ b/socket/tcp-turn.c
@@ -65,6 +65,7 @@ typedef struct {
   NiceSocket *base_socket;
 } TurnTcpPriv;
 
+#define MAX_UDP_MESSAGE_SIZE 65535
 
 static void socket_close (NiceSocket *sock);
 static gint socket_recv (NiceSocket *sock, NiceAddress *from,
@@ -183,32 +184,30 @@ static gboolean
 socket_send (NiceSocket *sock, const NiceAddress *to,
     guint len, const gchar *buf)
 {
-  gboolean ret = TRUE;
   TurnTcpPriv *priv = sock->priv;
   gchar padbuf[3] = {0, 0, 0};
   int padlen = (len%4) ? 4 - (len%4) : 0;
+  gchar buffer[MAX_UDP_MESSAGE_SIZE + sizeof(guint16) + sizeof(padbuf)];
+  guint buffer_len = 0;
 
   if (priv->compatibility != NICE_TURN_SOCKET_COMPATIBILITY_DRAFT9)
     padlen = 0;
 
   if (priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_GOOGLE) {
     guint16 tmpbuf = htons (len);
-    ret = nice_socket_send (priv->base_socket, to,
-        sizeof(guint16), (gchar *)&tmpbuf);
-
-    if (!ret)
-      return ret;
+    memcpy (buffer + buffer_len, (gchar *)&tmpbuf, sizeof(guint16));
+    buffer_len += sizeof(guint16);
   }
 
-  ret = nice_socket_send (priv->base_socket, to, len, buf);
-
-  if (!ret)
-    return ret;
+  memcpy (buffer + buffer_len, buf, len);
+  buffer_len += len;
 
-  if (priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_DRAFT9)
-    ret = nice_socket_send (priv->base_socket, to, padlen, padbuf);
+  if (priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_DRAFT9) {
+    memcpy (buffer + buffer_len, padbuf, padlen);
+    buffer_len += padlen;
+  }
+  return nice_socket_send (priv->base_socket, to, buffer_len, buffer);
 
-  return ret;
 }
 
 
-- 
2.11.4.GIT