queue transmits to tunnels

3 files changed, 287 insertions, 47 deletions

diff --git a/src/vpn/gnunet-daemon-exit.c b/src/vpn/gnunet-daemon-exit.c
index d12c59d20..cf9f58be3 100644
--- a/src/vpn/gnunet-daemon-exit.c
+++ b/src/vpn/gnunet-daemon-exit.c
@@ -150,6 +150,14 @@ struct redirect_state
 static struct GNUNET_CONTAINER_MultiHashMap *udp_services;
 static struct GNUNET_CONTAINER_MultiHashMap *tcp_services;
 
+struct tunnel_notify_queue
+{
+  struct tunnel_notify_queue* next;
+  struct tunnel_notify_queue* prev;
+  void* cls;
+  size_t len;
+};
+
 /**
  * Function that frees everything from a hashmap
  */
@@ -223,11 +231,37 @@ hash_redirect_info(GNUNET_HashCode* hash, struct redirect_info* u_i, size_t addr
 static size_t
 send_udp_to_peer_notify_callback (void *cls, size_t size, void *buf)
 {
-  struct GNUNET_MessageHeader *hdr = cls;
+  struct GNUNET_MESH_Tunnel** tunnel = cls;
+  GNUNET_MESH_tunnel_set_data(*tunnel, NULL);
+  struct GNUNET_MessageHeader *hdr = (struct GNUNET_MessageHeader*)(tunnel + 1);
   GNUNET_assert (size >= ntohs (hdr->size));
   memcpy (buf, hdr, ntohs (hdr->size));
   size = ntohs(hdr->size);
   GNUNET_free (cls);
+
+  if (NULL != GNUNET_MESH_tunnel_get_head(*tunnel))
+    {
+      struct tunnel_notify_queue* element = GNUNET_MESH_tunnel_get_head(*tunnel);
+      struct tunnel_notify_queue* head = GNUNET_MESH_tunnel_get_head(*tunnel);
+      struct tunnel_notify_queue* tail = GNUNET_MESH_tunnel_get_tail(*tunnel);
+
+      GNUNET_CONTAINER_DLL_remove(head, tail, element);
+
+      GNUNET_MESH_tunnel_set_head(*tunnel, head);
+      GNUNET_MESH_tunnel_set_tail(*tunnel, tail);
+
+      struct GNUNET_MESH_TransmitHandle* th = GNUNET_MESH_notify_transmit_ready (*tunnel,
+                                                                                 GNUNET_NO,
+                                                                                 42,
+                                                                                 GNUNET_TIME_relative_divide
+                                                                                 (GNUNET_CONSTANTS_MAX_CORK_DELAY, 2),
+                                                                                 (const struct GNUNET_PeerIdentity *)
+                                                                                 NULL, element->len,
+                                                                                 send_udp_to_peer_notify_callback, element->cls);
+      /* save the handle */
+      GNUNET_MESH_tunnel_set_data(*tunnel, th);
+    }
+
   return size;
 }
 
@@ -301,7 +335,9 @@ udp_from_helper (struct udp_pkt *udp, unsigned char *dadr, size_t addrlen)
   len =
     sizeof (struct GNUNET_MessageHeader) + sizeof (GNUNET_HashCode) +
     ntohs (udp->len);
-  msg = GNUNET_malloc (len);
+  struct GNUNET_MESH_Tunnel** ctunnel = GNUNET_malloc (sizeof(struct GNUNET_MESH_TUNNEL*) + len);
+  *ctunnel = tunnel;
+  msg = (struct GNUNET_MessageHeader*)(ctunnel + 1);
   msg->size = htons (len);
   msg->type = htons (state->type == SERVICE ? GNUNET_MESSAGE_TYPE_SERVICE_UDP_BACK : GNUNET_MESSAGE_TYPE_REMOTE_UDP_BACK);
   GNUNET_HashCode *desc = (GNUNET_HashCode *) (msg + 1);
@@ -312,14 +348,33 @@ udp_from_helper (struct udp_pkt *udp, unsigned char *dadr, size_t addrlen)
   void *_udp = desc + 1;
   memcpy (_udp, udp, ntohs (udp->len));
 
-  GNUNET_MESH_notify_transmit_ready (tunnel,
-                                     GNUNET_NO,
-                                     42,
-                                     GNUNET_TIME_relative_divide
-                                     (GNUNET_CONSTANTS_MAX_CORK_DELAY, 2),
-                                     (const struct GNUNET_PeerIdentity *)
-                                     NULL, len,
-                                     send_udp_to_peer_notify_callback, msg);
+  if (NULL == GNUNET_MESH_tunnel_get_data(tunnel))
+    {
+      /* No notify is pending */
+      struct GNUNET_MESH_TransmitHandle* th = GNUNET_MESH_notify_transmit_ready (tunnel,
+                                                                                 GNUNET_NO,
+                                                                                 42,
+                                                                                 GNUNET_TIME_relative_divide
+                                                                                 (GNUNET_CONSTANTS_MAX_CORK_DELAY, 2),
+                                                                                 (const struct GNUNET_PeerIdentity *)
+                                                                                 NULL, len,
+                                                                                 send_udp_to_peer_notify_callback, ctunnel);
+      /* save the handle */
+      GNUNET_MESH_tunnel_set_data(tunnel, th);
+    }
+  else
+    {
+      struct tunnel_notify_queue* head = GNUNET_MESH_tunnel_get_head(tunnel);
+      struct tunnel_notify_queue* tail = GNUNET_MESH_tunnel_get_tail(tunnel);
+
+      struct tunnel_notify_queue* element = GNUNET_malloc(sizeof(struct tunnel_notify_queue));
+      element->cls = ctunnel;
+      element->len = len;
+
+      GNUNET_CONTAINER_DLL_insert_tail(head, tail, element);
+      GNUNET_MESH_tunnel_set_head(tunnel, head);
+      GNUNET_MESH_tunnel_set_tail(tunnel, tail);
+    }
 }
 
 /**
@@ -378,7 +433,9 @@ tcp_from_helper (struct tcp_pkt *tcp, unsigned char *dadr, size_t addrlen,
   len =
     sizeof (struct GNUNET_MessageHeader) + sizeof (GNUNET_HashCode) + pktlen;
   GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "len: %d\n", pktlen);
-  msg = GNUNET_malloc (len);
+  struct GNUNET_MESH_Tunnel** ctunnel = GNUNET_malloc (sizeof(struct GNUNET_MESH_TUNNEL*) + len);
+  *ctunnel = tunnel;
+  msg = (struct GNUNET_MessageHeader*)(ctunnel + 1);
   msg->size = htons (len);
   msg->type = htons (state->type == SERVICE ? GNUNET_MESSAGE_TYPE_SERVICE_TCP_BACK : GNUNET_MESSAGE_TYPE_REMOTE_TCP_BACK);
   GNUNET_HashCode *desc = (GNUNET_HashCode *) (msg + 1);
@@ -389,14 +446,31 @@ tcp_from_helper (struct tcp_pkt *tcp, unsigned char *dadr, size_t addrlen,
   void *_tcp = desc + 1;
   memcpy (_tcp, tcp, pktlen);
 
-  GNUNET_MESH_notify_transmit_ready (tunnel,
+  if (NULL == GNUNET_MESH_tunnel_get_data(tunnel))
+    {
+      /* No notify is pending */
+      struct GNUNET_MESH_TransmitHandle* th = GNUNET_MESH_notify_transmit_ready (tunnel,
                                      GNUNET_NO,
                                      42,
                                      GNUNET_TIME_relative_divide
                                      (GNUNET_CONSTANTS_MAX_CORK_DELAY, 2),
                                      (const struct GNUNET_PeerIdentity *)NULL,
                                      len, send_udp_to_peer_notify_callback,
-                                     msg);
+                                     ctunnel);
+      /* save the handle */
+      GNUNET_MESH_tunnel_set_data(tunnel, th);
+    }
+  else
+    {
+      struct tunnel_notify_queue* head = GNUNET_MESH_tunnel_get_head(tunnel);
+      struct tunnel_notify_queue* tail = GNUNET_MESH_tunnel_get_tail(tunnel);
+
+      struct tunnel_notify_queue* element = GNUNET_malloc(sizeof(struct tunnel_notify_queue));
+      element->cls = ctunnel;
+      element->len = len;
+
+      GNUNET_CONTAINER_DLL_insert_tail(head, tail, element);
+    }
 }
 
 
