restructured the sampler

3 files changed, 160 insertions, 65 deletions

diff --git a/src/rps/gnunet-service-rps.c b/src/rps/gnunet-service-rps.c
index 0896e126b..96fcdacf3 100644
--- a/src/rps/gnunet-service-rps.c
+++ b/src/rps/gnunet-service-rps.c
@@ -518,6 +518,43 @@ nse_callback(void *cls, struct GNUNET_TIME_Absolute timestamp, double logestimat
   }
 }
 
+
+/**
+ * Callback called once the requested PeerIDs are ready.
+ *
+ * Sends those to the requesting client.
+ */
+void client_respond (void *cls,
+    struct GNUNET_PeerIdentity *ids, uint64_t num_peers)
+{
+  struct GNUNET_MQ_Envelope *ev;
+  struct GNUNET_RPS_CS_ReplyMessage *out_msg;
+  struct GNUNET_SERVER_Client *client;
+  struct client_ctx *cli_ctx;
+
+  client = (struct GNUNET_SERVER_Client *) cls;
+
+  ev = GNUNET_MQ_msg_extra (out_msg,
+                            num_peers * sizeof (struct GNUNET_PeerIdentity),
+                            GNUNET_MESSAGE_TYPE_RPS_CS_REPLY);
+  out_msg->num_peers = GNUNET_htonll (num_peers);
+
+  memcpy(&out_msg[1],
+      ids,
+      num_peers * sizeof (struct GNUNET_PeerIdentity));
+  GNUNET_free (ids);
+  
+  cli_ctx = GNUNET_SERVER_client_get_user_context (client, struct client_ctx);
+  if ( NULL == cli_ctx ) {
+    cli_ctx = GNUNET_new (struct client_ctx);
+    cli_ctx->mq = GNUNET_MQ_queue_for_server_client (client);
+    GNUNET_SERVER_client_set_user_context (client, cli_ctx);
+  }
+  
+  GNUNET_MQ_send (cli_ctx->mq, ev);
+}
+
+
 /**
  * Handle RPS request from the client.
  *
@@ -533,15 +570,7 @@ handle_client_request (void *cls,
   LOG(GNUNET_ERROR_TYPE_DEBUG, "Client requested (a) random peer(s).\n");
 
   struct GNUNET_RPS_CS_RequestMessage *msg;
-  //unsigned int n_arr[sampler_list->size];// =
-    //GNUNET_CRYPTO_random_permute(GNUNET_CRYPTO_QUALITY_STRONG, (unsigned int) sampler_list->size);
-  //struct GNUNET_MQ_Handle *mq;
-  struct client_ctx *cli_ctx;
-  struct GNUNET_MQ_Envelope *ev;
-  struct GNUNET_RPS_CS_ReplyMessage *out_msg;
   uint64_t num_peers;
-  const struct GNUNET_PeerIdentity *peers;
-  //uint64_t i;
 
 
   /* Estimate request rate */
@@ -558,37 +587,16 @@ handle_client_request (void *cls,
         GNUNET_TIME_absolute_get ());
     request_rate = T_relative_avg (request_deltas, req_counter);
   }
-  last_request = GNUNET_TIME_absolute_get();
+  last_request = GNUNET_TIME_absolute_get ();
   // TODO resize the size of the extended_samplers
 
 
   // TODO check message size
   msg = (struct GNUNET_RPS_CS_RequestMessage *) message;
-  cli_ctx = GNUNET_SERVER_client_get_user_context (client, struct client_ctx);
-  if ( NULL == cli_ctx ) {
-    cli_ctx = GNUNET_new(struct client_ctx);
-    cli_ctx->mq = GNUNET_MQ_queue_for_server_client (client);
-    GNUNET_SERVER_client_set_user_context (client, cli_ctx);
-  }
-  
-  // How many peers do we give back?
-  // Wait until we have enough random peers?
 
   num_peers = GNUNET_ntohll (msg->num_peers);
 
-  ev = GNUNET_MQ_msg_extra (out_msg,
-                            num_peers * sizeof (struct GNUNET_PeerIdentity),
-                            GNUNET_MESSAGE_TYPE_RPS_CS_REPLY);
-  out_msg->num_peers = msg->num_peers; // No conversion between network and network order
-
-  //&out_msg[1] = RPS_sampler_get_n_rand_peers (num_peers);
-  peers = RPS_sampler_get_n_rand_peers (num_peers);
-  memcpy(&out_msg[1],
-      peers,
-      num_peers * sizeof (struct GNUNET_PeerIdentity));
-  
-  GNUNET_MQ_send (cli_ctx->mq, ev);
-  //GNUNET_MQ_destroy(mq);
+  RPS_sampler_get_n_rand_peers (client_respond, client, num_peers);
 
   GNUNET_SERVER_receive_done (client,
                               GNUNET_OK);
@@ -859,13 +867,13 @@ do_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
     uint64_t first_border;
     uint64_t second_border;
     
