towards nicer scheduling code

1 files changed, 230 insertions, 68 deletions

diff --git a/src/ats/gnunet-service-ats_addresses.c b/src/ats/gnunet-service-ats_addresses.c
index 68d4b03de..f8cd2ab16 100644
--- a/src/ats/gnunet-service-ats_addresses.c
+++ b/src/ats/gnunet-service-ats_addresses.c
@@ -25,6 +25,7 @@
  * @author Christian Grothoff
  */
 #include "platform.h"
+#include "gnunet_ats_service.h"
 #include "gnunet-service-ats_addresses.h"
 #include "gnunet-service-ats_performance.h"
 #include "gnunet-service-ats_scheduling.h"
@@ -46,12 +47,12 @@ struct ATS_Address
 
   struct GNUNET_ATS_Information * ats;
 
+  struct GNUNET_TIME_Relative atsp_latency;
+
   struct GNUNET_BANDWIDTH_Value32NBO atsp_utilization_in;
 
   struct GNUNET_BANDWIDTH_Value32NBO atsp_utilization_out;
 
-  struct GNUNET_TIME_Relative atsp_latency;
-
   uint32_t atsp_distance;
 
   uint32_t atsp_cost_wan;
@@ -64,9 +65,10 @@ struct ATS_Address
 
   struct GNUNET_BANDWIDTH_Value32NBO assigned_bw_out;
 
-  struct GNUNET_BANDWIDTH_Value32NBO bw_in;
-
-  struct GNUNET_BANDWIDTH_Value32NBO bw_out;
+  /**
+   * Is this the active address for this peer?
+   */
+  int active;
 
 };
 
@@ -80,75 +82,138 @@ static unsigned long long total_quota_out;
 static unsigned int active_addr_count;
 
 
-struct CompareAddressContext
+/**
+ * Update a bandwidth assignment for a peer.  This trivial method currently
+ * simply assigns the same share to all active connections.
+ *
+ * @param cls unused
+ * @param key unused
+ * @param value the 'struct ATS_Address'
+ * @return GNUNET_OK (continue to iterate)
+ */
+static int 
+update_bw_it (void *cls,
+              const GNUNET_HashCode * key,
+              void *value)
 {
-  struct ATS_Address * search;
-  struct ATS_Address * result;
-};
+  struct ATS_Address *aa = value;
+
+  if (GNUNET_YES != aa->active)
+    return GNUNET_OK;
+  aa->assigned_bw_in.value__ = htonl (total_quota_in / active_addr_count);
+  aa->assigned_bw_out.value__ = htonl (total_quota_out / active_addr_count);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "New bandwidth for peer %s is %u/%u\n",
+              GNUNET_i2s (&aa->peer),
+              ntohl (aa->assigned_bw_in.value__),
+              ntohl (aa->assigned_bw_out.value__));
+  GAS_scheduling_transmit_address_suggestion (&aa->peer, 
+                                              aa->plugin, 
+                                              aa->addr, aa->addr_len, 
+                                              aa->session_id, 
+                                              aa->ats, aa->ats_count, 
+                                              aa->assigned_bw_out, aa->assigned_bw_in);
+  GAS_reservations_set_bandwidth (&aa->peer,
+                                  aa->assigned_bw_in);
+  GAS_performance_notify_clients (&aa->peer, 
+                                  aa->plugin, 
+                                  aa->addr, aa->addr_len, 
+                                  aa->ats, aa->ats_count, 
+                                  aa->assigned_bw_out, aa->assigned_bw_in);
+  return GNUNET_OK;
+}
 
 
+/**
+ * Some (significant) input changed, recalculate bandwidth assignment
+ * for all peers.
+ */
 static void
+recalculate_assigned_bw ()
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Recalculating bandwidth for all active connections\n");
+  GNUNET_CONTAINER_multihashmap_iterate (addresses, 
+                                         &update_bw_it,
+                                         NULL);
+}
+
+
+/**
+ * Destroy the given address.
+ *
+ * @param addr address to destroy
+ * @return GNUNET_YES if bandwidth allocations should be recalcualted
+ */
+static int
 destroy_address (struct ATS_Address *addr)
 {
+  int ret;
+
+  ret = GNUNET_NO;
   GNUNET_assert (GNUNET_YES == 
                  GNUNET_CONTAINER_multihashmap_remove(addresses, 
                                                       &addr->peer.hashPubKey, 
                                                       addr));
-  if (ntohl (addr->bw_in.value__) > 0)
+  if (GNUNET_YES == addr->active)
   {
     active_addr_count--;
-    // FIXME: update address assignment for other peers...
+    ret = GNUNET_YES;
   }
   GNUNET_free_non_null (addr->ats);
   GNUNET_free (addr->plugin);
   GNUNET_free (addr);
+  return ret;
 }
 
 