diff --git a/src/vpn/gnunet-daemon-vpn.c b/src/vpn/gnunet-daemon-vpn.c
index 9ff43cd7f..ece7b554a 100644
--- a/src/vpn/gnunet-daemon-vpn.c
+++ b/src/vpn/gnunet-daemon-vpn.c
@@ -45,6 +45,14 @@ struct GNUNET_MESH_Handle *mesh_handle;
 struct GNUNET_CONTAINER_MultiHashMap* hashmap;
 static struct GNUNET_CONTAINER_Heap *heap;
 
+struct tunnel_notify_queue
+{
+  struct tunnel_notify_queue* next;
+  struct tunnel_notify_queue* prev;
+  size_t len;
+  void* cls;
+};
+
 /**
  * If there are at least this many address-mappings, old ones will be removed
  */
@@ -185,6 +193,7 @@ static size_t
 send_pkt_to_peer_notify_callback (void *cls, size_t size, void *buf)
 {
   struct GNUNET_MESH_Tunnel **tunnel = cls;
+  GNUNET_MESH_tunnel_set_data(*tunnel, NULL);
   struct GNUNET_MessageHeader *hdr =
     (struct GNUNET_MessageHeader *) (tunnel + 1);
   GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "send_pkt_to_peer_notify_callback: buf = %x; size = %u;\n", buf, size);