-    GNUNET_array_grow(gossip_list, gossip_list_size, sampler_size);
+    GNUNET_array_grow (gossip_list, gossip_list_size, sampler_size);
 
-    first_border = round(alpha * gossip_list_size);
+    first_border = round (alpha * gossip_list_size);
     for ( i = 0 ; i < first_border ; i++ )
     { // TODO use RPS_sampler_get_n_rand_peers
       /* Update gossip list with peers received through PUSHes */
-      r_index = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_STRONG,
+      r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
                                        push_list_size);
       gossip_list[i] = push_list[r_index];
       // TODO change the in_flags accordingly
@@ -875,7 +883,7 @@ do_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
     for ( i = first_border ; i < second_border ; i++ )
     {
       /* Update gossip list with peers received through PULLs */
-      r_index = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_STRONG,
+      r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
                                        pull_list_size);
       gossip_list[i] = pull_list[r_index];
       // TODO change the in_flags accordingly
@@ -884,7 +892,7 @@ do_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
     for ( i = second_border ; i < gossip_list_size ; i++ )
     {
       /* Update gossip list with peers from history */
-      peer = RPS_sampler_get_n_rand_peers (1),
+      peer = RPS_sampler_get_n_rand_peers_ (1);
       gossip_list[i] = *peer;
       // TODO change the in_flags accordingly
     }
@@ -1023,7 +1031,7 @@ init_peer_cb (void *cls,
     if (ipc->i < gossip_list_size)
     {
       memcpy(&gossip_list[ipc->i],
-          RPS_sampler_get_n_rand_peers (1),
+          RPS_sampler_get_n_rand_peers_ (1),
           (gossip_list_size - ipc->i) * sizeof(struct GNUNET_PeerIdentity));
     }
     rps_start (ipc->server);
diff --git a/src/rps/gnunet-service-rps_sampler.c b/src/rps/gnunet-service-rps_sampler.c
index 93b6b1263..1ab00aaf5 100644
--- a/src/rps/gnunet-service-rps_sampler.c
+++ b/src/rps/gnunet-service-rps_sampler.c
@@ -133,6 +133,47 @@ struct RPS_Sampler
 };
 
 /**
+ * Closure to _get_n_rand_peers_ready_cb()
+ */
+struct RPS_GetNRandPeersReadyCls
+{
+  /**
+   * Number of peers we are waiting for.
+   */
+  uint64_t num_peers;
+
+  /**
+   * Number of peers we currently have.
+   */
+  uint64_t cur_num_peers;
+
+  /**
+   * Pointer to the array holding the ids.
+   */
+  struct GNUNET_PeerIdentity *ids;
+
+  /**
+   * Callback to be called when all ids are available.
+   */
+  RPS_sampler_n_rand_peers_ready_cb callback;
+
+  /**
+   * Closure given to the callback
+   */
+  void *cls;
+};
+
+/**
+ * Callback that is called from _get_rand_peer() when the PeerID is ready.
+ *
+ * @param cls the closure given alongside this function.
+ * @param id the PeerID that was returned
+ */
+typedef void
+(*RPS_sampler_rand_peer_ready_cb) (void *cls,
+        const struct GNUNET_PeerIdentity *id);
+
+/**
  * Global sampler variable.
  */
 struct RPS_Sampler *sampler;
@@ -160,6 +201,31 @@ static uint64_t client_get_index;
 
 
 /**
+ * Callback to _get_rand_peer() used by _get_n_rand_peers().
+ *
+ * Checks whether all n peers are available. If they are, 
+ * give those back.
+ */
+  void
+RPS_sampler_get_n_rand_peers_ready_cb (void *cls,
+    const struct GNUNET_PeerIdentity *id)
+{
+  struct RPS_GetNRandPeersReadyCls *n_peers_cls;
+
+  n_peers_cls = (struct RPS_GetNRandPeersReadyCls *) cls;
+
+  if (n_peers_cls->num_peers == n_peers_cls->cur_num_peers)
+  {
+    GNUNET_assert (NULL != n_peers_cls->callback);
+
+    n_peers_cls->callback (n_peers_cls->cls, n_peers_cls->ids, n_peers_cls->num_peers);
+    
+    GNUNET_free (n_peers_cls);
+  }
+}
+
+
+/**
  * Reinitialise a previously initialised sampler element.
  *
  * @param sampler pointer to the memory that keeps the value.
@@ -507,20 +573,22 @@ RPS_sampler_get_n_rand_peers_ (uint64_t n)
  *
  * @return a random PeerID of the PeerIDs previously put into the sampler.
  */
-  const struct GNUNET_PeerIdentity * 
-RPS_sampler_get_rand_peer ()
+  void
+RPS_sampler_get_rand_peer (RPS_sampler_rand_peer_ready_cb cb,
+    void *cls, struct GNUNET_PeerIdentity *id)
 {
-  struct GNUNET_PeerIdentity *peer;
+  do
+  {
+  *id = sampler->sampler_elements[client_get_index]->peer_id;
 
-  // use _get_rand_peer_ ?
-  peer = GNUNET_new (struct GNUNET_PeerIdentity);
-  *peer = sampler->sampler_elements[client_get_index]->peer_id;
   RPS_sampler_elem_reinit (sampler->sampler_elements[client_get_index]);
   if ( client_get_index == sampler->sampler_size )
     client_get_index = 0;
   else
     client_get_index++;
-  return peer;
+  } while (NOT_EMPTY == sampler->sampler_elements[client_get_index]->is_empty);
+
+  cb (cls, id);
 }
 
 
