- simplify multicast client flow control, base all on ACKs
[oweals/gnunet.git] / src / mesh / gnunet-service-mesh.c
index 8a3a0e4b423248f4936a8d293743c6f0196ed8f8..a294bbd398a7443380a2971192a36b8544e25079 100644 (file)
@@ -115,12 +115,8 @@ struct MeshData
   /** Tunnel it belongs to. */
   struct MeshTunnel *t;
 
-  /** In case of a multicast, task to allow a client to send more data if
-   * some neighbor is too slow. */
-  GNUNET_SCHEDULER_TaskIdentifier *task;
-
   /** How many remaining neighbors we need to send this to. */
-  unsigned int *reference_counter;
+  unsigned int reference_counter;
 
   /** Size of the data. */
   size_t data_len;
@@ -1635,6 +1631,7 @@ announce_application (void *cls, const struct GNUNET_HashCode * key, void *value
 
   block.id = my_full_id;
   c =  GNUNET_CONTAINER_multihashmap_get (applications, key);
+  GNUNET_assert(NULL != c);
   block.type = (long) GNUNET_CONTAINER_multihashmap_get (c->apps, key);
   if (0 == block.type)
   {
@@ -1773,23 +1770,9 @@ announce_id (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 static void
 data_descriptor_decrement_rc (struct MeshData *mesh_data)
 {
-  /* Make sure it's a multicast packet */
-  GNUNET_assert (NULL != mesh_data->reference_counter);
-
-  if (0 == --(*(mesh_data->reference_counter)))
+  if (0 == --(mesh_data->reference_counter))
   {
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Last copy!\n");
-    if (NULL != mesh_data->task)
-    {
-      if (GNUNET_SCHEDULER_NO_TASK != *(mesh_data->task))
-      {
-        GNUNET_SCHEDULER_cancel (*(mesh_data->task));
-        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " notifying client...\n");
-        GNUNET_SERVER_receive_done (mesh_data->t->owner->handle, GNUNET_OK);
-      }
-      GNUNET_free (mesh_data->task);
-    }
-    GNUNET_free (mesh_data->reference_counter);
     GNUNET_free (mesh_data->data);
     GNUNET_free (mesh_data);
   }
@@ -1845,32 +1828,6 @@ client_is_subscribed (uint16_t message_type, struct MeshClient *c)
 }
 
 
-/**
- * Allow a client to send more data after transmitting a multicast message
- * which some neighbor has not yet accepted altough a reasonable time has
- * passed.
- *
- * @param cls Closure (DataDescriptor containing the task identifier)
- * @param tc Task Context
- * 
- * FIXME reference counter cshould be just int
- */
-static void
-client_allow_send (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
-  struct MeshData *mdata = cls;
-
-  if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
-    return;
-  GNUNET_assert (NULL != mdata->reference_counter);
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "CLIENT ALLOW SEND DESPITE %u COPIES PENDING\n",
-              *(mdata->reference_counter));
-  *(mdata->task) = GNUNET_SCHEDULER_NO_TASK;
-  GNUNET_SERVER_receive_done (mdata->t->owner->handle, GNUNET_OK);
-}
-
-
 /**
  * Check whether client wants traffic from a tunnel.
  *
@@ -2303,8 +2260,7 @@ send_message (const struct GNUNET_MessageHeader *message,
     m->ttl = htonl (ntohl (m->ttl) - 1);
   }
   info->mesh_data->data_len = size;
-  info->mesh_data->reference_counter = GNUNET_malloc (sizeof (unsigned int));
-  *info->mesh_data->reference_counter = 1;
+  info->mesh_data->reference_counter = 1;
   neighbor = peer_info_get (peer);
   for (p = neighbor->path_head; NULL != p; p = p->next)
   {
@@ -3303,7 +3259,7 @@ tunnel_send_multicast_iterator (void *cls, GNUNET_PEER_Id neighbor_id)
   info = GNUNET_malloc (sizeof (struct MeshTransmissionDescriptor));
 
   info->mesh_data = mdata;
-  (*(mdata->reference_counter)) ++;
+  (mdata->reference_counter) ++;
   info->destination = neighbor_id;
   GNUNET_PEER_resolve (neighbor_id, &neighbor);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   sending to %s...\n",
@@ -3324,7 +3280,9 @@ tunnel_send_multicast_iterator (void *cls, GNUNET_PEER_Id neighbor_id)
  *
  * @param t Tunnel in which to send the data.
  * @param msg Message to be sent.
- * @param internal Has the service generated this message?
+ * @param internal DEPRECATED Has the service generated this message?
+ * 
+ * FIXME remove internal if no use comes up
  */
 static void
 tunnel_send_multicast (struct MeshTunnel *t,
@@ -3338,7 +3296,6 @@ tunnel_send_multicast (struct MeshTunnel *t,
 
   mdata = GNUNET_malloc (sizeof (struct MeshData));
   mdata->data_len = ntohs (msg->size);
-  mdata->reference_counter = GNUNET_malloc (sizeof (unsigned int));
   mdata->t = t;
   mdata->data = GNUNET_malloc (mdata->data_len);
   memcpy (mdata->data, msg, mdata->data_len);
@@ -3358,7 +3315,6 @@ tunnel_send_multicast (struct MeshTunnel *t,
                   "  message at %s!\n",
                   GNUNET_i2s(&my_full_id));
       GNUNET_free (mdata->data);
-      GNUNET_free (mdata->reference_counter);
       GNUNET_free (mdata);
       return;
     }
@@ -3371,31 +3327,13 @@ tunnel_send_multicast (struct MeshTunnel *t,
   {
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  not a data packet, no ttl\n");
   }
-  if (NULL != t->owner &&
-      GNUNET_YES != t->owner->shutting_down &&
-      GNUNET_NO == internal)
-  {
-    mdata->task = GNUNET_malloc (sizeof (GNUNET_SCHEDULER_TaskIdentifier));
-    (*(mdata->task)) =
-        GNUNET_SCHEDULER_add_delayed (unacknowledged_wait_time, &client_allow_send,
-                                      mdata);
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "timeout task %u\n",
-                *(mdata->task));
-  }
 
   tree_iterate_children (t->tree, &tunnel_send_multicast_iterator, mdata);
-  if (*(mdata->reference_counter) == 0)
+  if (mdata->reference_counter == 0)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "  no one to send data to\n");
     GNUNET_free (mdata->data);
-    GNUNET_free (mdata->reference_counter);
-    if (NULL != mdata->task)
-    {
-      GNUNET_SCHEDULER_cancel(*(mdata->task));
-      GNUNET_free (mdata->task);
-      GNUNET_SERVER_receive_done (t->owner->handle, GNUNET_OK);
-    }
     GNUNET_free (mdata);
     t->fwd_queue_n--;
   }