@@ -193,6 +202,30 @@ send_pkt_to_peer_notify_callback (void *cls, size_t size, void *buf)
   size = ntohs(hdr->size);
   GNUNET_free (cls);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sent!\n");
+
+  if (NULL != GNUNET_MESH_tunnel_get_head(*tunnel))
+    {
+      struct tunnel_notify_queue* element = GNUNET_MESH_tunnel_get_head(*tunnel);
+      struct tunnel_notify_queue* head = GNUNET_MESH_tunnel_get_head(*tunnel);
+      struct tunnel_notify_queue* tail = GNUNET_MESH_tunnel_get_tail(*tunnel);
+
+      GNUNET_CONTAINER_DLL_remove(head, tail, element);
+
+      GNUNET_MESH_tunnel_set_head(*tunnel, head);
+      GNUNET_MESH_tunnel_set_tail(*tunnel, tail);
+
+      struct GNUNET_MESH_TransmitHandle* th = GNUNET_MESH_notify_transmit_ready (*tunnel,
+                                                                                 GNUNET_NO,
+                                                                                 42,
+                                                                                 GNUNET_TIME_relative_divide
+                                                                                 (GNUNET_CONSTANTS_MAX_CORK_DELAY, 2),
+                                                                                 (const struct GNUNET_PeerIdentity *)
+                                                                                 NULL, element->len,
+                                                                                 send_pkt_to_peer_notify_callback, element->cls);
+      /* save the handle */
+      GNUNET_MESH_tunnel_set_data(*tunnel, th);
+    }
+
   return size;
 }
 