@@ -531,35 +599,39 @@ RPS_sampler_get_rand_peer ()
  * corrsponding peer to the client.
  * Random with or without consumption?
  *
- * @return n random PeerIDs of the PeerIDs previously put into the sampler.
+ * @param cb callback that will be called once the ids are ready.
+ * @param cls closure given to @a cb
+ * @param num_peers the number of peers requested
  */
-  const struct GNUNET_PeerIdentity *
-RPS_sampler_get_n_rand_peers (uint64_t n)
+  void
+RPS_sampler_get_n_rand_peers (RPS_sampler_n_rand_peers_ready_cb cb,
+    void *cls, uint64_t num_peers)
 {
   // use _get_rand_peers_ ?
   if ( 0 == sampler->sampler_size )
   {
     LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Sgrp: List empty - Returning NULL\n");
-    return NULL;
   }
   else
   {
     // TODO check if we have too much (distinct) sampled peers
     // If we are not ready yet maybe schedule for later
     struct GNUNET_PeerIdentity *peers;
-    const struct GNUNET_PeerIdentity *peer;
     uint64_t i;
+    struct RPS_GetNRandPeersReadyCls *cb_cls;
 
-    peers = GNUNET_malloc (n * sizeof (struct GNUNET_PeerIdentity));
+    peers = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
 
-    for ( i = 0 ; i < n ; i++ ) {
-      //peers[i] = RPS_sampler_get_rand_peer_(sampler->sampler_elements);
-      peer = RPS_sampler_get_rand_peer ();
-      memcpy (&peers[i], peer, sizeof (struct GNUNET_PeerIdentity));
-      //GNUNET_free (peer);
-    }
-    return peers;
+    cb_cls = GNUNET_new (struct RPS_GetNRandPeersReadyCls);
+    cb_cls->num_peers = num_peers;
+    cb_cls->cur_num_peers = 0;
+    cb_cls->callback = NULL;
+    cb_cls->cls = NULL;
+
+    for ( i = 0 ; i < num_peers ; i++ )
+      RPS_sampler_get_rand_peer (RPS_sampler_get_n_rand_peers_ready_cb,
+          cb_cls, &peers[i]);
   }
 }
 
diff --git a/src/rps/gnunet-service-rps_sampler.h b/src/rps/gnunet-service-rps_sampler.h
index 3772f9f0b..b6aaf1b41 100644
--- a/src/rps/gnunet-service-rps_sampler.h
+++ b/src/rps/gnunet-service-rps_sampler.h
@@ -49,6 +49,18 @@ typedef void
     const struct GNUNET_PeerIdentity *id);
 
 /**
+ * Callback that is called from _get_n_rand_peers() when the PeerIDs are ready.
+ *
+ * @param cls the closure given alongside this function.
+ * @param ids the PeerIDs that were returned
+ *        to be freed
+ */
+  typedef void
+(*RPS_sampler_n_rand_peers_ready_cb) (void *cls,
+    struct GNUNET_PeerIdentity *ids, uint64_t num_peers);
+
+
+/**
  * A sampler sampling a stream of PeerIDs.
  */
 //struct RPS_Sampler;
@@ -102,15 +114,15 @@ RPS_sampler_reinitialise_by_value (const struct GNUNET_PeerIdentity *id);
 
 
 /**
- * Get one random peer out of the sampled peers.
+ * Get n random peers out of the sampled peers.
  *
  * We might want to reinitialise this sampler after giving the
  * corrsponding peer to the client.
- *
- * @return a random PeerID of the PeerIDs previously put into the sampler.
+ * Random with or without consumption?
+ * Only used internally
  */
-  const struct GNUNET_PeerIdentity * 
-RPS_sampler_get_rand_peer ();
+  const struct GNUNET_PeerIdentity *
+RPS_sampler_get_n_rand_peers_ (uint64_t n);
 
 
 /**
@@ -120,10 +132,13 @@ RPS_sampler_get_rand_peer ();
  * corrsponding peer to the client.
  * Random with or without consumption?
  *
- * @return n random PeerIDs of the PeerIDs previously put into the sampler.
+ * @param cb callback that will be called once the ids are ready.
+ * @param cls closure given to @a cb
+ * @param num_peers the number of peers requested
  */
-  const struct GNUNET_PeerIdentity *
-RPS_sampler_get_n_rand_peers (uint64_t n);
+    void
+RPS_sampler_get_n_rand_peers (RPS_sampler_n_rand_peers_ready_cb cb,
+    void *cls, uint64_t num_peers);
 
 
 /**