fair, global message buffer implemented
[oweals/gnunet.git] / src / cadet / gnunet-service-cadet-new_channel.c
index ef1858c4b4fdb7eda4423007713c8e548e7787ef..00866b3d540e39a4c44322461ed36f34360b8c86 100644 (file)
@@ -128,6 +128,11 @@ struct CadetReliableMessage
    */
   struct CadetTunnelQueueEntry *qe;
 
+  /**
+   * Data message we are trying to send.
+   */
+  struct GNUNET_CADET_ChannelAppDataMessage *data_message;
+
   /**
    * How soon should we retry if we fail to get an ACK?
    * Messages in the queue are sorted by this value.
@@ -141,9 +146,23 @@ struct CadetReliableMessage
   struct GNUNET_TIME_Relative retry_delay;
 
   /**
-   * Data message we are trying to send.
+   * Time when we first successfully transmitted the message
+   * (that is, set @e num_transmissions to 1).
    */
-  struct GNUNET_CADET_ChannelAppDataMessage *data_message;
+  struct GNUNET_TIME_Absolute first_transmission_time;
+
+  /**
+   * Identifier of the connection that this message took when it
+   * was first transmitted.  Only useful if @e num_transmissions is 1.
+   */
+  struct GNUNET_CADET_ConnectionTunnelIdentifier connection_taken;
+
+  /**
+   * How often was this message transmitted?  #GNUNET_SYSERR if there
+   * was an error transmitting the message, #GNUNET_NO if it was not
+   * yet transmitted ever, otherwise the number of (re) transmissions.
+   */
+  int num_transmissions;
 
 };
 
@@ -501,9 +520,12 @@ send_channel_open (void *cls);
  * create message.  Delays for a bit until we retry.
  *
  * @param cls our `struct CadetChannel`.
+ * @param cid identifier of the connection within the tunnel, NULL
+ *            if transmission failed
  */
 static void
