[GNUnet-SVN] r35353 - in gnunet/src: fragmentation include transport

[gnunet]

Author: grothoff
Date: 2015-03-07 20:35:26 +0100 (Sat, 07 Mar 2015)
New Revision: 35353

Modified:
   gnunet/src/fragmentation/defragmentation.c
   gnunet/src/fragmentation/fragmentation.c
   gnunet/src/include/gnunet_fragmentation_lib.h
   gnunet/src/transport/plugin_transport_udp.c
Log:
trying to fix #3694: reduce duplicate ACKing

Modified: gnunet/src/fragmentation/defragmentation.c
===================================================================
--- gnunet/src/fragmentation/defragmentation.c  2015-03-07 19:10:51 UTC (rev 
35352)
+++ gnunet/src/fragmentation/defragmentation.c  2015-03-07 19:35:26 UTC (rev 
35353)
@@ -149,7 +149,7 @@
   struct MessageContext *tail;
 
   /**
-   * Closure for 'proc' and 'ackp'.
+   * Closure for @e proc and @e ackp.
    */
   void *cls;
 
@@ -195,7 +195,7 @@
  * @param mtu the maximum message size for each fragment
  * @param num_msgs how many fragmented messages
  *                 to we defragment at most at the same time?
- * @param cls closure for proc and ackp
+ * @param cls closure for @a proc and @a ackp
  * @param proc function to call with defragmented messages
  * @param ackp function to call with acknowledgements (to send
  *             back to the other side)
@@ -255,7 +255,8 @@
  * @param tc the scheduler context
  */
 static void
-send_ack (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+send_ack (void *cls,
+          const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   struct MessageContext *mc = cls;
   struct GNUNET_DEFRAGMENT_Context *dc = mc->dc;
@@ -267,9 +268,12 @@
   fa.fragment_id = htonl (mc->fragment_id);
   fa.bits = GNUNET_htonll (mc->bits);
   GNUNET_STATISTICS_update (mc->dc->stats,
-                            _("# acknowledgements sent for fragment"), 1,
+                            _("# acknowledgements sent for fragment"),
+                            1,
                             GNUNET_NO);
-  dc->ackp (dc->cls, mc->fragment_id, &fa.header);
+  dc->ackp (dc->cls,
+            mc->fragment_id,
+            &fa.header);
 }
 
 
@@ -402,7 +406,9 @@
  *
  * @param dc the context
  * @param msg the message that was received
- * @return GNUNET_OK on success, GNUNET_NO if this was a duplicate, 
GNUNET_SYSERR on error
+ * @return #GNUNET_OK on success,
+ *         #GNUNET_NO if this was a duplicate,
+ *         #GNUNET_SYSERR on error
  */
 int
 GNUNET_DEFRAGMENT_process_fragment (struct GNUNET_DEFRAGMENT_Context *dc,
@@ -453,7 +459,10 @@
     GNUNET_break_op (0);
     return GNUNET_SYSERR;
   }
-  GNUNET_STATISTICS_update (dc->stats, _("# fragments received"), 1, 
GNUNET_NO);
+  GNUNET_STATISTICS_update (dc->stats,
+                            _("# fragments received"),
+                            1,
+                            GNUNET_NO);
   num_fragments = (ntohs (msg->size) + dc->mtu - sizeof (struct 
FragmentHeader)-1) / (dc->mtu - sizeof (struct FragmentHeader));
   last = 0;
   for (mc = dc->head; NULL != mc; mc = mc->next)
@@ -495,7 +504,9 @@
       mc->bits = (1LL << n) - 1;        /* set lowest 'bits' bit */
     if (dc->list_size >= dc->num_msgs)
       discard_oldest_mc (dc);
-    GNUNET_CONTAINER_DLL_insert (dc->head, dc->tail, mc);
+    GNUNET_CONTAINER_DLL_insert (dc->head,
+                                 dc->tail,
+                                 mc);
     dc->list_size++;
   }
 