@@ -217,14 +250,32 @@ send_pkt_to_peer (void *cls,
   GNUNET_assert(NULL != tunnel);
   GNUNET_assert(NULL != *tunnel);
 
-  GNUNET_MESH_notify_transmit_ready (*tunnel,
-                                     GNUNET_NO,
-                                     42,
-                                     GNUNET_TIME_relative_divide(GNUNET_CONSTANTS_MAX_CORK_DELAY, 2),
-                                     (const struct GNUNET_PeerIdentity *)NULL,
-                                     ntohs(hdr->size),
-                                     send_pkt_to_peer_notify_callback,
-                                     cls);
+  if (NULL == GNUNET_MESH_tunnel_get_data(*tunnel))
+    {
+      struct GNUNET_MESH_TransmitHandle* th = GNUNET_MESH_notify_transmit_ready (*tunnel,
+                                                                                 GNUNET_NO,
+                                                                                 42,
+                                                                                 GNUNET_TIME_relative_divide(GNUNET_CONSTANTS_MAX_CORK_DELAY, 2),
+                                                                                 (const struct GNUNET_PeerIdentity *)NULL,
+                                                                                 ntohs(hdr->size),
+                                                                                 send_pkt_to_peer_notify_callback,
+                                                                                 cls);
+      GNUNET_MESH_tunnel_set_data(*tunnel, th);
+    }
+  else
+    {
+     struct tunnel_notify_queue* head = GNUNET_MESH_tunnel_get_head(*tunnel);
+     struct tunnel_notify_queue* tail = GNUNET_MESH_tunnel_get_tail(*tunnel);
+     struct tunnel_notify_queue* element = GNUNET_malloc(sizeof *element);
+
+     element->cls = cls;
+     element->len = ntohs(hdr->size);
+
+     GNUNET_CONTAINER_DLL_insert_tail(head, tail, element);
+
+     GNUNET_MESH_tunnel_set_head(*tunnel, head);
+     GNUNET_MESH_tunnel_set_tail(*tunnel, tail);
+    }
 }
 
 /**
diff --git a/src/vpn/gnunet-service-dns.c b/src/vpn/gnunet-service-dns.c
index b2b123a15..c0ef1411d 100644
--- a/src/vpn/gnunet-service-dns.c
+++ b/src/vpn/gnunet-service-dns.c
@@ -25,6 +25,7 @@
 #include "platform.h"
 #include "gnunet_getopt_lib.h"
 #include "gnunet_service_lib.h"
+#include <gnunet_constants.h>
 #include "gnunet_network_lib.h"
 #include "gnunet_os_lib.h"
 #include "gnunet-service-dns-p.h"
@@ -107,6 +108,15 @@ struct receive_dht_cls {
     struct GNUNET_DHT_GetHandle* handle;
 };
 
+struct tunnel_notify_queue
+{
+  struct tunnel_notify_queue* next;
+  struct tunnel_notify_queue* prev;
+  void* cls;
+  size_t len;
+  GNUNET_CONNECTION_TransmitReadyNotify cb;
+};
+
 /**
  * Hijack all outgoing DNS-Traffic but for traffic leaving "our" port.
  */
@@ -212,7 +222,8 @@ mesh_send_response (void *cls, size_t size, void *buf)
   GNUNET_assert (size >= sizeof (struct GNUNET_MessageHeader));
   struct GNUNET_MessageHeader *hdr = buf;
   uint32_t *sz = cls;
-  struct dns_pkt *dns = (struct dns_pkt *) (sz + 1);
+  struct GNUNET_MESH_Tunnel **tunnel = (struct GNUNET_MESH_Tunnel**)(sz+1);
+  struct dns_pkt *dns = (struct dns_pkt *) (tunnel + 1);
   hdr->type = htons (GNUNET_MESSAGE_TYPE_REMOTE_ANSWER_DNS);
   hdr->size = htons (*sz + sizeof (struct GNUNET_MessageHeader));
 
@@ -224,6 +235,28 @@ mesh_send_response (void *cls, size_t size, void *buf)
 
   memcpy (hdr + 1, dns, *sz);
   GNUNET_free (cls);
+
+  if (NULL != GNUNET_MESH_tunnel_get_head(*tunnel))
+    {
+      struct tunnel_notify_queue* element = GNUNET_MESH_tunnel_get_head(*tunnel);
+      struct tunnel_notify_queue* head = GNUNET_MESH_tunnel_get_head(*tunnel);
+      struct tunnel_notify_queue* tail = GNUNET_MESH_tunnel_get_tail(*tunnel);
+
+      GNUNET_CONTAINER_DLL_remove(head, tail, element);
+
+      GNUNET_MESH_tunnel_set_head(*tunnel, head);
+      GNUNET_MESH_tunnel_set_tail(*tunnel, tail);
+      struct GNUNET_MESH_TransmitHandle* th = GNUNET_MESH_notify_transmit_ready (*tunnel,
+                                                                                 GNUNET_NO,
+                                                                                 42,
+                                                                                 GNUNET_TIME_relative_divide
+                                                                                 (GNUNET_CONSTANTS_MAX_CORK_DELAY, 2),
+                                                                                 (const struct GNUNET_PeerIdentity *)
+                                                                                 NULL, element->len,
+                                                                                 element->cb, element->cls);
+      /* save the handle */
+      GNUNET_MESH_tunnel_set_data(*tunnel, th);
+    }
   return ntohs (hdr->size);
 }
 