-channel_open_sent_cb (void *cls)
+channel_open_sent_cb (void *cls,
+                      const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
 {
   struct CadetChannel *ch = cls;
 
@@ -755,9 +777,12 @@ GCCH_channel_incoming_new (struct CadetTunnel *t,
  * ACKs for ACKs ;-).
  *
  * @param cls our `struct CadetChannel`.
+ * @param cid identifier of the connection within the tunnel, NULL
+ *            if transmission failed
  */
 static void
-send_ack_cb (void *cls)
+send_ack_cb (void *cls,
+             const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
 {
   struct CadetChannel *ch = cls;
 
@@ -832,9 +857,11 @@ send_open_ack (void *cls)
  * #GNUNET_MESSAGE_TYPE_CADET_CHANNEL_OPEN_ACK.
  *
  * @param ch channel that got the duplicate open
+ * @param cti identifier of the connection that delivered the message
  */
 void
-GCCH_handle_duplicate_open (struct CadetChannel *ch)
+GCCH_handle_duplicate_open (struct CadetChannel *ch,
+                            const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
 {
   if (NULL == ch->dest)
   {
@@ -949,7 +976,8 @@ GCCH_bind (struct CadetChannel *ch,
   if (GNUNET_YES == ch->is_loopback)
   {
     ch->state = CADET_CHANNEL_OPEN_SENT;
-    GCCH_handle_channel_open_ack (ch);
+    GCCH_handle_channel_open_ack (ch,
+                                  NULL);
   }
   else
   {
@@ -1034,9 +1062,11 @@ GCCH_channel_local_destroy (struct CadetChannel *ch,
  * (the port is open on the other side). Begin transmissions.
  *
  * @param ch channel to destroy
+ * @param cti identifier of the connection that delivered the message
  */
 void
-GCCH_handle_channel_open_ack (struct CadetChannel *ch)
+GCCH_handle_channel_open_ack (struct CadetChannel *ch,
+                              const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
 {
   switch (ch->state)
   {
@@ -1084,8 +1114,8 @@ GCCH_handle_channel_open_ack (struct CadetChannel *ch)
  * Test if element @a e1 comes before element @a e2.
  *
  * @param cls closure, to a flag where we indicate duplicate packets
- * @param e1 an element of to sort
- * @param e2 another element to sort
+ * @param m1 a message of to sort
+ * @param m2 another message to sort
  * @return #GNUNET_YES if @e1 < @e2, otherwise #GNUNET_NO
  */
 static int
@@ -1108,7 +1138,7 @@ is_before (void *cls,
   }
   else
   {
-    /* result is small, thus v2 > v1, thus e1 < e2 */
+    /* result is small, thus v2 > v1, thus m1 < m2 */
     return GNUNET_YES;
   }
 }
@@ -1119,10 +1149,12 @@ is_before (void *cls,
  * and send an ACK to the other end (once flow control allows it!)
  *
  * @param ch channel that got data
+ * @param cti identifier of the connection that delivered the message
  * @param msg message that was received
  */
 void
 GCCH_handle_channel_plaintext_data (struct CadetChannel *ch,
+                                    const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
                                     const struct GNUNET_CADET_ChannelAppDataMessage *msg)
 {
   struct GNUNET_MQ_Envelope *env;
@@ -1242,6 +1274,13 @@ GCCH_handle_channel_plaintext_data (struct CadetChannel *ch,
 
       /* Yep, need to drop. Drop the oldest message in
          the buffer. */
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "Queue full due slow client on %s, dropping oldest message\n",
+           GCCH_2s (ch));
+      GNUNET_STATISTICS_update (stats,
+                                "# messages dropped due to slow client",
+                                1,
+                                GNUNET_NO);
       drop = ccc->head_recv;
       GNUNET_CONTAINER_DLL_remove (ccc->head_recv,
                                    ccc->tail_recv,
@@ -1315,9 +1354,12 @@ GCCH_handle_channel_plaintext_data (struct CadetChannel *ch,
  * wait for ACK (or retransmit).
  *
  * @param cls the `struct CadetReliableMessage` that was sent
+ * @param cid identifier of the connection within the tunnel, NULL
+ *            if transmission failed
  */
 static void
-data_sent_cb (void *cls);
+data_sent_cb (void *cls,
+              const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid);
 
 
 /**
@@ -1351,10 +1393,12 @@ retry_transmission (void *cls)
  * the queue and tell our client that it can send more.
  *
  * @param ch the channel that got the PLAINTEXT_DATA_ACK
+ * @param cti identifier of the connection that delivered the message
  * @param crm the message that got acknowledged
  */
 static void
 handle_matching_ack (struct CadetChannel *ch,
+                     const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
                      struct CadetReliableMessage *crm)
 {
   GNUNET_CONTAINER_DLL_remove (ch->head_sent,
@@ -1372,6 +1416,18 @@ handle_matching_ack (struct CadetChannel *ch,
     GCT_send_cancel (crm->qe);
     crm->qe = NULL;
   }
+  if ( (1 == crm->num_transmissions) &&
+       (NULL != cti) )
+  {
+    GCC_ack_observed (cti);
+    if (0 == memcmp (cti,
+                     &crm->connection_taken,
+                     sizeof (struct GNUNET_CADET_ConnectionTunnelIdentifier)))
+    {
+      GCC_latency_observed (cti,
+                            GNUNET_TIME_absolute_get_duration (crm->first_transmission_time));
+    }
+  }
   GNUNET_free (crm->data_message);
   GNUNET_free (crm);
   send_ack_to_client (ch,
@@ -1386,10 +1442,12 @@ handle_matching_ack (struct CadetChannel *ch,
  * Possibly resume transmissions.
  *
  * @param ch channel that got the ack
+ * @param cti identifier of the connection that delivered the message
  * @param ack details about what was received
  */
 void
 GCCH_handle_channel_plaintext_data_ack (struct CadetChannel *ch,
+                                        const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti,
                                         const struct GNUNET_CADET_ChannelDataAckMessage *ack)
 {
   struct CadetReliableMessage *crm;
@@ -1427,6 +1485,7 @@ GCCH_handle_channel_plaintext_data_ack (struct CadetChannel *ch,
            ntohl (crm->data_message->mid.mid),
            GCCH_2s (ch));
       handle_matching_ack (ch,
+                           cti,
                            crm);
       found = GNUNET_YES;
       continue;
@@ -1446,6 +1505,7 @@ GCCH_handle_channel_plaintext_data_ack (struct CadetChannel *ch,
            ntohl (crm->data_message->mid.mid),
            GCCH_2s (ch));
       handle_matching_ack (ch,
+                           cti,
                            crm);
       found = GNUNET_YES;
     }
@@ -1483,9 +1543,12 @@ GCCH_handle_channel_plaintext_data_ack (struct CadetChannel *ch,
  * the tunnel.
  *
  * @param ch channel to destroy
+ * @param cti identifier of the connection that delivered the message,
+ *            NULL if we are simulating receiving a destroy due to shutdown
  */
 void
-GCCH_handle_remote_destroy (struct CadetChannel *ch)
+GCCH_handle_remote_destroy (struct CadetChannel *ch,
+                            const struct GNUNET_CADET_ConnectionTunnelIdentifier *cti)
 {
   struct CadetChannelClient *ccc;
 
@@ -1542,9 +1605,12 @@ cmp_crm_by_next_retry (void *cls,
  * wait for ACK (or retransmit).
  *
  * @param cls the `struct CadetReliableMessage` that was sent
+ * @param cid identifier of the connection within the tunnel, NULL
+ *            if transmission failed
  */
 static void
-data_sent_cb (void *cls)
+data_sent_cb (void *cls,
+              const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
 {
   struct CadetReliableMessage *crm = cls;
   struct CadetChannel *ch = crm->ch;
@@ -1566,6 +1632,23 @@ data_sent_cb (void *cls)
                         : GNUNET_YES);
     return;
   }
+  if (NULL == cid)
+  {
+    /* There was an error sending. */
+    crm->num_transmissions = GNUNET_SYSERR;
+  }
+  else if (GNUNET_SYSERR != crm->num_transmissions)
+  {
+    /* Increment transmission counter, and possibly store @a cid
+       if this was the first transmission. */
+    crm->num_transmissions++;
+    if (1 == crm->num_transmissions)
+    {
+      crm->first_transmission_time = GNUNET_TIME_absolute_get ();
+      crm->connection_taken = *cid;
+      GCC_ack_expected (cid);
+    }
+  }
   if (0 == crm->retry_delay.rel_value_us)
     crm->retry_delay = ch->expected_delay;
   else
@@ -1775,7 +1858,8 @@ GCCH_handle_local_ack (struct CadetChannel *ch,
   }
 
   if ( (com->mid.mid != ch->mid_recv.mid) &&
-       (GNUNET_NO == ch->out_of_order) )
+       (GNUNET_NO == ch->out_of_order) &&
+       (GNUNET_YES == ch->reliable) )
   {
     LOG (GNUNET_ERROR_TYPE_DEBUG,
          "Got LOCAL_ACK, %s-%X ready to receive more data (but next one is out-of-order %u vs. %u)!\n",