rps: add debug function to api to get view of service

3 files changed, 250 insertions, 1 deletions

diff --git a/src/rps/gnunet-service-rps.c b/src/rps/gnunet-service-rps.c
index 56c3c52b6f..0499eb0ab9 100644
--- a/src/rps/gnunet-service-rps.c
+++ b/src/rps/gnunet-service-rps.c
@@ -1856,6 +1856,11 @@ struct ClientContext
   struct ReplyCls *rep_cls_tail;
 
   /**
+   * @brief How many updates this client expects to receive.
+   */
+  int64_t view_updates_left;
+
+  /**
    * The client handle to send the reply to
    */
   struct GNUNET_SERVICE_Client *client;
@@ -2940,6 +2945,107 @@ handle_client_seed (void *cls,
 }
 
 /**
+ * @brief Send view to client
+ *
+ * @param cli_ctx the context of the client
+ * @param view_array the peerids of the view as array (can be empty)
+ * @param view_size the size of the view array (can be 0)
+ */
+void
+send_view (const struct ClientContext *cli_ctx,
+           const struct GNUNET_PeerIdentity *view_array,
+           uint64_t view_size)
+{
+  struct GNUNET_MQ_Envelope *ev;
+  struct GNUNET_RPS_CS_DEBUG_ViewReply *out_msg;
+
+  if (NULL == view_array)
+  {
+    view_size = View_size ();
+    view_array = View_get_as_array();
+  }
+
+  ev = GNUNET_MQ_msg_extra (out_msg,
+                            view_size * sizeof (struct GNUNET_PeerIdentity),
+                            GNUNET_MESSAGE_TYPE_RPS_CS_DEBUG_VIEW_REPLY);
+  out_msg->num_peers = htonl (view_size);
+
+  GNUNET_memcpy (&out_msg[1],
+          view_array,
+          view_size * sizeof (struct GNUNET_PeerIdentity));
+  GNUNET_MQ_send (cli_ctx->mq, ev);
+}
+
+/**
+ * @brief sends updates to clients that are interested
+ */
+static void
+clients_notify_view_update (void)
+{
+  struct ClientContext *cli_ctx_iter;
+  uint64_t num_peers;
+  const struct GNUNET_PeerIdentity *view_array;
+
+  num_peers = View_size ();
+  view_array = View_get_as_array();
+  /* check size of view is small enough */
+  if (GNUNET_MAX_MESSAGE_SIZE < num_peers)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "View is too big to send\n");
+    return;
+  }
+
+  for (cli_ctx_iter = cli_ctx_head;
+       NULL != cli_ctx_iter;
+       cli_ctx_iter = cli_ctx_head->next)
+  {
+    if (1 < cli_ctx_iter->view_updates_left)
+    {
+      /* Client wants to receive limited amount of updates */
+      cli_ctx_iter->view_updates_left -= 1;
+    } else if (1 == cli_ctx_iter->view_updates_left)
+    {
+      /* Last update of view for client */
+      cli_ctx_iter->view_updates_left = -1;
+    } else if (0 > cli_ctx_iter->view_updates_left) {
+      /* Client is not interested in updates */
+      continue;
+    }
+    /* else _updates_left == 0 - infinite amount of updates */
+
+    /* send view */
+    send_view (cli_ctx_iter, view_array, num_peers);
+  }
+}
+
+
+/**
+ * Handle RPS request from the client.
+ *
+ * @param cls closure
+ * @param message the actual message
+ */
+static void
+handle_client_view_request (void *cls,
+                            const struct GNUNET_RPS_CS_DEBUG_ViewRequest *msg)
+{
+  struct ClientContext *cli_ctx = cls;
+  uint64_t num_updates;
+
+  num_updates = ntohl (msg->num_updates);
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Client requested %" PRIu64 " updates of view.\n",
+       num_updates);
+
+  GNUNET_assert (NULL != cli_ctx);
+  cli_ctx->view_updates_left = num_updates;
+  send_view (cli_ctx, NULL, 0);
+  GNUNET_SERVICE_client_continue (cli_ctx->client);
+}
+
+/**
  * Handle a CHECK_LIVE message from another peer.
  *
  * This does nothing. But without calling #GNUNET_CADET_receive_done()
@@ -3554,7 +3660,6 @@ do_mal_round (void *cls)
 }
 #endif /* ENABLE_MALICIOUS */
 