@@ -231,6 +264,7 @@ static size_t
 mesh_send (void *cls, size_t size, void *buf)
 {
   struct tunnel_cls *cls_ = (struct tunnel_cls *) cls;
+  GNUNET_MESH_tunnel_set_data(cls_->tunnel, NULL);
 
   GNUNET_assert(cls_->hdr.size <= size);
 
@@ -238,23 +272,74 @@ mesh_send (void *cls, size_t size, void *buf)
   cls_->hdr.size = htons(cls_->hdr.size);
 
   memcpy(buf, &cls_->hdr, size);
+
+  if (NULL != GNUNET_MESH_tunnel_get_head(cls_->tunnel))
+    {
+      struct tunnel_notify_queue* element = GNUNET_MESH_tunnel_get_head(cls_->tunnel);
+      struct tunnel_notify_queue* head = GNUNET_MESH_tunnel_get_head(cls_->tunnel);
+      struct tunnel_notify_queue* tail = GNUNET_MESH_tunnel_get_tail(cls_->tunnel);
+
+      GNUNET_CONTAINER_DLL_remove(head, tail, element);
+
+      GNUNET_MESH_tunnel_set_head(cls_->tunnel, head);
+      GNUNET_MESH_tunnel_set_tail(cls_->tunnel, tail);
+      struct GNUNET_MESH_TransmitHandle* th = GNUNET_MESH_notify_transmit_ready (cls_->tunnel,
+                                                                                 GNUNET_NO,
+                                                                                 42,
+                                                                                 GNUNET_TIME_relative_divide
+                                                                                 (GNUNET_CONSTANTS_MAX_CORK_DELAY, 2),
+                                                                                 (const struct GNUNET_PeerIdentity *)
+                                                                                 NULL, element->len,
+                                                                                 element->cb, element->cls);
+      /* save the handle */
+      GNUNET_MESH_tunnel_set_data(cls_->tunnel, th);
+    }
+
+  GNUNET_free(cls);
   return size;
 }
 
 
-void mesh_connect (void* cls, const struct GNUNET_PeerIdentity* peer, const struct GNUNET_TRANSPORT_ATS_Information *atsi __attribute__((unused))) {
-  if (NULL == peer) return;
-  struct tunnel_cls *cls_ = (struct tunnel_cls*)cls;
-  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "Connected to peer %s, sending query with id %d\n", GNUNET_i2s(peer), ntohs(cls_->dns.s.id));
-
-  GNUNET_MESH_notify_transmit_ready(cls_->tunnel,
-                                    GNUNET_YES,
-                                    42,
-                                    GNUNET_TIME_UNIT_MINUTES,
-                                    NULL,
-                                    cls_->hdr.size,
-                                    mesh_send,
-                                    cls);
+void
+mesh_connect (void *cls, const struct GNUNET_PeerIdentity *peer,
+              const struct GNUNET_TRANSPORT_ATS_Information *atsi
+              __attribute__ ((unused)))
+{
+  if (NULL == peer)
+    return;
+  struct tunnel_cls *cls_ = (struct tunnel_cls *) cls;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Connected to peer %s, sending query with id %d\n",
+              GNUNET_i2s (peer), ntohs (cls_->dns.s.id));
+
+  if (NULL == GNUNET_MESH_tunnel_get_data (cls_->tunnel))
+    {
+      struct GNUNET_MESH_TransmitHandle *th =
+        GNUNET_MESH_notify_transmit_ready (cls_->tunnel,
+                                           GNUNET_YES,
+                                           42,
+                                           GNUNET_TIME_UNIT_MINUTES,
+                                           NULL,
+                                           cls_->hdr.size,
+                                           mesh_send,
+                                           cls);
+
+      GNUNET_MESH_tunnel_set_data (cls_->tunnel, th);
+    }
+  else
+    {
+      struct tunnel_notify_queue* head = GNUNET_MESH_tunnel_get_head(cls_->tunnel);
+      struct tunnel_notify_queue* tail = GNUNET_MESH_tunnel_get_tail(cls_->tunnel);
+
+      struct tunnel_notify_queue* element = GNUNET_malloc(sizeof(struct tunnel_notify_queue));
+      element->cls = cls;
+      element->len = cls_->hdr.size;
+      element->cb = mesh_send;
+
+      GNUNET_CONTAINER_DLL_insert_tail(head, tail, element);
+      GNUNET_MESH_tunnel_set_head(cls_->tunnel, head);
+      GNUNET_MESH_tunnel_set_tail(cls_->tunnel, tail);
+    }
 }
 
 