@@ -3550,7 +3488,7 @@ tunnel_get_child_fwd_ack (void *cls,
  *
  * @param t Tunnel.
  *
- * @return Maximum PID allowed (uint32 MAX), -1 if node has no children.
+ * @return Maximum PID allowed (uint32 MAX), -1LL if node has no children.
  */
 static int64_t
 tunnel_get_children_fwd_ack (struct MeshTunnel *t)
@@ -3607,7 +3545,7 @@ tunnel_set_client_fwd_ack (struct MeshTunnel *t,
  * @param t Tunnel on which to look.
  * 
  * @return Corresponding ACK value (max uint32_t).
- *         If no clients are suscribed, -1.
+ *         If no clients are suscribed, -1LL.
  */
 static int64_t
 tunnel_get_clients_fwd_ack (struct MeshTunnel *t)
@@ -3622,9 +3560,9 @@ tunnel_get_clients_fwd_ack (struct MeshTunnel *t)
     return -1LL;
   }
 
-  for (ack = -1, i = 0; i < t->nclients; i++)
+  for (ack = -1LL, i = 0; i < t->nclients; i++)
   {
-    if (-1 == ack ||
+    if (-1LL == ack ||
         (GNUNET_YES == t->speed_min &&
          GNUNET_YES == GMC_is_pid_bigger (ack, t->clients_fc[i].fwd_ack)) ||
         (GNUNET_NO == t->speed_min &&
@@ -3652,24 +3590,27 @@ tunnel_get_clients_fwd_ack (struct MeshTunnel *t)
 static uint32_t
 tunnel_get_fwd_ack (struct MeshTunnel *t)
 {
+  uint32_t ack;
   uint32_t count;
   uint32_t buffer_free;
   int64_t child_ack;
   int64_t client_ack;
-  uint32_t ack;
 
   count = t->fwd_pid - t->skip;
   buffer_free = t->fwd_queue_max - t->fwd_queue_n;
   ack = count + buffer_free; // Might overflow 32 bits, it's ok!
   child_ack = tunnel_get_children_fwd_ack (t);
   client_ack = tunnel_get_clients_fwd_ack (t);
-  if (-1 == child_ack)
+  if (-1LL == child_ack)
   {
     // Node has no children, child_ack AND core buffer are irrelevant.
-    GNUNET_break (-1 != client_ack); // No children AND no clients? Not good!
+    GNUNET_break (-1LL != client_ack); // No children AND no clients? Not good!
     return (uint32_t) client_ack;
   }
-
+  if (-1LL == client_ack)
+  {
+    client_ack = ack;
+  }
   if (GNUNET_YES == t->speed_min)
   {
     ack = GMC_min_pid ((uint32_t) child_ack, ack);
@@ -3682,7 +3623,8 @@ tunnel_get_fwd_ack (struct MeshTunnel *t)
   }
   if (GNUNET_YES == t->nobuffer && GMC_is_pid_bigger(ack, t->fwd_pid))
     ack = t->fwd_pid + 1; // Might overflow 32 bits, it's ok!
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "c %u, bf %u, ch %u, cl %u, ACK: %u\n",
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "c %u, bf %u, ch %lld, cl %lld, ACK: %u\n",
               count, buffer_free, child_ack, client_ack, ack);
   return ack;
 }
@@ -4385,8 +4327,9 @@ send_core_path_create (void *cls, size_t size, void *buf)
   msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE);
   msg->tid = ntohl (t->id.tid);
 
+  opt = 0;
   if (GNUNET_YES == t->speed_min)
-    opt = MESH_TUNNEL_OPT_SPEED_MIN;
+    opt |= MESH_TUNNEL_OPT_SPEED_MIN;
   if (GNUNET_YES == t->nobuffer)
     opt |= MESH_TUNNEL_OPT_NOBUFFER;
   msg->opt = htonl(opt);
@@ -4729,7 +4672,7 @@ queue_send (void *cls, size_t size, void *buf)
         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   multicast\n");
         {
           struct MeshTransmissionDescriptor *info = queue->cls;
-          if (*(info->mesh_data->reference_counter) == 1)
+          if (info->mesh_data->reference_counter == 1)
             t->fwd_queue_n--;
           // FIXME fc (t->fwd_queue_n--)
         }
@@ -4786,7 +4729,8 @@ queue_send (void *cls, size_t size, void *buf)
     }
 
     /* If more data in queue, send next */
-    if (NULL != peer->queue_head)
+    queue = queue_get_next(peer);
+    if (NULL != queue)
     {
         struct GNUNET_PeerIdentity id;
 
@@ -4798,10 +4742,17 @@ queue_send (void *cls, size_t size, void *buf)
                                               0,
                                               GNUNET_TIME_UNIT_FOREVER_REL,
                                               &id,
-                                              peer->queue_head->size,
+                                              queue->size,
                                               &queue_send,
                                               peer);
     }
+    else
+    {
+      if (NULL != peer->queue_head)
+        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                    "*********   %s stalled\n",
+                    GNUNET_i2s(&my_full_id));
+    }
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   return %d\n", data_size);
     return data_size;
 }
@@ -4977,6 +4928,9 @@ handle_mesh_path_create (void *cls, const struct GNUNET_PeerIdentity *peer,
                    GNUNET_YES : GNUNET_NO;
     t->nobuffer = (0 != (opt & MESH_TUNNEL_OPT_NOBUFFER)) ?
                   GNUNET_YES : GNUNET_NO;
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "  speed_min: %d, nobuffer:%d\n",
+                t->speed_min, t->nobuffer);
 
     if (GNUNET_YES == t->nobuffer)
     {
@@ -7393,7 +7347,7 @@ handle_local_multicast (void *cls, struct GNUNET_SERVER_Client *client,
     handle_mesh_data_multicast (client, &my_full_id, &copy->header, NULL, 0);
   }
 
-  /* receive done gets called when last copy is sent to a neighbor */
+  GNUNET_SERVER_receive_done (t->owner->handle, GNUNET_OK);
   return;
 }