[gnunet] 109/164: Added multi strata estimator

[gnunet]

This is an automated email from the git hooks/post-receive script.

grothoff pushed a commit to branch master
in repository gnunet.

commit 4765c460b952c8212cf96db25a8b177664eb5a81
Author: Elias Summermatter <elias.summermatter@seccom.ch>
AuthorDate: Tue May 25 09:40:00 2021 +0200

    Added multi strata estimator
---
 src/setu/gnunet-service-setu.c                  | 51 ++++++++---------
 src/setu/gnunet-service-setu_strata_estimator.c | 75 ++++++++++++++++---------
 src/setu/gnunet-service-setu_strata_estimator.h | 33 +++++++----
 src/setu/ibf.c                                  |  6 --
 src/setu/perf_setu_api.c                        |  8 +--
 5 files changed, 97 insertions(+), 76 deletions(-)

diff --git a/src/setu/gnunet-service-setu.c b/src/setu/gnunet-service-setu.c
index 284beee46..d56fcd1f2 100644
--- a/src/setu/gnunet-service-setu.c
+++ b/src/setu/gnunet-service-setu.c
@@ -302,7 +302,7 @@ struct Operation
    * Copy of the set's strata estimator at the time of
    * creation of this operation.
    */
-  struct StrataEstimator *se;
+  struct MultiStrataEstimator *se;
 
   /**
    * The IBF we currently receive.
@@ -574,7 +574,7 @@ struct Set
    * The strata estimator is only generated once for each set.  The IBF keys
    * are derived from the element hashes with salt=0.
    */
-  struct StrataEstimator *se;
+  struct MultiStrataEstimator *se;
 
   /**
    * Evaluate operations are held in a linked list.
@@ -803,10 +803,10 @@ struct message_control_flow_element
 static void
 load_config(struct Operation * op) {
 
-    /**
     setu_cfg = GNUNET_CONFIGURATION_create();
     GNUNET_CONFIGURATION_load(setu_cfg,"perf_setu.conf");
 
+    /**
     long long number;
     float fl;
     GNUNET_CONFIGURATION_get_value_float(setu_cfg,"IBF", 
"BUCKET_NUMBER_FACTOR", &fl);
@@ -818,12 +818,11 @@ load_config(struct Operation * op) {
     GNUNET_CONFIGURATION_get_value_number(setu_cfg,"PERFORMANCE", "TRADEOFF", 
&number);
     op->rtt_bandwidth_tradeoff = number;
 
-    GNUNET_CONFIGURATION_get_value_float(setu_cfg,"PERFORMANCE", 
"MAX_SET_DIFF_FACTOR_DIFFERENTIAL", &fl);
-    op->max_set_diff_factor_diff_sync = fl;
 
     GNUNET_CONFIGURATION_get_value_number(setu_cfg,"BOUNDARIES", 
"UPPER_ELEMENT", &number);
     op->upper_element_boundary = number;
-    **/
+     **/
+
     op->peer_site = 0;
     op->active_passive_switches = 0;
 }
