Eliminated mesh API buffering

9 files changed, 140 insertions, 56 deletions

diff --git a/src/dns/gnunet-service-dns.c b/src/dns/gnunet-service-dns.c
index 570f742d8..10fc00282 100644
--- a/src/dns/gnunet-service-dns.c
+++ b/src/dns/gnunet-service-dns.c
@@ -1647,7 +1647,7 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
       GNUNET_APPLICATION_TYPE_END
     };
     mesh = GNUNET_MESH_connect (cfg,
-                                1, NULL,
+                                NULL,
                                 &accept_dns_tunnel, 
                                 &destroy_dns_tunnel,
                                 mesh_handlers,
diff --git a/src/exit/gnunet-daemon-exit.c b/src/exit/gnunet-daemon-exit.c
index 2614ba908..337fca481 100644
--- a/src/exit/gnunet-daemon-exit.c
+++ b/src/exit/gnunet-daemon-exit.c
@@ -3202,7 +3202,7 @@ run (void *cls, char *const *args GNUNET_UNUSED,
   connections_map = GNUNET_CONTAINER_multihashmap_create (65536);
   connections_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
   mesh_handle 
-    = GNUNET_MESH_connect (cfg, 42 /* queue size */, NULL, 
+    = GNUNET_MESH_connect (cfg, NULL, 
                            &new_tunnel, 
                            &clean_tunnel, handlers,
                            apptypes);
diff --git a/src/include/gnunet_mesh_service.h b/src/include/gnunet_mesh_service.h
index c235bdd42..044fe1cee 100644
--- a/src/include/gnunet_mesh_service.h
+++ b/src/include/gnunet_mesh_service.h
@@ -154,9 +154,6 @@ typedef uint32_t GNUNET_MESH_ApplicationType;
  * Connect to the mesh service.
  *
  * @param cfg configuration to use
- * @param queue_size size of the data message queue, shared among all tunnels
- *                   (each tunnel is guaranteed to accept at least one message,
- *                    no matter what is the status of other tunnels)
  * @param cls closure for the various callbacks that follow
  *            (including handlers in the handlers array)
  * @param new_tunnel function called when an *inbound* tunnel is created
@@ -172,8 +169,7 @@ typedef uint32_t GNUNET_MESH_ApplicationType;
  *         (in this case, init is never called)
  */
 struct GNUNET_MESH_Handle *
-GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
-                     unsigned int queue_size, void *cls,
+GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg, void *cls,
                      GNUNET_MESH_InboundTunnelNotificationHandler new_tunnel,
                      GNUNET_MESH_TunnelEndHandler cleaner,
                      const struct GNUNET_MESH_MessageHandler *handlers,
diff --git a/src/include/gnunet_protocols.h b/src/include/gnunet_protocols.h
index 6d90f5265..a475ab4cc 100644
--- a/src/include/gnunet_protocols.h
+++ b/src/include/gnunet_protocols.h
@@ -848,27 +848,32 @@ extern "C"
 /**
  * Set tunnel speed to slowest peer
  */
-#define GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_MIN    282
+#define GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_MIN       282
 
 /**
  * Set tunnel speed to fastest peer
  */
-#define GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_MAX   283
+#define GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_MAX       283
 
 /**
  * Set tunnel buffering on.
  */
-#define GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_BUFFER   284
+#define GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_BUFFER    284
 
 /**
  * Set tunnel buffering off.
  */
-#define GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_NOBUFFER   285
+#define GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_NOBUFFER  285
+
+/**
+ * Local ACK for data.
+ */
+#define GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK              286
 
 /**
  * 640kb should be enough for everybody
  */
-#define GNUNET_MESSAGE_TYPE_MESH_RESERVE_END            288
+#define GNUNET_MESSAGE_TYPE_MESH_RESERVE_END            299
 
 
 
diff --git a/src/mesh/mesh.h b/src/mesh/mesh.h
index f00c5b256..66d6d6cb1 100644
--- a/src/mesh/mesh.h
+++ b/src/mesh/mesh.h
@@ -169,24 +169,24 @@ struct GNUNET_MESH_TunnelNotification
 struct GNUNET_MESH_PeerControl
 {
 
-  /**
-   * Type: GNUNET_MESSAGE_TYPE_MESH_LOCAL_CONNECT_PEER_[ADD|DEL|[UN]BLACKLIST]
-   *       (client to service, client created tunnel)
-   *       GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_[CONNECTED|DISCONNECTED]
-   *       (service to client)
-   *
-   * Size: sizeof(struct GNUNET_MESH_PeerControl)
-   */
+    /**
+     * Type: GNUNET_MESSAGE_TYPE_MESH_LOCAL_CONNECT_PEER_[ADD|DEL|[UN]BLACKLIST]
+     *       (client to service, client created tunnel)
+     *       GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_[CONNECTED|DISCONNECTED]
+     *       (service to client)
+     *
+     * Size: sizeof(struct GNUNET_MESH_PeerControl)
+     */
   struct GNUNET_MessageHeader header;
 
-  /**
-   * ID of a tunnel controlled by this client.
-   */
+    /**
+     * ID of a tunnel controlled by this client.
+     */
   MESH_TunnelNumber tunnel_id GNUNET_PACKED;
 
-  /**
-   * Peer to connect/disconnect.
-   */
+    /**
+     * Peer to connect/disconnect.
+     */
   struct GNUNET_PeerIdentity peer;
 };
 
@@ -199,17 +199,19 @@ struct GNUNET_MESH_ConnectPeerByType
     /**
      * Type: GNUNET_MESSAGE_TYPE_MESH_LOCAL_CONNECT_PEER_BY_TYPE |
      *       GNUNET_MESSAGE_TYPE_MESH_LOCAL_DISCONNECT_PEER_BY_TYPE
+     * 
+     * Size: sizeof(struct GNUNET_MESH_ConnectPeerByType)
      */
   struct GNUNET_MessageHeader header;
 
-  /**
-   * ID of a tunnel controlled by this client.
-   */
+    /**
+     * ID of a tunnel controlled by this client.
+     */
   MESH_TunnelNumber tunnel_id GNUNET_PACKED;
 
-  /**
-   * Type specification
-   */
+    /**
+     * Type specification
+     */
   GNUNET_MESH_ApplicationType type GNUNET_PACKED;
 };
 
@@ -221,16 +223,43 @@ struct GNUNET_MESH_ConnectPeerByString
 {
     /**
      * Type: GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_ADD_BY_STRING
+     * 
+     * Size: sizeof(struct GNUNET_MESH_ConnectPeerByString) + strlen (string)
      */
   struct GNUNET_MessageHeader header;
 
-  /**
-   * ID of a tunnel controlled by this client.
-   */
+    /**
+     * ID of a tunnel controlled by this client.
+     */
   MESH_TunnelNumber tunnel_id GNUNET_PACKED;
 
   /* String describing the service */
 };
+
+
+/**
+ * Message to allow the client send more data to the service
+ * (always service -> client).
+ */
+struct GNUNET_MESH_LocalAck
+{
+    /**
+     * Type: GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK
+     */
+  struct GNUNET_MessageHeader header;
+
+    /**
+     * ID of the tunnel allowed to send more data.
+     */
+  MESH_TunnelNumber tunnel_id GNUNET_PACKED;
+
+    /**
+     * ID of the last packet allowed.
+     */
+  uint32_t max_pid GNUNET_PACKED;
+};
+
+
 GNUNET_NETWORK_STRUCT_END
 
 /******************************************************************************/
diff --git a/src/mesh/mesh_api.c b/src/mesh/mesh_api.c
index 935c99574..f02c17a2a 100644
--- a/src/mesh/mesh_api.c
+++ b/src/mesh/mesh_api.c
@@ -43,6 +43,15 @@
 
 
 /******************************************************************************/
+/************************        CONSTANTS         ****************************/
+/******************************************************************************/
+
+#define HIGH_PID 0xFFFF0000
+#define LOW_PID 0x0000FFFF
+
+#define PID_OVERFLOW(pid, max) (pid > HIGH_PID && max < LOW_PID)
+
+/******************************************************************************/
 /************************      DATA STRUCTURES     ****************************/
 /******************************************************************************/
 
@@ -234,9 +243,13 @@ struct GNUNET_MESH_Tunnel
 {
 
     /**
-     * DLL
+     * DLL next
      */
   struct GNUNET_MESH_Tunnel *next;
+
+    /**
+     * DLL prev
+     */
   struct GNUNET_MESH_Tunnel *prev;
 
     /**
@@ -303,12 +316,23 @@ struct GNUNET_MESH_Tunnel
      * Is the tunnel throttled to the slowest peer?
      */
   int speed_min;
-  
+
     /**
      * Is the tunnel allowed to buffer?
      */
   int buffering;
 
+    /**
+     * Next packet PID.
+     */
+  uint32_t pid;
+
+    /**
+     * Maximum allowed PID.
+     */
+  uint32_t max_pid;
+
+
 };
 
 
@@ -1006,15 +1030,14 @@ process_incoming_data (struct GNUNET_MESH_Handle *h,
       if (GNUNET_OK !=
           handler->callback (h->cls, t, &t->ctx, peer, payload, &atsi))
       {
-        LOG (GNUNET_ERROR_TYPE_DEBUG, "MESH: callback caused disconnection\n");
+        LOG (GNUNET_ERROR_TYPE_DEBUG, "callback caused disconnection\n");
         GNUNET_MESH_disconnect (h);
         return GNUNET_NO;
       }
       else
       {
         LOG (GNUNET_ERROR_TYPE_DEBUG,
-             "MESH: callback completed successfully\n");
-
+             "callback completed successfully\n");
       }
     }
   }