@@ -814,7 +899,9 @@ open_port ()
  * Read a response-packet of the UDP-Socket
  */
 static void
-read_response (void *cls __attribute__((unused)), const struct GNUNET_SCHEDULER_TaskContext *tc)
+read_response (void *cls
+               __attribute__ ((unused)),
+               const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   struct sockaddr_in addr;
   socklen_t addrlen = sizeof (addr);
@@ -859,14 +946,42 @@ read_response (void *cls __attribute__((unused)), const struct GNUNET_SCHEDULER_
       {
         if (query_states[dns->s.id].tunnel != NULL)
           {
-            uint32_t *c = GNUNET_malloc(4 + r);
+            /* This response should go through a tunnel */
+            uint32_t *c = GNUNET_malloc (4 + sizeof(struct GNUNET_MESH_Tunnel*) + r);
             *c = r;
-            memcpy(c+1, dns, r);
-            GNUNET_MESH_notify_transmit_ready (query_states[dns->s.id].tunnel,
-                                               GNUNET_YES,
-                                               32,
-                                               GNUNET_TIME_UNIT_MINUTES,
-                                               NULL, r + sizeof(struct GNUNET_MessageHeader), mesh_send_response, c);
+            struct GNUNET_MESH_Tunnel** t = (struct GNUNET_MESH_Tunnel**)(c + 1);
+            *t = query_states[dns->s.id].tunnel;
+            memcpy (t + 1, dns, r);
+            if (NULL ==
+                GNUNET_MESH_tunnel_get_data (query_states[dns->s.id].tunnel))
+              {
+                struct GNUNET_MESH_TransmitHandle *th =
+                  GNUNET_MESH_notify_transmit_ready (query_states[dns->s.id].tunnel,
+                                                     GNUNET_YES,
+                                                     32,
+                                                     GNUNET_TIME_UNIT_MINUTES,
+                                                     NULL,
+                                                     r +
+                                                     sizeof (struct
+                                                             GNUNET_MessageHeader),
+                                                     mesh_send_response, c);
+                GNUNET_MESH_tunnel_set_data (query_states[dns->s.id].tunnel,
+                                             th);
+              }
+            else
+              {
+                struct tunnel_notify_queue* head = GNUNET_MESH_tunnel_get_head(query_states[dns->s.id].tunnel);
+                struct tunnel_notify_queue* tail = GNUNET_MESH_tunnel_get_tail(query_states[dns->s.id].tunnel);
+
+                struct tunnel_notify_queue* element = GNUNET_malloc(sizeof(struct tunnel_notify_queue));
+                element->cls = c;
+                element->len = r+sizeof(struct GNUNET_MessageHeader);
+                element->cb = mesh_send_response;
+
+                GNUNET_CONTAINER_DLL_insert_tail(head, tail, element);
+                GNUNET_MESH_tunnel_set_head(query_states[dns->s.id].tunnel, head);
+                GNUNET_MESH_tunnel_set_tail(query_states[dns->s.id].tunnel, tail);
+              }
           }
         else
           {
@@ -886,12 +1001,12 @@ read_response (void *cls __attribute__((unused)), const struct GNUNET_SCHEDULER_
 
             GNUNET_CONTAINER_DLL_insert_after (head, tail, tail, answer);
 
-            GNUNET_SERVER_notify_transmit_ready (query_states[dns->s.id].
-                                                 client, len,
+            GNUNET_SERVER_notify_transmit_ready (query_states
+                                                 [dns->s.id].client, len,
                                                  GNUNET_TIME_UNIT_FOREVER_REL,
                                                  &send_answer,
-                                                 query_states[dns->s.id].
-                                                 client);
+                                                 query_states[dns->s.
+                                                              id].client);
           }
       }
   }