+struct CompareAddressContext
+{
+  const struct ATS_Address * search;
+  struct ATS_Address * result;
+};
+
+
 static int 
 compare_address_it (void *cls,
                     const GNUNET_HashCode * key,
                     void *value)
 {
   struct CompareAddressContext * cac = cls;
-  struct ATS_Address * aa = (struct ATS_Address *) value;
-
-  /* compare sessions */
-  if (aa->session_id != cac->search->session_id)
-    return GNUNET_YES;
-
-  if (aa->addr_len != cac->search->addr_len)
-  {
-    return GNUNET_YES;
-  }
+  struct ATS_Address * aa = value;
 
-  if (0 == strcmp(aa->plugin, cac->search->plugin))
-  {
+  if ( ( (aa->addr_len != cac->search->addr_len) ||
+         (0 != strcmp(aa->plugin, cac->search->plugin)) ||
+         (0 != memcmp (aa->addr, cac->search->addr, aa->addr_len)) ) &&
+       ( (aa->session_id != cac->search->session_id) ||
+         (cac->search->session_id == 0) ))
     return GNUNET_YES;
-  }
-
-  if (0 == memcmp (aa->addr, cac->search->addr, aa->addr_len))
-  {
-    cac->result = aa;
-    return GNUNET_NO;
-  }
-  return GNUNET_YES;
+  cac->result = aa;
+  return GNUNET_NO;
 }
 
 
+/**
+ * Find an existing equivalent address record. 
+ * Compares by peer identity and network address OR by session ID
+ * (one of the two must match).
+ *
+ * @param peer peer to lookup addresses for
+ * @param addr existing address record
+ * @return existing address record, NULL for none
+ */
 struct ATS_Address *
 find_address (const struct GNUNET_PeerIdentity *peer,
-              struct ATS_Address * addr)
+              const struct ATS_Address * addr)
 {
   struct CompareAddressContext cac;
+
   cac.result = NULL;
   cac.search = addr;
-
   GNUNET_CONTAINER_multihashmap_get_multiple(addresses,
-         &peer->hashPubKey,
-         compare_address_it,
-         &cac);
-
+                                             &peer->hashPubKey,
+                                             compare_address_it,
+                                             &cac);
   return cac.result;
 }
 
@@ -163,6 +228,7 @@ GAS_addresses_update (const struct GNUNET_PeerIdentity *peer,
 {
   struct ATS_Address * aa;
   struct ATS_Address * old;
+  uint32_t i;
 
   aa = GNUNET_malloc (sizeof (struct ATS_Address) + plugin_addr_len);
   aa->ats = GNUNET_malloc(atsi_count * sizeof (struct GNUNET_ATS_Information));
@@ -185,18 +251,50 @@ GAS_addresses_update (const struct GNUNET_PeerIdentity *peer,
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "Added new address for peer `%s' %X\n",
                 GNUNET_i2s (peer), aa);
-    return;
+    old = aa;
   }
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Updated existing address for peer `%s' %X \n",
-              GNUNET_i2s (peer), old);
-  GNUNET_free_non_null (old->ats);
-  old->ats = NULL;
-  old->ats_count = 0;
-  old->ats = aa->ats;
-  old->ats_count = aa->ats_count;
-  GNUNET_free (aa->plugin);
-  GNUNET_free (aa);
+  else
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Updated existing address for peer `%s' %X \n",
+                GNUNET_i2s (peer), old);
+    GNUNET_free_non_null (old->ats);
+    old->session_id = session_id;
+    old->ats = NULL;
+    old->ats_count = 0;
+    old->ats = aa->ats;
+    old->ats_count = aa->ats_count;
+    GNUNET_free (aa->plugin);
+    GNUNET_free (aa);
+  }
+  for (i=0;i<atsi_count;i++)
+    switch (atsi[i].type)
+    {
+    case GNUNET_ATS_UTILIZATION_UP:
+      old->atsp_utilization_out.value__ = atsi[i].value;
+      break;
+    case GNUNET_ATS_UTILIZATION_DOWN:
+      old->atsp_utilization_in.value__ = atsi[i].value;
+      break;
+    case GNUNET_ATS_QUALITY_NET_DELAY:
+      old->atsp_latency.rel_value = ntohl (atsi[i].value);
+      break;
+    case GNUNET_ATS_QUALITY_NET_DISTANCE:
+      old->atsp_distance = ntohl (atsi[i].value);
+      break;
+    case GNUNET_ATS_COST_WAN:
+      old->atsp_cost_wan = ntohl (atsi[i].value);
+      break;
+    case GNUNET_ATS_COST_LAN:
+      old->atsp_cost_lan = ntohl (atsi[i].value);
+      break;
+    case GNUNET_ATS_COST_WLAN:
+      old->atsp_cost_wlan = ntohl (atsi[i].value);
+      break;
+    default:
+      GNUNET_break (0);
+      break;
+    }
 }
 
 