-
 /**
  * Send out PUSHes and PULLs, possibly update #view, samplers.
  *
@@ -3724,6 +3829,7 @@ do_round (void *cls)
     }
 
     GNUNET_array_grow (peers_to_clean, peers_to_clean_size, 0);
+    clients_notify_view_update();
   } else {
     LOG (GNUNET_ERROR_TYPE_DEBUG, "No update of the view.\n");
     GNUNET_STATISTICS_update(stats, "# rounds blocked", 1, GNUNET_NO);
@@ -3973,6 +4079,7 @@ client_connect_cb (void *cls,
     return client; /* Server was destroyed before a client connected. Shutting down */
   cli_ctx = GNUNET_new (struct ClientContext);
   cli_ctx->mq = GNUNET_SERVICE_client_get_mq (client);
+  cli_ctx->view_updates_left = -1;
   cli_ctx->client = client;
   GNUNET_CONTAINER_DLL_insert (cli_ctx_head,
                                cli_ctx_tail,
@@ -4216,6 +4323,10 @@ GNUNET_SERVICE_MAIN
    struct GNUNET_RPS_CS_ActMaliciousMessage,
    NULL),
 #endif /* ENABLE_MALICIOUS */
+ GNUNET_MQ_hd_fixed_size (client_view_request,
+   GNUNET_MESSAGE_TYPE_RPS_CS_DEBUG_VIEW_REQUEST,
+   struct GNUNET_RPS_CS_DEBUG_ViewRequest,
+   NULL),
  GNUNET_MQ_handler_end());
 
 /* end of gnunet-service-rps.c */
diff --git a/src/rps/rps.h b/src/rps/rps.h
index f5cc2e8d10..6a7fa3e147 100644
--- a/src/rps/rps.h
+++ b/src/rps/rps.h
@@ -176,6 +176,49 @@ struct GNUNET_RPS_CS_ActMaliciousMessage
 #endif /* ENABLE_MALICIOUS */
 
 
+/* Debug messages */
+
+/**
+ * Message from client to service indicating that
+ * clients wants to get updates of the view
+ */
+struct GNUNET_RPS_CS_DEBUG_ViewRequest
+{
+  /**
+   * Header including size and type in NBO
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Number of updates
+   * 0 for sending updates until cancellation
+   */
+  uint32_t num_updates GNUNET_PACKED;
+};
+
+/**
+ * Message from service to client containing current update of view
+ */
+struct GNUNET_RPS_CS_DEBUG_ViewReply
+{
+  /**
+   * Header including size and type in NBO
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Identifyer of the message.
+   */
+  uint32_t id GNUNET_PACKED;
+
+  /**
+   * Number of peers in the view
+   */
+  uint64_t num_peers GNUNET_PACKED;
+};
+  /* Followed by num_peers * GNUNET_PeerIdentity */
+
+
 /***********************************************************************
  * Defines from old gnunet-service-rps_peers.h
 ***********************************************************************/
diff --git a/src/rps/rps_api.c b/src/rps/rps_api.c
index ccd4800869..62ba9e226d 100644
--- a/src/rps/rps_api.c
+++ b/src/rps/rps_api.c
@@ -56,6 +56,16 @@ struct GNUNET_RPS_Handle
    * The id of the last request.
    */
   uint32_t current_request_id;
+
+  /**
+   * @brief Callback called on each update of the view
+   */
+  GNUNET_RPS_ViewUpdateCB view_update_cb;
+
+  /**
+   * @brief Callback called on each update of the view
+   */
+  void *view_update_cls;
 };
 
 