@@ -1023,6 +1046,39 @@ process_incoming_data (struct GNUNET_MESH_Handle *h,
 
 
 /**
+ * Process a local ACK message, enabling the client to send
+ * more data to the service.
+ * 
+ * @param h Mesh handle.
+ * @param message Message itself.
+ */
+static void
+process_ack (struct GNUNET_MESH_Handle *h,
+             const struct GNUNET_MessageHeader *message)
+{
+  struct GNUNET_MESH_LocalAck *msg;
+  struct GNUNET_MESH_Tunnel *t;
+  uint32_t ack;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Got an ACK!\n");
+  msg = (struct GNUNET_MESH_LocalAck *) message;
+
+  t = retrieve_tunnel (h, ntohl (msg->tunnel_id));
+
+  if (NULL == t)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+         "ACK on unknown tunnel %X\n",
+         ntohl (msg->tunnel_id));
+    return;
+  }
+  ack = ntohl (msg->max_pid);
+  if (ack > t->max_pid || PID_OVERFLOW (t->max_pid, ack))
+    t->max_pid = ack;
+}
+
+
+/**
  * Function to process all messages received from the service
  *
  * @param cls closure
@@ -1063,10 +1119,13 @@ msg_received (void *cls, const struct GNUNET_MessageHeader *msg)
     if (GNUNET_NO == process_incoming_data (h, msg))
       return;
     break;
-    /* We shouldn't get any other packages, log and ignore */
+  case GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK:
+    process_ack (h, msg);
+    break;
   default:
+    /* We shouldn't get any other packages, log and ignore */
     LOG (GNUNET_ERROR_TYPE_WARNING,
-         "MESH: unsolicited message form service (type %d)\n",
+         "unsolicited message form service (type %d)\n",
          ntohs (msg->type));
   }
   LOG (GNUNET_ERROR_TYPE_DEBUG, "message processed\n");
@@ -1266,9 +1325,6 @@ send_packet (struct GNUNET_MESH_Handle *h,
  * Connect to the mesh service.
  *
  * @param cfg configuration to use
- * @param queue_size size of the data message queue, shared among all tunnels
- *                   (each tunnel is guaranteed to accept at least one message,
- *                    no matter what is the status of other tunnels)
  * @param cls closure for the various callbacks that follow
  *            (including handlers in the handlers array)
  * @param new_tunnel function called when an *inbound* tunnel is created
@@ -1284,8 +1340,7 @@ send_packet (struct GNUNET_MESH_Handle *h,
  *         (in this case, init is never called)
  */
 struct GNUNET_MESH_Handle *
-GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
-                     unsigned int queue_size, void *cls,
+GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg, void *cls,
                      GNUNET_MESH_InboundTunnelNotificationHandler new_tunnel,
                      GNUNET_MESH_TunnelEndHandler cleaner,
                      const struct GNUNET_MESH_MessageHandler *handlers,
@@ -1296,7 +1351,6 @@ GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
   LOG (GNUNET_ERROR_TYPE_DEBUG, "GNUNET_MESH_connect()\n");
   h = GNUNET_malloc (sizeof (struct GNUNET_MESH_Handle));
   h->cfg = cfg;
-  h->max_queue_size = queue_size;
   h->new_tunnel = new_tunnel;
   h->cleaner = cleaner;
   h->client = GNUNET_CLIENT_connect ("mesh", cfg);
diff --git a/src/pt/gnunet-daemon-pt.c b/src/pt/gnunet-daemon-pt.c
index 73936a489..702b8a95c 100644
--- a/src/pt/gnunet-daemon-pt.c
+++ b/src/pt/gnunet-daemon-pt.c
@@ -932,7 +932,7 @@ run (void *cls, char *const *args GNUNET_UNUSED,
       GNUNET_SCHEDULER_shutdown ();
       return;
     }
-    mesh_handle = GNUNET_MESH_connect (cfg, 1, NULL, NULL, NULL,
+    mesh_handle = GNUNET_MESH_connect (cfg, NULL, NULL, NULL,
                                        mesh_handlers, mesh_types);
     if (NULL == mesh_handle)
     {
diff --git a/src/stream/stream_api.c b/src/stream/stream_api.c
index 595bf7a6c..930bd63e7 100644
--- a/src/stream/stream_api.c
+++ b/src/stream/stream_api.c
@@ -2813,10 +2813,12 @@ new_tunnel_notify (void *cls,
 
   if (GNUNET_NO == lsocket->listening)
   {
-    LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "%s: Destroying tunnel from peer %s as we don't have the lock\n",
-         GNUNET_i2s (&socket->other_peer),
-         GNUNET_i2s (&socket->other_peer));
+//     FIXME: socket uninitalized
+//     FIXME: cannot use GNUNET_i2s twice in same call (static buffer)
+//     LOG (GNUNET_ERROR_TYPE_DEBUG,
+//          "%s: Destroying tunnel from peer %s as we don't have the lock\n",
+//          GNUNET_i2s (&socket->other_peer),
+//          GNUNET_i2s (&socket->other_peer));
     GNUNET_MESH_tunnel_destroy (tunnel);
     return NULL;
   }
@@ -2949,7 +2951,6 @@ lock_status_change_cb (void *cls, const char *domain, uint32_t lock,
       GNUNET_MESH_ApplicationType ports[] = {lsocket->port, 0};
 
       lsocket->mesh = GNUNET_MESH_connect (lsocket->cfg,
-                                           RECEIVE_BUFFER_SIZE, /* FIXME: QUEUE size as parameter? */
                                            lsocket, /* Closure */
                                            &new_tunnel_notify,
                                            &tunnel_cleaner,
@@ -3036,7 +3037,6 @@ GNUNET_STREAM_open (const struct GNUNET_CONFIGURATION_Handle *cfg,
   } while (GNUNET_STREAM_OPTION_END != option);
   va_end (vargs);               /* End of variable args parsing */
   socket->mesh = GNUNET_MESH_connect (cfg, /* the configuration handle */
-                                      RECEIVE_BUFFER_SIZE,  /* QUEUE size as parameter? */
                                       socket, /* cls */
                                       NULL, /* No inbound tunnel handler */
                                       NULL, /* No in-tunnel cleaner */
diff --git a/src/vpn/gnunet-service-vpn.c b/src/vpn/gnunet-service-vpn.c
index 0b9a4dcb2..677b272ea 100644
--- a/src/vpn/gnunet-service-vpn.c
+++ b/src/vpn/gnunet-service-vpn.c
@@ -3177,7 +3177,7 @@ run (void *cls,
   vpn_argv[6] = NULL;
 
   mesh_handle =
-    GNUNET_MESH_connect (cfg_, 42 /* queue length */, NULL, 
+    GNUNET_MESH_connect (cfg_, NULL, 
                          &inbound_tunnel_cb, 
                          &tunnel_cleaner, 
                          mesh_handlers,