@@ -206,7 +304,6 @@ GAS_addresses_destroy (const struct GNUNET_PeerIdentity *peer,
                        const void *plugin_addr, size_t plugin_addr_len,
                        uint32_t session_id)
 {
-
   struct ATS_Address aa;
   struct ATS_Address *res;
 
@@ -215,7 +312,6 @@ GAS_addresses_destroy (const struct GNUNET_PeerIdentity *peer,
   aa.addr = plugin_addr;
   aa.plugin = (char*) plugin_name;
   aa.session_id = session_id;
-
   res = find_address (peer, &aa);
   if (res == NULL)
   {
@@ -225,10 +321,75 @@ GAS_addresses_destroy (const struct GNUNET_PeerIdentity *peer,
                 GNUNET_i2s (peer));
     return; 
   }
+  if ( (aa.session_id == session_id) &&
+       (session_id != 0) &&
+       (res->addr_len > 0) )
+  {
+    /* just session died */
+    res->session_id = 0;
+    if (GNUNET_YES == addr->active)
+    {
+      active_addr_count--;
+      ret = GNUNET_YES;
+      recalculate_assigned_bw ();
+    }
+    return;
+  }
+  /* destroy address entirely (either was only session or was
+     not even with a session) */
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Deleting address for peer `%s': `%s'\n",
               GNUNET_i2s (peer), plugin_name);
-  destroy_address (res);
+  if (GNUNET_YES == destroy_address (res))
+    recalculate_assigned_bw ();
+}
+
+
+/**
+ * Find a "good" address to use for a peer.  If we already have an existing
+ * address, we stick to it.  Otherwise, we pick by lowest distance and then
+ * by lowest latency.  
+ * 
+ * @param cls the 'struct ATS_Address**' where we store the result
+ * @param key unused
+ * @param value another 'struct ATS_Address*' to consider using
+ * @return GNUNET_OK (continue to iterate)
+ */
+static int 
+find_address_it (void *cls,
+                 const GNUNET_HashCode * key,
+                 void *value)
+{
+  struct ATS_Address **ap = cls;
+  struct ATS_Address * aa = (struct ATS_Address *) value;
+  struct ATS_Address * ab = *ap;
+
+  if (NULL == ab)
+  {
+    *ap = aa;
+    return GNUNET_OK;
+  }
+  if ( (ntohl (ab->assigned_bw_in.value__) == 0) &&
+       (ntohl (aa->assigned_bw_in.value__) > 0) )
+  {
+    /* stick to existing connection */
+    *ap = aa;
+    return GNUNET_OK;
+  }
+  if (ab->atsp_distance > aa->atsp_distance)
+  {
+    /* user shorter distance */
+    *ap = aa;
+    return GNUNET_OK;
+  }
+  if (ab->atsp_latency.rel_value > aa->atsp_latency.rel_value)
+  {
+    /* user lower latency */
+    *ap = aa;
+    return GNUNET_OK;
+  }
+  /* don't care */
+  return GNUNET_OK;
 }
 
 
@@ -237,7 +398,11 @@ GAS_addresses_request_address (const struct GNUNET_PeerIdentity *peer)
 {
   struct ATS_Address * aa;
 
-  aa = GNUNET_CONTAINER_multihashmap_get (addresses, &peer->hashPubKey);
+  aa = NULL;
+  GNUNET_CONTAINER_multihashmap_get_multiple (addresses, 
+                                              &peer->hashPubKey,
+                                              &find_address_it,
+                                              &aa);
   if (aa == NULL)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -245,25 +410,21 @@ GAS_addresses_request_address (const struct GNUNET_PeerIdentity *peer)
                 GNUNET_i2s (peer));
     return; 
   }
-  /* FIXME: ensure that we don't do this multiple times per peer! */
-  if (ntohl (aa->bw_in.value__) == 0)
+  if (aa->active == GNUNET_NO)
   {
+    aa->active = GNUNET_YES;
     active_addr_count++;
-    aa->bw_in.value__ = htonl (total_quota_in / active_addr_count);
-    aa->bw_out.value__ = htonl (total_quota_out / active_addr_count);
-    /* FIXME: update bw assignments for other addresses... */
+    recalculate_assigned_bw ();
+  }
+  else
+  {
+    /* just to be sure... */
+    GAS_scheduling_transmit_address_suggestion (peer, aa->plugin, 
+                                                aa->addr, aa->addr_len, 
+                                                aa->session_id, 
+                                                aa->ats, aa->ats_count, 
+                                                aa->assigned_bw_out, aa->assigned_bw_in);
   }
-  GAS_reservations_set_bandwidth (peer,
-                                  aa->bw_in);
-  GAS_scheduling_transmit_address_suggestion (peer, aa->plugin, 
-                                              aa->addr, aa->addr_len, 
-                                              aa->session_id, 
-                                              aa->ats, aa->ats_count, 
-                                              aa->bw_out, aa->bw_in);
-  GAS_performance_notify_clients (peer, aa->plugin, 
-                                  aa->addr, aa->addr_len, 
-                                  aa->ats, aa->ats_count, 
-                                  aa->bw_out, aa->bw_in);
 }
 
 
@@ -326,6 +487,7 @@ GAS_addresses_destroy_all ()
   if (addresses != NULL)
     GNUNET_CONTAINER_multihashmap_iterate(addresses, 
                                           &free_address_it, NULL);
+  GNUNET_assert (active_addr_count == 0);
 }