@@ -236,6 +246,86 @@ handle_reply (void *cls,
 }
 
 
+/* Get internals for debugging/profiling purposes */
+
+/**
+ * Request updates of view
+ *
+ * @param rps_handle handle to the rps service
+ * @param num_req_peers number of peers we want to receive
+ *        (0 for infinite updates)
+ * @param cls a closure that will be given to the callback
+ * @param ready_cb the callback called when the peers are available
+ */
+void
+GNUNET_RPS_view_request (struct GNUNET_RPS_Handle *rps_handle,
+                         uint32_t num_updates,
+                         GNUNET_RPS_ViewUpdateCB view_update_cb,
+                         void *cls)
+{
+  struct GNUNET_MQ_Envelope *ev;
+  struct GNUNET_RPS_CS_DEBUG_ViewRequest *msg;
+
+  rps_handle->view_update_cb = view_update_cb;
+  rps_handle->view_update_cls = cls;
+
+  ev = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_RPS_CS_DEBUG_VIEW_REQUEST);
+  msg->num_updates = htonl (num_updates);
+  GNUNET_MQ_send (rps_handle->mq, ev);
+}
+
+/**
+ * This function is called, when the service updates the view.
+ * It verifies that @a msg is well-formed.
+ *
+ * @param cls the closure
+ * @param msg the message
+ * @return #GNUNET_OK if @a msg is well-formed
+ */
+static int
+check_view_update (void *cls,
+                   const struct GNUNET_RPS_CS_DEBUG_ViewReply *msg)
+{
+  uint16_t msize = ntohs (msg->header.size);
+  uint32_t num_peers = ntohl (msg->num_peers);
+
+  msize -= sizeof (struct GNUNET_RPS_CS_DEBUG_ViewReply);
+  if ( (msize / sizeof (struct GNUNET_PeerIdentity) != num_peers) ||
+       (msize % sizeof (struct GNUNET_PeerIdentity) != 0) )
+  {
+    GNUNET_break (0);
+    return GNUNET_SYSERR;
+  }
+  return GNUNET_OK;
+}
+
+/**
+ * This function is called, when the service updated its view.
+ * It calls the callback the caller provided
+ * and disconnects afterwards.
+ *
+ * @param msg the message
+ */
+static void
+handle_view_update (void *cls,
+                    const struct GNUNET_RPS_CS_DEBUG_ViewReply *msg)
+{
+  struct GNUNET_RPS_Handle *h = cls;
+  struct GNUNET_PeerIdentity *peers;
+
+  /* Give the peers back */
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "New view of %" PRIu32 " peers:\n",
+       ntohl (msg->num_peers));
+
+  peers = (struct GNUNET_PeerIdentity *) &msg[1];
+  GNUNET_assert (NULL != h);
+  GNUNET_assert (NULL != h->view_update_cb);
+  h->view_update_cb (h->view_update_cls, ntohl (msg->num_peers), peers);
+}
+
+
+
 /**
  * Reconnect to the service
  */
@@ -281,6 +371,10 @@ reconnect (struct GNUNET_RPS_Handle *h)
                            GNUNET_MESSAGE_TYPE_RPS_CS_REPLY,
                            struct GNUNET_RPS_CS_ReplyMessage,
                            h),
+    GNUNET_MQ_hd_var_size (view_update,
+                           GNUNET_MESSAGE_TYPE_RPS_CS_DEBUG_VIEW_REPLY,
+                           struct GNUNET_RPS_CS_DEBUG_ViewReply,
+                           h),
     GNUNET_MQ_handler_end ()
   };
 
@@ -306,6 +400,7 @@ GNUNET_RPS_connect (const struct GNUNET_CONFIGURATION_Handle *cfg)
   struct GNUNET_RPS_Handle *h;
 
   h = GNUNET_new (struct GNUNET_RPS_Handle);
+  h->current_request_id = 0;
   h->cfg = cfg;
   reconnect (h);
   if (NULL == h->mq)