@@ -1940,7 +1939,7 @@ handle_union_p2p_strata_estimator (void *cls,
   perf_rtt.se.received += 1;
   perf_rtt.se.received_var_bytes += ntohs (msg->header.size) - sizeof(struct 
StrataEstimatorMessage);
   struct Operation *op = cls;
-  struct StrataEstimator *remote_se;
+  struct MultiStrataEstimator *remote_se;
   unsigned int diff;
   uint64_t other_size;
   size_t len;
@@ -2003,18 +2002,21 @@ handle_union_p2p_strata_estimator (void *cls,
     return;
   }
   GNUNET_assert (NULL != op->se);
-  diff = strata_estimator_difference (remote_se,
+  strata_estimator_difference (remote_se,
                                       op->se);
 
+  /* Calculate remote local diff */
+  int diff_remote = remote_se->stratas[0]->strata[0]->remote_decoded_count;
+  int diff_local = remote_se->stratas[0]->strata[0]->local_decoded_count;
+  diff = diff_remote + diff_local;
 
 
-  /* Calculate remote local diff */
-  int remote_elements_decoded = remote_se->strata[0]->remote_decoded_count;
-  int local_elements_decoded = remote_se->strata[0]->local_decoded_count;
 
-  uint64_t total_decoded = remote_elements_decoded + local_elements_decoded;
-  uint64_t diff_remote = ((float)remote_elements_decoded / 
(float)total_decoded) * diff;
-  uint64_t diff_local = ((float)local_elements_decoded / (float)total_decoded) 
* diff;
+    LOG (GNUNET_ERROR_TYPE_ERROR,
+         "ESTIMATION IS total: %d remote:%d local: %d\n",
+         diff,
+         diff_remote,
+         diff_local);
 
 
   /** Calculate avg element size **/
@@ -2532,7 +2534,6 @@ handle_union_p2p_ibf (void *cls,
 {
   struct Operation *op = cls;
   unsigned int buckets_in_message;
-
     /**
    * Check that the message is received only in supported phase
    */
@@ -2666,16 +2667,15 @@ maybe_finish (struct Operation *op)
 
   num_demanded = GNUNET_CONTAINER_multihashmap_size (
     op->demanded_hashes);
-  unsigned int send_done  =  GNUNET_CONTAINER_multihashmap_iterate 
(op->message_control_flow,
+  int send_done  =  GNUNET_CONTAINER_multihashmap_iterate 
(op->message_control_flow,
                                                           
&determinate_done_message_iterator,
                                                           op);
-
   if (PHASE_FINISH_WAITING == op->phase)
   {
     LOG (GNUNET_ERROR_TYPE_DEBUG,
          "In PHASE_FINISH_WAITING, pending %u demands -> %d\n",
          num_demanded, op->peer_site);
-    if (GNUNET_OK == send_done)
+    if (-1 != send_done)
     {
       struct GNUNET_MQ_Envelope *ev;
 
@@ -2693,7 +2693,7 @@ maybe_finish (struct Operation *op)
     LOG (GNUNET_ERROR_TYPE_DEBUG,
          "In PHASE_FINISH_CLOSING, pending %u demands %d\n",
          num_demanded, op->peer_site);
-    if (GNUNET_OK == send_done)
+    if (-1 != send_done)
     {
       op->phase = PHASE_FINISHED;
       send_client_done (op);
@@ -3279,7 +3279,7 @@ handle_union_p2p_demand (void *cls,
     /**
     * Check that the message is received only in supported phase
     */
-    uint8_t allowed_phases[] = {PHASE_ACTIVE_DECODING, PHASE_PASSIVE_DECODING};
+    uint8_t allowed_phases[] = {PHASE_ACTIVE_DECODING, PHASE_PASSIVE_DECODING, 
PHASE_FINISH_WAITING};
     if( GNUNET_OK !=
         check_valid_phase(allowed_phases,sizeof(allowed_phases),op)){
         GNUNET_break (0);
@@ -3370,6 +3370,7 @@ handle_union_p2p_demand (void *cls,
                            GNUNET_SET_STATUS_ADD_REMOTE);
   }
   GNUNET_CADET_receive_done (op->channel);
+  maybe_finish(op);
 }
 
 
@@ -3548,7 +3549,7 @@ handle_union_p2p_done (void *cls,
   /**
   * Check that the message is received only in supported phase
   */
-  uint8_t allowed_phases[] = {PHASE_ACTIVE_DECODING, PHASE_FINISH_WAITING};
+  uint8_t allowed_phases[] = {PHASE_ACTIVE_DECODING, PHASE_PASSIVE_DECODING};
   if( GNUNET_OK !=
     check_valid_phase(allowed_phases,sizeof(allowed_phases),op)){
     GNUNET_break (0);
@@ -3890,7 +3891,7 @@ handle_client_create_set (void *cls,
   }
   set = GNUNET_new (struct Set);
   {
-    struct StrataEstimator *se;
+    struct MultiStrataEstimator *se;
 
     se = strata_estimator_create (SE_STRATA_COUNT,
                                   SE_IBF_SIZE,
@@ -4600,7 +4601,7 @@ handle_client_accept (void *cls,
                             1,
                             GNUNET_NO);
   {
-    const struct StrataEstimator *se;
+    const struct MultiStrataEstimator *se;
     struct GNUNET_MQ_Envelope *ev;
     struct StrataEstimatorMessage *strata_msg;
     char *buf;
@@ -4618,13 +4619,13 @@ handle_client_accept (void *cls,
 
     /* kick off the operation */
     se = op->se;
-    buf = GNUNET_malloc (se->strata_count * IBF_BUCKET_SIZE * se->ibf_size);
+    buf = GNUNET_malloc (se->stratas[0]->strata_count * IBF_BUCKET_SIZE * 
se->stratas[0]->ibf_size * 4);
     len = strata_estimator_write (se,
                                   buf);
     perf_rtt.se.sent += 1;
     perf_rtt.se.sent_var_bytes += len;
 
-    if (len < se->strata_count * IBF_BUCKET_SIZE * se->ibf_size)
+    if (len < se->stratas[0]->strata_count * IBF_BUCKET_SIZE * 
se->stratas[0]->ibf_size * 4)
       type = GNUNET_MESSAGE_TYPE_SETU_P2P_SEC;
     else
       type = GNUNET_MESSAGE_TYPE_SETU_P2P_SE;
diff --git a/src/setu/gnunet-service-setu_strata_estimator.c 
b/src/setu/gnunet-service-setu_strata_estimator.c
index 0d7e40396..55b7ad5ba 100644
--- a/src/setu/gnunet-service-setu_strata_estimator.c
+++ b/src/setu/gnunet-service-setu_strata_estimator.c
@@ -37,11 +37,12 @@
  * @return number of bytes written to @a buf
  */
 size_t
-strata_estimator_write (const struct StrataEstimator *se,
+strata_estimator_write (const struct MultiStrataEstimator *se,
                         void *buf)
 {
   char *sbuf = buf;
   size_t osize;
+  uint64_t sbuf_offset = 0;
 
   GNUNET_assert (NULL != se);
   for (unsigned int i = 0; i < se->strata_count; i++)
@@ -88,7 +89,7 @@ int
 strata_estimator_read (const void *buf,
                        size_t buf_len,
                        int is_compressed,
-                       struct StrataEstimator *se)
+                       struct MultiStrataEstimator *se)
 {
   size_t osize;
   char *dbuf;
@@ -96,7 +97,7 @@ strata_estimator_read (const void *buf,
   dbuf = NULL;
   if (GNUNET_YES == is_compressed)
   {
-    osize = se->ibf_size * IBF_BUCKET_SIZE * se->strata_count;
+    osize = se->stratas[0]->ibf_size * IBF_BUCKET_SIZE * 
se->stratas[0]->strata_count * 4;
     dbuf = GNUNET_decompress (buf,
                               buf_len,
                               osize);
@@ -109,7 +110,7 @@ strata_estimator_read (const void *buf,
     buf_len = osize;
   }
 
-  if (buf_len != se->strata_count * se->ibf_size * IBF_BUCKET_SIZE)
+  if (buf_len != se->stratas[0]->strata_count * se->stratas[0]->ibf_size * 
IBF_BUCKET_SIZE * 4)
   {
     GNUNET_break (0);  /* very odd error */
     GNUNET_free (dbuf);
@@ -133,17 +134,27 @@ strata_estimator_read (const void *buf,
  * @param key key to add
  */
 void
-strata_estimator_insert (struct StrataEstimator *se,
+strata_estimator_insert (struct MultiStrataEstimator *se,
                          struct IBF_Key key)
 {
-  uint64_t v;
-  unsigned int i;
 
-  v = key.key_val;
+
   /* count trailing '1'-bits of v */
-  for (i = 0; v & 1; v >>= 1, i++)
+  for(int strata_ctr=0; strata_ctr < 4; strata_ctr++) {
+      unsigned int i;
+      uint64_t v;
+
+      struct IBF_Key salted_key;
+      salt_key (&key,
+                strata_ctr,
+                &salted_key);
+      v = salted_key.key_val;
+      for (i = 0; v & 1; v >>= 1, i++) {
+          ibf_insert(se->stratas[strata_ctr]->strata[i], salted_key);
+      }
+  }
     /* empty */;
-  ibf_insert (se->strata[i], key);
+
 }
 
 
@@ -154,17 +165,21 @@ strata_estimator_insert (struct StrataEstimator *se,
  * @param key key to remove
  */
 void
-strata_estimator_remove (struct StrataEstimator *se,
+strata_estimator_remove (struct MultiStrataEstimator *se,
                          struct IBF_Key key)
 {
-  uint64_t v;
-  unsigned int i;
 
-  v = key.key_val;
   /* count trailing '1'-bits of v */
-  for (i = 0; v & 1; v >>= 1, i++)
-    /* empty */;
-  ibf_remove (se->strata[i], key);
+  for(int strata_ctr=0; strata_ctr < 4; strata_ctr++) {
+      uint64_t v;
+      unsigned int i;
+      v = key.key_val;
+      for (i = 0; v & 1; v >>= 1, i++) {
+          /* empty */;
+          ibf_remove(se->stratas[strata_ctr]->strata[i], key);
+      }
+  }
+
 }
 
 
@@ -176,7 +191,7 @@ strata_estimator_remove (struct StrataEstimator *se,
  * @param ibf_hashnum hashnum parameter of each ibf
  * @return a freshly allocated, empty strata estimator, NULL on error
  */
-struct StrataEstimator *
+struct MultiStrataEstimator *
 strata_estimator_create (unsigned int strata_count,
                          uint32_t ibf_size,
                          uint8_t ibf_hashnum)
@@ -214,9 +229,9 @@ strata_estimator_create (unsigned int strata_count,
  * @param se2 second strata estimator
  * @return the estimated difference
  */
-unsigned int
-strata_estimator_difference (const struct StrataEstimator *se1,
-                             const struct StrataEstimator *se2)
+void
+strata_estimator_difference (const struct MultiStrataEstimator *se1,
+                             const struct MultiStrataEstimator *se2)
 {
   unsigned int count;
 
@@ -254,10 +269,16 @@ strata_estimator_difference (const struct StrataEstimator 
*se1,
         ibf_destroy (diff);
         return count * (1 << (i + 1));
       }
-    }
-    ibf_destroy (diff);
+      break_all_counting_loops:;
+      //GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "LOCAL: 
%u\n",se1->stratas[strata_ctr]->strata[0]->local_decoded_count);
+      //GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "REMOTE: 
%u\n",se1->stratas[strata_ctr]->strata[0]->remote_decoded_count);
+      avg_local_diff += 
se1->stratas[strata_ctr]->strata[0]->local_decoded_count;
+      avg_remote_diff += 
se1->stratas[strata_ctr]->strata[0]->remote_decoded_count;
   }
-  return count;
+  //GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "TOTAL LOCAL: %u\n",avg_local_diff);
+  //GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "TOTAL REMOTE: %u\n",avg_remote_diff);
+  se1->stratas[0]->strata[0]->local_decoded_count = avg_local_diff / 4;
+  se1->stratas[0]->strata[0]->remote_decoded_count = avg_remote_diff / 4;
 }
 
 
@@ -267,8 +288,8 @@ strata_estimator_difference (const struct StrataEstimator 
*se1,
  * @param se the strata estimator to copy
  * @return the copy
  */
-struct StrataEstimator *
-strata_estimator_dup (struct StrataEstimator *se)
+struct MultiStrataEstimator *
+strata_estimator_dup (struct MultiStrataEstimator *se)
 {
   struct StrataEstimator *c;
 
@@ -289,7 +310,7 @@ strata_estimator_dup (struct StrataEstimator *se)
  * @param se strata estimator to destroy.
  */
 void
-strata_estimator_destroy (struct StrataEstimator *se)
+strata_estimator_destroy (struct MultiStrataEstimator *se)
 {
   for (unsigned int i = 0; i < se->strata_count; i++)
     ibf_destroy (se->strata[i]);
diff --git a/src/setu/gnunet-service-setu_strata_estimator.h 
b/src/setu/gnunet-service-setu_strata_estimator.h
index afdbcdbbf..8c45d01ca 100644
--- a/src/setu/gnunet-service-setu_strata_estimator.h
+++ b/src/setu/gnunet-service-setu_strata_estimator.h
@@ -61,6 +61,15 @@ struct StrataEstimator
   unsigned int ibf_size;
 };
 
+struct MultiStrataEstimator
+{
+    /**
+     * Array of strata estimators
+     */
+    struct StrataEstimator **stratas;
+
+};
+
 
 /**
  * Write the given strata estimator to the buffer.
@@ -70,7 +79,7 @@ struct StrataEstimator
  * @return number of bytes written to @a buf
  */
 size_t
-strata_estimator_write (const struct StrataEstimator *se,
+strata_estimator_write (const struct MultiStrataEstimator *se,
                         void *buf);
 
 
@@ -88,7 +97,7 @@ int
 strata_estimator_read (const void *buf,
                        size_t buf_len,
                        int is_compressed,
-                       struct StrataEstimator *se);
+                       struct MultiStrataEstimator *se);
 
 
 /**
@@ -99,7 +108,7 @@ strata_estimator_read (const void *buf,
  * @param ibf_hashnum hashnum parameter of each ibf
  * @return a freshly allocated, empty strata estimator, NULL on error
  */
-struct StrataEstimator *
+struct MultiStrataEstimator *
 strata_estimator_create (unsigned int strata_count,
                          uint32_t ibf_size,
                          uint8_t ibf_hashnum);
@@ -111,11 +120,11 @@ strata_estimator_create (unsigned int strata_count,
  *
  * @param se1 first strata estimator
  * @param se2 second strata estimator
- * @return abs(|se1| - |se2|)
+ * @return nothing
  */
-unsigned int
-strata_estimator_difference (const struct StrataEstimator *se1,
-                             const struct StrataEstimator *se2);
+void
+strata_estimator_difference (const struct MultiStrataEstimator *se1,
+                             const struct MultiStrataEstimator *se2);
 
 
 /**
@@ -125,7 +134,7 @@ strata_estimator_difference (const struct StrataEstimator 
*se1,
  * @param key key to add
  */
 void
-strata_estimator_insert (struct StrataEstimator *se,
+strata_estimator_insert (struct MultiStrataEstimator *se,
                          struct IBF_Key key);
 
 
@@ -136,7 +145,7 @@ strata_estimator_insert (struct StrataEstimator *se,
  * @param key key to remove
  */
 void
-strata_estimator_remove (struct StrataEstimator *se,
+strata_estimator_remove (struct MultiStrataEstimator *se,
                          struct IBF_Key key);
 
 
@@ -146,7 +155,7 @@ strata_estimator_remove (struct StrataEstimator *se,
  * @param se strata estimator to destroy.
  */
 void
-strata_estimator_destroy (struct StrataEstimator *se);
+strata_estimator_destroy (struct MultiStrataEstimator *se);
 
 
 /**
@@ -155,8 +164,8 @@ strata_estimator_destroy (struct StrataEstimator *se);
  * @param se the strata estimator to copy
  * @return the copy
  */
-struct StrataEstimator *
-strata_estimator_dup (struct StrataEstimator *se);
+struct MultiStrataEstimator *
+strata_estimator_dup (struct MultiStrataEstimator *se);
 
 
 #if 0                           /* keep Emacsens' auto-indent happy */
diff --git a/src/setu/ibf.c b/src/setu/ibf.c
index f4cabafa0..22145c652 100644
--- a/src/setu/ibf.c
+++ b/src/setu/ibf.c
@@ -252,12 +252,6 @@ ibf_decode (struct InvertibleBloomFilter *ibf,
          (-1 != ibf->count[i].count_val) )
       continue;
 
-    if (1 == ibf->count[i].count_val) {
-        ibf->remote_decoded_count++;
-    } else {
-        ibf->local_decoded_count++;
-    }
-
     hash.key_hash_val = IBF_KEY_HASH_VAL (ibf->key_sum[i]);
 
     /* test if the hash matches the key */
diff --git a/src/setu/perf_setu_api.c b/src/setu/perf_setu_api.c
index c6872a5be..7188dc963 100644
--- a/src/setu/perf_setu_api.c
+++ b/src/setu/perf_setu_api.c
@@ -425,10 +425,6 @@ static void run_petf_thread(int total_runs) {
 //Father code (before child processes start)
     for (int processed = 0; processed < total_runs;) {
         for (int id = 0; id < core_count; id++) {
-            perf_thread();
-            processed += 1;
-        }
-            /**
             if(processed >= total_runs) break;
 
             if ((child_pid = fork()) == 0) {
@@ -437,7 +433,7 @@ static void run_petf_thread(int total_runs) {
             }
             processed += 1;
         }
-        while ((wpid = wait(&status)) > 0); **/
+        while ((wpid = wait(&status)) > 0);
     }
 }
 
@@ -460,7 +456,7 @@ static void execute_perf() {
             gcvt(x, 4, factor);
 
             char tradeoff[10] = "200000";
-            if((full_diff_ctr % 2) == 0) {
+            if((full_diff_ctr % 2) == 1) {
                 gcvt(0.25, 4, tradeoff);
             } else {
 

-- 
To stop receiving notification emails like this one, please contact
gnunet@gnunet.org.