@@ -518,7 +529,9 @@
   else
   {
     duplicate = GNUNET_YES;
-    GNUNET_STATISTICS_update (dc->stats, _("# duplicate fragments received"), 
1,
+    GNUNET_STATISTICS_update (dc->stats,
+                              _("# duplicate fragments received"),
+                              1,
                               GNUNET_NO);
   }
 
@@ -529,9 +542,12 @@
       bc++;
 
   /* notify about complete message */
-  if ((duplicate == GNUNET_NO) && (0 == mc->bits))
+  if ( (duplicate == GNUNET_NO) &&
+       (0 == mc->bits) )
   {
-    GNUNET_STATISTICS_update (dc->stats, _("# messages defragmented"), 1,
+    GNUNET_STATISTICS_update (dc->stats,
+                              _("# messages defragmented"),
+                              1,
                               GNUNET_NO);
     /* message complete, notify! */
     dc->proc (dc->cls, mc->msg);
@@ -541,10 +557,11 @@
   {
     dc->latency = estimate_latency (mc);
   }
-  delay = GNUNET_TIME_relative_multiply (dc->latency, bc + 1);
+  delay = GNUNET_TIME_relative_multiply (dc->latency,
+                                         bc + 1);
   if ( (last + fid == num_fragments) ||
-       (0 == mc->bits) ||
-       (GNUNET_YES == duplicate))
+       ( (0 == mc->bits) &&
+         (GNUNET_YES != duplicate)) )
   {
     /* message complete or duplicate or last missing fragment in
        linear sequence; ACK now! */
@@ -552,7 +569,9 @@
   }
   if (NULL != mc->ack_task)
     GNUNET_SCHEDULER_cancel (mc->ack_task);
-  mc->ack_task = GNUNET_SCHEDULER_add_delayed (delay, &send_ack, mc);
+  mc->ack_task = GNUNET_SCHEDULER_add_delayed (delay,
+                                               &send_ack,
+                                               mc);
   if (duplicate == GNUNET_YES)
     return GNUNET_NO;
   return GNUNET_YES;

Modified: gnunet/src/fragmentation/fragmentation.c
===================================================================
--- gnunet/src/fragmentation/fragmentation.c    2015-03-07 19:10:51 UTC (rev 
35352)
+++ gnunet/src/fragmentation/fragmentation.c    2015-03-07 19:35:26 UTC (rev 
35353)
@@ -80,7 +80,7 @@
   GNUNET_FRAGMENT_MessageProcessor proc;
 
   /**
-   * Closure for 'proc'.
+   * Closure for @e proc.
    */
   void *proc_cls;
 
@@ -98,7 +98,7 @@
   /**
    * Task performing work for the fragmenter.
    */
-  struct GNUNET_SCHEDULER_Task * task;
+  struct GNUNET_SCHEDULER_Task *task;
 
   /**
    * Our fragmentation ID. (chosen at random)
@@ -121,12 +121,12 @@
   unsigned int num_transmissions;
 
   /**
-   * GNUNET_YES if we called 'proc' and are now waiting for 
'GNUNET_FRAGMENT_transmission_done'
+   * #GNUNET_YES if we called @e proc and are now waiting for 
#GNUNET_FRAGMENT_transmission_done()
    */
   int8_t proc_busy;
 
   /**
-   * GNUNET_YES if we are waiting for an ACK.
+   * #GNUNET_YES if we are waiting for an ACK.
    */
   int8_t wack;
 
@@ -139,13 +139,39 @@
 
 
 /**
+ * Convert an ACK message to a printable format suitable for logging.
+ *
+ * @param ack message to print
+ * @return ack in human-readable format
+ */
+const char *
+GNUNET_FRAGMENT_print_ack (const struct GNUNET_MessageHeader *ack)
+{
+  static char buf[128];
+  const struct FragmentAcknowledgement *fa;
+
+  if (sizeof (struct FragmentAcknowledgement) !=
+      htons (ack->size))
+    return "<malformed ack>";
+  fa = (const struct FragmentAcknowledgement *) ack;
+  GNUNET_snprintf (buf,
+                   sizeof (buf),
+                   "%u-%llX",
+                   ntohl (fa->fragment_id),
+                   GNUNET_ntohll (fa->bits));
+  return buf;
+}
+
+
+/**
  * Transmit the next fragment to the other peer.
  *
- * @param cls the 'struct GNUNET_FRAGMENT_Context'
+ * @param cls the `struct GNUNET_FRAGMENT_Context`
  * @param tc scheduler context
  */
 static void
-transmit_next (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+transmit_next (void *cls,
+               const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   struct GNUNET_FRAGMENT_Context *fc = cls;
   char msg[fc->mtu];
@@ -246,11 +272,11 @@
 
 /**
  * Create a fragmentation context for the given message.
- * Fragments the message into fragments of size "mtu" or
- * less.  Calls 'proc' on each un-acknowledged fragment,
- * using both the expected 'delay' between messages and
- * acknowledgements and the given 'tracker' to guide the
- * frequency of calls to 'proc'.
+ * Fragments the message into fragments of size @a mtu or
+ * less.  Calls @a proc on each un-acknowledged fragment,
+ * using both the expected @a msg_delay between messages and
+ * acknowledgements and the given @a tracker to guide the
+ * frequency of calls to @a proc.
  *
  * @param stats statistics context
  * @param mtu the maximum message size for each fragment
@@ -261,7 +287,7 @@
  *              and ACK based on previous messages
  * @param msg the message to fragment
  * @param proc function to call for each fragment to transmit
- * @param proc_cls closure for proc
+ * @param proc_cls closure for @a proc
  * @return the fragmentation context
  */
 struct GNUNET_FRAGMENT_Context *
@@ -336,10 +362,10 @@
  *
  * @param fc fragmentation context
  * @param msg acknowledgement message we received
- * @return GNUNET_OK if this ack completes the work of the 'fc'
+ * @return #GNUNET_OK if this ack completes the work of the 'fc'
  *                   (all fragments have been received);
- *         GNUNET_NO if more messages are pending
- *         GNUNET_SYSERR if this ack is not valid for this fc
+ *         #GNUNET_NO if more messages are pending
+ *         #GNUNET_SYSERR if this ack is not valid for this fc
  */
 int
 GNUNET_FRAGMENT_process_ack (struct GNUNET_FRAGMENT_Context *fc,

Modified: gnunet/src/include/gnunet_fragmentation_lib.h
===================================================================
--- gnunet/src/include/gnunet_fragmentation_lib.h       2015-03-07 19:10:51 UTC 
(rev 35352)
+++ gnunet/src/include/gnunet_fragmentation_lib.h       2015-03-07 19:35:26 UTC 
(rev 35353)
@@ -1,6 +1,6 @@
 /*
      This file is part of GNUnet
-     Copyright (C) 2009, 2011 Christian Grothoff (and other contributing 
authors)
+     Copyright (C) 2009, 2011, 2015 Christian Grothoff (and other contributing 
authors)
 
      GNUnet is free software; you can redistribute it and/or modify
      it under the terms of the GNU General Public License as published
@@ -51,24 +51,24 @@
 /**
  * Function that is called with messages created by the fragmentation
  * module.  In the case of the 'proc' callback of the
- * GNUNET_FRAGMENT_context_create function, this function must
- * eventually call 'GNUNET_FRAGMENT_context_transmission_done'.
+ * #GNUNET_FRAGMENT_context_create() function, this function must
+ * eventually call #GNUNET_FRAGMENT_context_transmission_done().
  *
  * @param cls closure
  * @param msg the message that was created
  */
-typedef void (*GNUNET_FRAGMENT_MessageProcessor) (void *cls,
-                                                  const struct
-                                                  GNUNET_MessageHeader * msg);
+typedef void
+(*GNUNET_FRAGMENT_MessageProcessor) (void *cls,
+                                     const struct GNUNET_MessageHeader *msg);
 
 
 /**
  * Create a fragmentation context for the given message.
- * Fragments the message into fragments of size "mtu" or
- * less.  Calls 'proc' on each un-acknowledged fragment,
- * using both the expected 'delay' between messages and
- * acknowledgements and the given 'tracker' to guide the
- * frequency of calls to 'proc'.
+ * Fragments the message into fragments of size @a mtu or
+ * less.  Calls @a proc on each un-acknowledged fragment,
+ * using both the expected @a msg_delay between messages and
+ * acknowledgements and the given @a tracker to guide the
+ * frequency of calls to @a proc.
  *
  * @param stats statistics context
  * @param mtu the maximum message size for each fragment
@@ -110,10 +110,10 @@
  *
  * @param fc fragmentation context
  * @param msg acknowledgement message we received
- * @return GNUNET_OK if this ack completes the work of the 'fc'
+ * @return #GNUNET_OK if this ack completes the work of the 'fc'
  *                   (all fragments have been received);
- *         GNUNET_NO if more messages are pending
- *         GNUNET_SYSERR if this ack is not valid for this fc
+ *         #GNUNET_NO if more messages are pending
+ *         #GNUNET_SYSERR if this ack is not valid for this fc
  */
 int
 GNUNET_FRAGMENT_process_ack (struct GNUNET_FRAGMENT_Context *fc,
@@ -137,6 +137,16 @@
 
 
 /**
+ * Convert an ACK message to a printable format suitable for logging.
+ *
+ * @param ack message to print
+ * @return ack in human-readable format
+ */
+const char *
+GNUNET_FRAGMENT_print_ack (const struct GNUNET_MessageHeader *ack);
+
+
+/**
  * Defragmentation context (one per connection).
  */
 struct GNUNET_DEFRAGMENT_Context;
@@ -152,9 +162,10 @@
  * @param id unique message ID (modulo collisions)
  * @param msg the message that was created
  */
-typedef void (*GNUNET_DEFRAGMENT_AckProcessor) (void *cls, uint32_t id,
-                                                const struct
-                                                GNUNET_MessageHeader * msg);
+typedef void
+(*GNUNET_DEFRAGMENT_AckProcessor) (void *cls,
+                                   uint32_t id,
+                                   const struct GNUNET_MessageHeader *msg);
 
 
 /**
@@ -164,7 +175,7 @@
  * @param mtu the maximum message size for each fragment
  * @param num_msgs how many fragmented messages
  *                 to we defragment at most at the same time?
- * @param cls closure for proc and ackp
+ * @param cls closure for @a proc and @a ackp
  * @param proc function to call with defragmented messages
  * @param ackp function to call with acknowledgements (to send
  *             back to the other side)
@@ -172,7 +183,8 @@
  */
 struct GNUNET_DEFRAGMENT_Context *
 GNUNET_DEFRAGMENT_context_create (struct GNUNET_STATISTICS_Handle *stats,
-                                  uint16_t mtu, unsigned int num_msgs,
+                                  uint16_t mtu,
+                                  unsigned int num_msgs,
                                   void *cls,
                                   GNUNET_FRAGMENT_MessageProcessor proc,
                                   GNUNET_DEFRAGMENT_AckProcessor ackp);
@@ -192,7 +204,9 @@
  *
  * @param dc the context
  * @param msg the message that was received
- * @return GNUNET_OK on success, GNUNET_NO if this was a duplicate, 
GNUNET_SYSERR on error
+ * @return #GNUNET_OK on success,
+ *         #GNUNET_NO if this was a duplicate,
+ *         #GNUNET_SYSERR on error
  */
 int
 GNUNET_DEFRAGMENT_process_fragment (struct GNUNET_DEFRAGMENT_Context *dc,

Modified: gnunet/src/transport/plugin_transport_udp.c
===================================================================
--- gnunet/src/transport/plugin_transport_udp.c 2015-03-07 19:10:51 UTC (rev 
35352)
+++ gnunet/src/transport/plugin_transport_udp.c 2015-03-07 19:35:26 UTC (rev 
35353)
@@ -1980,6 +1980,12 @@
     return;
   }
   udp_ack = (const struct UDP_ACK_Message *) msg;
+  ack = (const struct GNUNET_MessageHeader *) &udp_ack[1];
+  if (ntohs (ack->size) != ntohs (msg->size) - sizeof(struct UDP_ACK_Message))
+  {
+    GNUNET_break_op(0);
+    return;
+  }
   address = GNUNET_HELLO_address_allocate (&udp_ack->sender,
                                            PLUGIN_NAME,
                                            udp_addr,
@@ -2000,10 +2006,11 @@
   if (NULL == s->frag_ctx)
   {
     LOG (GNUNET_ERROR_TYPE_WARNING | GNUNET_ERROR_TYPE_BULK,
-         "Fragmentation context of address %s for ACK not found\n",
+         "Fragmentation context of address %s for ACK (%s) not found\n",
          udp_address_to_string (plugin,
                                 address->address,
-                                address->address_length));
+                                address->address_length),
+         GNUNET_FRAGMENT_print_ack (ack));
     GNUNET_HELLO_address_free (address);
     return;
   }
@@ -2017,12 +2024,6 @@
        GNUNET_i2s (&udp_ack->sender));
   s->flow_delay_from_other_peer = GNUNET_TIME_relative_to_absolute 
(flow_delay);
 
-  ack = (const struct GNUNET_MessageHeader *) &udp_ack[1];
-  if (ntohs (ack->size) != ntohs (msg->size) - sizeof(struct UDP_ACK_Message))
-  {
-    GNUNET_break_op(0);
-    return;
-  }
 
   if (GNUNET_OK !=
       GNUNET_FRAGMENT_process_ack (s->frag_ctx->frag,