diff options
Diffstat (limited to 'src/rps/gnunet-rps-profiler.c')
| -rw-r--r-- | src/rps/gnunet-rps-profiler.c | 2732 |
1 files changed, 2732 insertions, 0 deletions
diff --git a/src/rps/gnunet-rps-profiler.c b/src/rps/gnunet-rps-profiler.c new file mode 100644 index 0000000000..4a7a89b15d --- /dev/null +++ b/src/rps/gnunet-rps-profiler.c @@ -0,0 +1,2732 @@ +/* + This file is part of GNUnet. + Copyright (C) 2009, 2012 GNUnet e.V. + + GNUnet is free software: you can redistribute it and/or modify it + under the terms of the GNU Affero General Public License as published + by the Free Software Foundation, either version 3 of the License, + or (at your option) any later version. + + GNUnet is distributed in the hope that it will be useful, but + WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ +/** + * @file rps/test_rps.c + * @brief Testcase for the random peer sampling service. Starts + * a peergroup with a given number of peers, then waits to + * receive size pushes/pulls from each peer. Expects to wait + * for one message from each peer. + */ +#include "platform.h" +//#include "rps_test_lib.h" +#include "gnunet_util_lib.h" +#include "gnunet_testbed_service.h" + +#include "gnunet_rps_service.h" +#include "rps-test_util.h" +#include "gnunet-service-rps_sampler_elem.h" + +#include <inttypes.h> + + +#define BIT(n) (1 << (n)) + +/** + * How many peers do we start? + */ +static uint32_t num_peers; + +/** + * @brief numer of bits required to represent the largest peer id + */ +static unsigned bits_needed; + +/** + * How long do we run the test? + * In seconds. + */ +static uint32_t timeout_s; + +/** + * How long do we run the test? + */ +//#define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30) +static struct GNUNET_TIME_Relative timeout; + + +/** + * Portion of malicious peers + */ +static double portion = .1; + +/** + * Type of malicious peer to test + */ +static unsigned int mal_type = 0; + +/** + * Handles to all of the running peers + */ +static struct GNUNET_TESTBED_Peer **testbed_peers; + +enum STAT_TYPE +{ + STAT_TYPE_ROUNDS, /* 0 */ + STAT_TYPE_BLOCKS, /* 1 */ + STAT_TYPE_BLOCKS_MANY_PUSH, /* 2 */ + STAT_TYPE_BLOCKS_NO_PUSH, /* 3 */ + STAT_TYPE_BLOCKS_NO_PULL, /* 4 */ + STAT_TYPE_BLOCKS_MANY_PUSH_NO_PULL, /* 5 */ + STAT_TYPE_BLOCKS_NO_PUSH_NO_PULL, /* 6 */ + STAT_TYPE_ISSUED_PUSH_SEND, /* 7 */ + STAT_TYPE_ISSUED_PULL_REQ, /* 8 */ + STAT_TYPE_ISSUED_PULL_REP, /* 9 */ + STAT_TYPE_SENT_PUSH_SEND, /* 10 */ + STAT_TYPE_SENT_PULL_REQ, /* 11 */ + STAT_TYPE_SENT_PULL_REP, /* 12 */ + STAT_TYPE_RECV_PUSH_SEND, /* 13 */ + STAT_TYPE_RECV_PULL_REQ, /* 14 */ + STAT_TYPE_RECV_PULL_REP, /* 15 */ + STAT_TYPE_MAX, /* 16 */ +}; + +struct STATcls +{ + struct RPSPeer *rps_peer; + enum STAT_TYPE stat_type; +}; + + +/** + * @brief Converts string representation to the corresponding #STAT_TYPE enum. + * + * @param stat_str string representation of statistics specifier + * + * @return corresponding enum + */ +enum STAT_TYPE stat_str_2_type (const char *stat_str) +{ + if (0 == strncmp ("# rounds blocked - no pull replies", stat_str, strlen ("# rounds blocked - no pull replies"))) + { + return STAT_TYPE_BLOCKS_NO_PULL; + } + else if (0 == strncmp ("# rounds blocked - too many pushes, no pull replies", stat_str, strlen ("# rounds blocked - too many pushes, no pull replies"))) + { + return STAT_TYPE_BLOCKS_MANY_PUSH_NO_PULL; + } + else if (0 == strncmp ("# rounds blocked - too many pushes", stat_str, strlen ("# rounds blocked - too many pushes"))) + { + return STAT_TYPE_BLOCKS_MANY_PUSH; + } + else if (0 == strncmp ("# rounds blocked - no pushes, no pull replies", stat_str, strlen ("# rounds blocked - no pushes, no pull replies"))) + { + return STAT_TYPE_BLOCKS_NO_PUSH_NO_PULL; + } + else if (0 == strncmp ("# rounds blocked - no pushes", stat_str, strlen ("# rounds blocked - no pushes"))) + { + return STAT_TYPE_BLOCKS_NO_PUSH; + } + else if (0 == strncmp ("# rounds blocked", stat_str, strlen ("# rounds blocked"))) + { + return STAT_TYPE_BLOCKS; + } + else if (0 == strncmp ("# rounds", stat_str, strlen ("# rounds"))) + { + return STAT_TYPE_ROUNDS; + } + else if (0 == strncmp ("# push send issued", stat_str, strlen ("# push send issued"))) + { + return STAT_TYPE_ISSUED_PUSH_SEND; + } + else if (0 == strncmp ("# pull request send issued", stat_str, strlen ("# pull request send issued"))) + { + return STAT_TYPE_ISSUED_PULL_REQ; + } + else if (0 == strncmp ("# pull reply send issued", stat_str, strlen ("# pull reply send issued"))) + { + return STAT_TYPE_ISSUED_PULL_REP; + } + else if (0 == strncmp ("# pushes sent", stat_str, strlen ("# pushes sent"))) + { + return STAT_TYPE_SENT_PUSH_SEND; + } + else if (0 == strncmp ("# pull requests sent", stat_str, strlen ("# pull requests sent"))) + { + return STAT_TYPE_SENT_PULL_REQ; + } + else if (0 == strncmp ("# pull replys sent", stat_str, strlen ("# pull replys sent"))) + { + return STAT_TYPE_SENT_PULL_REP; + } + else if (0 == strncmp ("# push message received", stat_str, strlen ("# push message received"))) + { + return STAT_TYPE_RECV_PUSH_SEND; + } + else if (0 == strncmp ("# pull request message received", stat_str, strlen ("# pull request message received"))) + { + return STAT_TYPE_RECV_PULL_REQ; + } + else if (0 == strncmp ("# pull reply messages received", stat_str, strlen ("# pull reply messages received"))) + { + return STAT_TYPE_RECV_PULL_REP; + } + return STAT_TYPE_MAX; +} + + +/** + * @brief Converts #STAT_TYPE enum to the equivalent string representation that + * is stored with the statistics service. + * + * @param stat_type #STAT_TYPE enum + * + * @return string representation that matches statistics value + */ +char* stat_type_2_str (enum STAT_TYPE stat_type) +{ + switch (stat_type) + { + case STAT_TYPE_ROUNDS: + return "# rounds"; + case STAT_TYPE_BLOCKS: + return "# rounds blocked"; + case STAT_TYPE_BLOCKS_MANY_PUSH: + return "# rounds blocked - too many pushes"; + case STAT_TYPE_BLOCKS_NO_PUSH: + return "# rounds blocked - no pushes"; + case STAT_TYPE_BLOCKS_NO_PULL: + return "# rounds blocked - no pull replies"; + case STAT_TYPE_BLOCKS_MANY_PUSH_NO_PULL: + return "# rounds blocked - too many pushes, no pull replies"; + case STAT_TYPE_BLOCKS_NO_PUSH_NO_PULL: + return "# rounds blocked - no pushes, no pull replies"; + case STAT_TYPE_ISSUED_PUSH_SEND: + return "# push send issued"; + case STAT_TYPE_ISSUED_PULL_REQ: + return "# pull request send issued"; + case STAT_TYPE_ISSUED_PULL_REP: + return "# pull reply send issued"; + case STAT_TYPE_SENT_PUSH_SEND: + return "# pushes sent"; + case STAT_TYPE_SENT_PULL_REQ: + return "# pull requests sent"; + case STAT_TYPE_SENT_PULL_REP: + return "# pull replys sent"; + case STAT_TYPE_RECV_PUSH_SEND: + return "# push message received"; + case STAT_TYPE_RECV_PULL_REQ: + return "# pull request message received"; + case STAT_TYPE_RECV_PULL_REP: + return "# pull reply messages received"; + case STAT_TYPE_MAX: + default: + return "ERROR"; + ; + } +} + + +/** + * @brief Indicates whether peer should go off- or online + */ +enum PEER_ONLINE_DELTA { + /** + * @brief Indicates peer going online + */ + PEER_GO_ONLINE = 1, + /** + * @brief Indicates peer going offline + */ + PEER_GO_OFFLINE = -1, +}; + +/** + * Operation map entry + */ +struct OpListEntry +{ + /** + * DLL next ptr + */ + struct OpListEntry *next; + + /** + * DLL prev ptr + */ + struct OpListEntry *prev; + + /** + * The testbed operation + */ + struct GNUNET_TESTBED_Operation *op; + + /** + * Depending on whether we start or stop RPS service at the peer, set this to + * #PEER_GO_ONLINE (1) or #PEER_GO_OFFLINE (-1) + */ + enum PEER_ONLINE_DELTA delta; + + /** + * Index of the regarding peer + */ + unsigned int index; +}; + +/** + * OpList DLL head + */ +static struct OpListEntry *oplist_head; + +/** + * OpList DLL tail + */ +static struct OpListEntry *oplist_tail; + + +/** + * A pending reply: A request was sent and the reply is pending. + */ +struct PendingReply +{ + /** + * DLL next,prev ptr + */ + struct PendingReply *next; + struct PendingReply *prev; + + /** + * Handle to the request we are waiting for + */ + struct GNUNET_RPS_Request_Handle *req_handle; + + /** + * The peer that requested + */ + struct RPSPeer *rps_peer; +}; + + +/** + * A pending request: A request was not made yet but is scheduled for later. + */ +struct PendingRequest +{ + /** + * DLL next,prev ptr + */ + struct PendingRequest *next; + struct PendingRequest *prev; + + /** + * Handle to the request we are waiting for + */ + struct GNUNET_SCHEDULER_Task *request_task; + + /** + * The peer that requested + */ + struct RPSPeer *rps_peer; +}; + + +/** + * Information we track for each peer. + */ +struct RPSPeer +{ + /** + * Index of the peer. + */ + unsigned int index; + + /** + * Handle for RPS connect operation. + */ + struct GNUNET_TESTBED_Operation *op; + + /** + * Handle to RPS service. + */ + struct GNUNET_RPS_Handle *rps_handle; + + /** + * ID of the peer. + */ + struct GNUNET_PeerIdentity *peer_id; + + /** + * A request handle to check for an request + */ + //struct GNUNET_RPS_Request_Handle *req_handle; + + /** + * Peer on- or offline? + */ + int online; + + /** + * Number of Peer IDs to request during the whole test + */ + unsigned int num_ids_to_request; + + /** + * Pending requests DLL + */ + struct PendingRequest *pending_req_head; + struct PendingRequest *pending_req_tail; + + /** + * Number of pending requests + */ + unsigned int num_pending_reqs; + + /** + * Pending replies DLL + */ + struct PendingReply *pending_rep_head; + struct PendingReply *pending_rep_tail; + + /** + * Number of pending replies + */ + unsigned int num_pending_reps; + + /** + * Number of received PeerIDs + */ + unsigned int num_recv_ids; + + /** + * Pending operation on that peer + */ + const struct OpListEntry *entry_op_manage; + + /** + * Testbed operation to connect to statistics service + */ + struct GNUNET_TESTBED_Operation *stat_op; + + /** + * Handle to the statistics service + */ + struct GNUNET_STATISTICS_Handle *stats_h; + + /** + * @brief flags to indicate which statistics values have been already + * collected from the statistics service. + * Used to check whether we are able to shutdown. + */ + uint32_t stat_collected_flags; + + /** + * @brief File name of the file the stats are finally written to + */ + const char *file_name_stats; + + /** + * @brief File name of the file the stats are finally written to + */ + const char *file_name_probs; + + /** + * @brief The current view + */ + struct GNUNET_PeerIdentity *cur_view; + + /** + * @brief Number of peers in the #cur_view. + */ + uint32_t cur_view_count; + + /** + * @brief Number of occurrences in other peer's view + */ + uint32_t count_in_views; + + /** + * @brief statistics values + */ + uint64_t stats[STAT_TYPE_MAX]; +}; + +/** + * Information for all the peers. + */ +static struct RPSPeer *rps_peers; + +/** + * Peermap to get the index of a given peer ID quick. + */ +static struct GNUNET_CONTAINER_MultiPeerMap *peer_map; + +/** + * IDs of the peers. + */ +static struct GNUNET_PeerIdentity *rps_peer_ids; + +/** + * ID of the targeted peer. + */ +static struct GNUNET_PeerIdentity *target_peer; + +/** + * ID of the peer that requests for the evaluation. + */ +static struct RPSPeer *eval_peer; + +/** + * Number of online peers. + */ +static unsigned int num_peers_online; + +/** + * @brief The added sizes of the peer's views + */ +static unsigned int view_sizes; + +/** + * Return value from 'main'. + */ +static int ok; + +/** + * Identifier for the churn task that runs periodically + */ +static struct GNUNET_SCHEDULER_Task *post_test_task; + +/** + * Identifier for the churn task that runs periodically + */ +static struct GNUNET_SCHEDULER_Task *shutdown_task; + +/** + * Identifier for the churn task that runs periodically + */ +static struct GNUNET_SCHEDULER_Task *churn_task; + +/** + * Called to initialise the given RPSPeer + */ +typedef void (*InitPeer) (struct RPSPeer *rps_peer); + +/** + * @brief Called directly after connecting to the service + * + * @param rps_peer Specific peer the function is called on + * @param h the handle to the rps service + */ +typedef void (*PreTest) (struct RPSPeer *rps_peer, struct GNUNET_RPS_Handle *h); + +/** + * @brief Executes functions to test the api/service for a given peer + * + * Called from within #rps_connect_complete_cb () + * Implemented by #churn_test_cb, #profiler_cb, #mal_cb, #single_req_cb, + * #delay_req_cb, #seed_big_cb, #single_peer_seed_cb, #seed_cb, #req_cancel_cb + * + * @param rps_peer the peer the task runs on + */ +typedef void (*MainTest) (struct RPSPeer *rps_peer); + +/** + * Callback called once the requested random peers are available + */ +typedef void (*ReplyHandle) (void *cls, + uint64_t n, + const struct GNUNET_PeerIdentity *recv_peers); + +/** + * Called directly before disconnecting from the service + */ +typedef void (*PostTest) (struct RPSPeer *peer); + +/** + * Function called after disconnect to evaluate test success + */ +typedef int (*EvaluationCallback) (void); + +/** + * @brief Do we have Churn? + */ +enum OPTION_CHURN { + /** + * @brief If we have churn this is set + */ + HAVE_CHURN, + /** + * @brief If we have no churn this is set + */ + HAVE_NO_CHURN, +}; + +/** + * @brief Is it ok to quit the test before the timeout? + */ +enum OPTION_QUICK_QUIT { + /** + * @brief It is ok for the test to quit before the timeout triggers + */ + HAVE_QUICK_QUIT, + + /** + * @brief It is NOT ok for the test to quit before the timeout triggers + */ + HAVE_NO_QUICK_QUIT, +}; + +/** + * @brief Do we collect statistics at the end? + */ +enum OPTION_COLLECT_STATISTICS { + /** + * @brief We collect statistics at the end + */ + COLLECT_STATISTICS, + + /** + * @brief We do not collect statistics at the end + */ + NO_COLLECT_STATISTICS, +}; + +/** + * @brief Do we collect views during run? + */ +enum OPTION_COLLECT_VIEW { + /** + * @brief We collect view during run + */ + COLLECT_VIEW, + + /** + * @brief We do not collect the view during run + */ + NO_COLLECT_VIEW, +}; + +/** + * Structure to define a single test + */ +struct SingleTestRun +{ + /** + * Name of the test + */ + char *name; + + /** + * Called with a single peer in order to initialise that peer + */ + InitPeer init_peer; + + /** + * Called directly after connecting to the service + */ + PreTest pre_test; + + /** + * Main function for each peer + */ + MainTest main_test; + + /** + * Callback called once the requested peers are available + */ + ReplyHandle reply_handle; + + /** + * Called directly before disconnecting from the service + */ + PostTest post_test; + + /** + * Function to evaluate the test results + */ + EvaluationCallback eval_cb; + + /** + * Request interval + */ + uint32_t request_interval; + + /** + * Number of Requests to make. + */ + uint32_t num_requests; + + /** + * Run with (-out) churn + */ + enum OPTION_CHURN have_churn; + + /** + * Quit test before timeout? + */ + enum OPTION_QUICK_QUIT have_quick_quit; + + /** + * Collect statistics at the end? + */ + enum OPTION_COLLECT_STATISTICS have_collect_statistics; + + /** + * Collect view during run? + */ + enum OPTION_COLLECT_VIEW have_collect_view; + + /** + * @brief Mark which values from the statistics service to collect at the end + * of the run + */ + uint32_t stat_collect_flags; +} cur_test_run; + +/** + * Did we finish the test? + */ +static int post_test; + +/** + * Are we shutting down? + */ +static int in_shutdown; + +/** + * Append arguments to file + */ +static void +tofile_ (const char *file_name, const char *line) +{ + struct GNUNET_DISK_FileHandle *f; + /* char output_buffer[512]; */ + size_t size; + /* int size; */ + size_t size2; + + if (NULL == (f = GNUNET_DISK_file_open (file_name, + GNUNET_DISK_OPEN_APPEND | + GNUNET_DISK_OPEN_WRITE | + GNUNET_DISK_OPEN_CREATE, + GNUNET_DISK_PERM_USER_READ | + GNUNET_DISK_PERM_USER_WRITE | + GNUNET_DISK_PERM_GROUP_READ | + GNUNET_DISK_PERM_OTHER_READ))) + { + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "Not able to open file %s\n", + file_name); + return; + } + /* size = GNUNET_snprintf (output_buffer, + sizeof (output_buffer), + "%llu %s\n", + GNUNET_TIME_absolute_get ().abs_value_us, + line); + if (0 > size) + { + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "Failed to write string to buffer (size: %i)\n", + size); + return; + } */ + + size = strlen (line) * sizeof (char); + + size2 = GNUNET_DISK_file_write (f, line, size); + if (size != size2) + { + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "Unable to write to file! (Size: %lu, size2: %lu)\n", + size, + size2); + if (GNUNET_YES != GNUNET_DISK_file_close (f)) + { + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "Unable to close file\n"); + } + return; + } + + if (GNUNET_YES != GNUNET_DISK_file_close (f)) + { + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "Unable to close file\n"); + } +} + +/** + * This function is used to facilitate writing important information to disk + */ +#define tofile(file_name, ...) do {\ + char tmp_buf[512];\ + int size;\ + size = GNUNET_snprintf(tmp_buf,sizeof(tmp_buf),__VA_ARGS__);\ + if (0 > size)\ + GNUNET_log (GNUNET_ERROR_TYPE_WARNING,\ + "Failed to create tmp_buf\n");\ + else\ + tofile_(file_name,tmp_buf);\ + } while (0); + + +/** + * Write the ids and their according index in the given array to a file + * Unused + */ +/* static void +ids_to_file (char *file_name, + struct GNUNET_PeerIdentity *peer_ids, + unsigned int num_peer_ids) +{ + unsigned int i; + + for (i=0 ; i < num_peer_ids ; i++) + { + to_file (file_name, + "%u\t%s", + i, + GNUNET_i2s_full (&peer_ids[i])); + } +} */ + +/** + * Test the success of a single test + */ +static int +evaluate (void) +{ + unsigned int i; + int tmp_ok; + + tmp_ok = 1; + + for (i = 0; i < num_peers; i++) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "%u. peer [%s] received %u of %u expected peer_ids: %i\n", + i, + GNUNET_i2s (rps_peers[i].peer_id), + rps_peers[i].num_recv_ids, + rps_peers[i].num_ids_to_request, + (rps_peers[i].num_ids_to_request == rps_peers[i].num_recv_ids)); + tmp_ok &= (rps_peers[i].num_ids_to_request == rps_peers[i].num_recv_ids); + } + return tmp_ok? 0 : 1; +} + + +/** + * Creates an oplist entry and adds it to the oplist DLL + */ +static struct OpListEntry * +make_oplist_entry () +{ + struct OpListEntry *entry; + + entry = GNUNET_new (struct OpListEntry); + GNUNET_CONTAINER_DLL_insert_tail (oplist_head, oplist_tail, entry); + return entry; +} + + +/** + * @brief Checks if given peer already received its statistics value from the + * statistics service. + * + * @param rps_peer the peer to check for + * + * @return #GNUNET_YES if so + * #GNUNET_NO otherwise + */ +static int check_statistics_collect_completed_single_peer ( + const struct RPSPeer *rps_peer) +{ + if (cur_test_run.stat_collect_flags != + (cur_test_run.stat_collect_flags & + rps_peer->stat_collected_flags)) + { + return GNUNET_NO; + } + return GNUNET_YES; +} +/** + * @brief Checks if all peers already received their statistics value from the + * statistics service. + * + * @return #GNUNET_YES if so + * #GNUNET_NO otherwise + */ +static int check_statistics_collect_completed () +{ + uint32_t i; + + for (i = 0; i < num_peers; i++) + { + if (GNUNET_NO == check_statistics_collect_completed_single_peer (&rps_peers[i])) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "At least Peer %" PRIu32 " did not yet receive all statistics values\n", + i); + return GNUNET_NO; + } + } + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "All peers received their statistics values\n"); + return GNUNET_YES; +} + +/** + * Task run on timeout to shut everything down. + */ +static void +shutdown_op (void *cls) +{ + unsigned int i; + + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "Shutdown task scheduled, going down.\n"); + in_shutdown = GNUNET_YES; + if (NULL != post_test_task) + { + GNUNET_SCHEDULER_cancel (post_test_task); + } + if (NULL != churn_task) + { + GNUNET_SCHEDULER_cancel (churn_task); + churn_task = NULL; + } + for (i = 0; i < num_peers; i++) + { + if (NULL != rps_peers[i].rps_handle) + { + GNUNET_RPS_disconnect (rps_peers[i].rps_handle); + } + if (NULL != rps_peers[i].op) + { + GNUNET_TESTBED_operation_done (rps_peers[i].op); + } + } +} + + +/** + * Task run on timeout to collect statistics and potentially shut down. + */ +static void +post_test_op (void *cls) +{ + unsigned int i; + + post_test_task = NULL; + post_test = GNUNET_YES; + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "Post test task scheduled, going down.\n"); + if (NULL != churn_task) + { + GNUNET_SCHEDULER_cancel (churn_task); + churn_task = NULL; + } + for (i = 0; i < num_peers; i++) + { + if (NULL != rps_peers[i].op) + { + GNUNET_TESTBED_operation_done (rps_peers[i].op); + rps_peers[i].op = NULL; + } + if (NULL != cur_test_run.post_test) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Executing post_test for peer %u\n", i); + cur_test_run.post_test (&rps_peers[i]); + } + } + /* If we do not collect statistics, shut down directly */ + if (NO_COLLECT_STATISTICS == cur_test_run.have_collect_statistics || + GNUNET_YES == check_statistics_collect_completed()) + { + GNUNET_SCHEDULER_shutdown (); + } +} + + +/** + * Seed peers. + */ +static void +seed_peers (void *cls) +{ + struct RPSPeer *peer = cls; + unsigned int amount; + unsigned int i; + + // TODO if malicious don't seed mal peers + amount = round (.5 * num_peers); + + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Seeding peers:\n"); + for (i = 0 ; i < amount ; i++) + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Seeding %u. peer: %s\n", + i, + GNUNET_i2s (&rps_peer_ids[i])); + + GNUNET_RPS_seed_ids (peer->rps_handle, amount, rps_peer_ids); +} + + +/** + * Get the id of peer i. + */ + void +info_cb (void *cb_cls, + struct GNUNET_TESTBED_Operation *op, + const struct GNUNET_TESTBED_PeerInformation *pinfo, + const char *emsg) +{ + struct OpListEntry *entry = (struct OpListEntry *) cb_cls; + + if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test) + { + return; + } + + if (NULL == pinfo || NULL != emsg) + { + GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Got Error: %s\n", emsg); + GNUNET_TESTBED_operation_done (entry->op); + return; + } + + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Peer %u is %s\n", + entry->index, + GNUNET_i2s (pinfo->result.id)); + + rps_peer_ids[entry->index] = *(pinfo->result.id); + rps_peers[entry->index].peer_id = &rps_peer_ids[entry->index]; + + GNUNET_assert (GNUNET_OK == + GNUNET_CONTAINER_multipeermap_put (peer_map, + &rps_peer_ids[entry->index], + &rps_peers[entry->index], + GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)); + tofile ("/tmp/rps/peer_ids", + "%u\t%s\n", + entry->index, + GNUNET_i2s_full (&rps_peer_ids[entry->index])); + + GNUNET_CONTAINER_DLL_remove (oplist_head, oplist_tail, entry); + GNUNET_TESTBED_operation_done (entry->op); + GNUNET_free (entry); +} + + +/** + * Callback to be called when RPS service connect operation is completed + * + * @param cls the callback closure from functions generating an operation + * @param op the operation that has been finished + * @param ca_result the RPS service handle returned from rps_connect_adapter + * @param emsg error message in case the operation has failed; will be NULL if + * operation has executed successfully. + */ +static void +rps_connect_complete_cb (void *cls, + struct GNUNET_TESTBED_Operation *op, + void *ca_result, + const char *emsg) +{ + struct RPSPeer *rps_peer = cls; + struct GNUNET_RPS_Handle *rps = ca_result; + + if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test) + { + return; + } + + rps_peer->rps_handle = rps; + rps_peer->online = GNUNET_YES; + num_peers_online++; + + GNUNET_assert (op == rps_peer->op); + if (NULL != emsg) + { + GNUNET_log (GNUNET_ERROR_TYPE_ERROR, + "Failed to connect to RPS service: %s\n", + emsg); + ok = 1; + GNUNET_SCHEDULER_shutdown (); + return; + } + + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Started client successfully\n"); + + cur_test_run.main_test (rps_peer); +} + + +/** + * Adapter function called to establish a connection to + * the RPS service. + * + * @param cls closure + * @param cfg configuration of the peer to connect to; will be available until + * GNUNET_TESTBED_operation_done() is called on the operation returned + * from GNUNET_TESTBED_service_connect() + * @return service handle to return in 'op_result', NULL on error + */ +static void * +rps_connect_adapter (void *cls, + const struct GNUNET_CONFIGURATION_Handle *cfg) +{ + struct GNUNET_RPS_Handle *h; + + h = GNUNET_RPS_connect (cfg); + + if (NULL != cur_test_run.pre_test) + cur_test_run.pre_test (cls, h); + + return h; +} + +/** + * Called to open a connection to the peer's statistics + * + * @param cls peer context + * @param cfg configuration of the peer to connect to; will be available until + * GNUNET_TESTBED_operation_done() is called on the operation returned + * from GNUNET_TESTBED_service_connect() + * @return service handle to return in 'op_result', NULL on error + */ +static void * +stat_connect_adapter (void *cls, + const struct GNUNET_CONFIGURATION_Handle *cfg) +{ + struct RPSPeer *peer = cls; + + peer->stats_h = GNUNET_STATISTICS_create ("rps-profiler", cfg); + return peer->stats_h; +} + +/** + * Called to disconnect from peer's statistics service + * + * @param cls peer context + * @param op_result service handle returned from the connect adapter + */ +static void +stat_disconnect_adapter (void *cls, void *op_result) +{ + struct RPSPeer *peer = cls; + + //GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch_cancel + // (peer->stats_h, "core", "# peers connected", + // stat_iterator, peer)); + //GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch_cancel + // (peer->stats_h, "nse", "# peers connected", + // stat_iterator, peer)); + GNUNET_STATISTICS_destroy (op_result, GNUNET_NO); + peer->stats_h = NULL; +} + +/** + * Called after successfully opening a connection to a peer's statistics + * service; we register statistics monitoring for CORE and NSE here. + * + * @param cls the callback closure from functions generating an operation + * @param op the operation that has been finished + * @param ca_result the service handle returned from GNUNET_TESTBED_ConnectAdapter() + * @param emsg error message in case the operation has failed; will be NULL if + * operation has executed successfully. + */ +static void +stat_complete_cb (void *cls, struct GNUNET_TESTBED_Operation *op, + void *ca_result, const char *emsg ) +{ + //struct GNUNET_STATISTICS_Handle *sh = ca_result; + //struct RPSPeer *peer = (struct RPSPeer *) cls; + + if (NULL != emsg) + { + GNUNET_break (0); + return; + } + //GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch + // (sh, "core", "# peers connected", + // stat_iterator, peer)); + //GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch + // (sh, "nse", "# peers connected", + // stat_iterator, peer)); +} + + +/** + * Adapter function called to destroy connection to + * RPS service. + * + * @param cls closure + * @param op_result service handle returned from the connect adapter + */ +static void +rps_disconnect_adapter (void *cls, + void *op_result) +{ + struct RPSPeer *peer = cls; + struct GNUNET_RPS_Handle *h = op_result; + + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "disconnect_adapter()\n"); + GNUNET_assert (NULL != peer); + GNUNET_RPS_disconnect (h); + peer->rps_handle = NULL; +} + + +/*********************************************************************** + * Definition of tests +***********************************************************************/ + +/** + * Callback to call on receipt of a reply + * + * @param cls closure + * @param n number of peers + * @param recv_peers the received peers + */ +static void +default_reply_handle (void *cls, + uint64_t n, + const struct GNUNET_PeerIdentity *recv_peers) +{ + struct RPSPeer *rps_peer; + struct PendingReply *pending_rep = (struct PendingReply *) cls; + unsigned int i; + + rps_peer = pending_rep->rps_peer; + GNUNET_CONTAINER_DLL_remove (rps_peer->pending_rep_head, + rps_peer->pending_rep_tail, + pending_rep); + rps_peer->num_pending_reps--; + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "[%s] got %" PRIu64 " peers:\n", + GNUNET_i2s (rps_peer->peer_id), + n); + + for (i = 0; i < n; i++) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "%u: %s\n", + i, + GNUNET_i2s (&recv_peers[i])); + + rps_peer->num_recv_ids++; + } + + if (0 == evaluate () && HAVE_QUICK_QUIT == cur_test_run.have_quick_quit) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test succeeded before timeout\n"); + GNUNET_assert (NULL != post_test_task); + GNUNET_SCHEDULER_cancel (post_test_task); + post_test_task = GNUNET_SCHEDULER_add_now (&post_test_op, NULL); + GNUNET_assert (NULL!= post_test_task); + } +} + +/** + * Request random peers. + */ +static void +request_peers (void *cls) +{ + struct PendingRequest *pending_req = cls; + struct RPSPeer *rps_peer; + struct PendingReply *pending_rep; + + if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test) + return; + rps_peer = pending_req->rps_peer; + GNUNET_assert (1 <= rps_peer->num_pending_reqs); + GNUN |