X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Frps%2Ftest_rps.c;h=c5a2c04582b4044fef9f2cdf62125d515222b4f3;hb=0166171bb810849613164867bc3450bf5bf73a5a;hp=32bce40986a52b6e5ee836cf4cc401015260695a;hpb=c8a94577fe87d86aa1537e85e5e294e80c64f356;p=oweals%2Fgnunet.git diff --git a/src/rps/test_rps.c b/src/rps/test_rps.c index 32bce4098..c5a2c0458 100644 --- a/src/rps/test_rps.c +++ b/src/rps/test_rps.c @@ -1,6 +1,6 @@ /* This file is part of GNUnet. - Copyright (C) 2009, 2012 Christian Grothoff (and other contributing authors) + 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 General Public License as published @@ -14,11 +14,11 @@ You should have received a copy of the GNU General Public License along with GNUnet; see the file COPYING. If not, write to the - Free Software Foundation, Inc., 59 Temple Place - Suite 330, - Boston, MA 02111-1307, USA. + Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, + Boston, MA 02110-1301, USA. */ /** - * @file rps/test_rps_multipeer.c + * @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 @@ -27,7 +27,10 @@ #include "platform.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 @@ -35,7 +38,7 @@ /** * How many peers do we start? */ -uint32_t num_peers; +static uint32_t num_peers; /** * How long do we run the test? @@ -59,6 +62,19 @@ static unsigned int mal_type = 0; */ static struct GNUNET_TESTBED_Peer **testbed_peers; +/** + * @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 @@ -81,10 +97,10 @@ struct OpListEntry struct GNUNET_TESTBED_Operation *op; /** - * Depending on whether we start or stop NSE service at the peer set this to 1 - * or -1 + * Depending on whether we start or stop RPS service at the peer, set this to + * #PEER_GO_ONLINE (1) or #PEER_GO_OFFLINE (-1) */ - int delta; + enum PEER_ONLINE_DELTA delta; /** * Index of the regarding peer @@ -103,6 +119,52 @@ static struct OpListEntry *oplist_head; 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. */ @@ -139,14 +201,83 @@ struct RPSPeer int online; /** - * Received PeerIDs + * 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 */ - struct GNUNET_PeerIdentity *rec_ids; + unsigned int num_pending_reps; /** * Number of received PeerIDs */ - unsigned int num_rec_ids; + 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; +}; + +enum STAT_TYPE +{ + STAT_TYPE_ROUNDS = 0x1, /* 1 */ + STAT_TYPE_BLOCKS = 0x2, /* 2 */ + STAT_TYPE_BLOCKS_MANY_PUSH = 0x4, /* 3 */ + STAT_TYPE_BLOCKS_FEW_PUSH = 0x8, /* 4 */ + STAT_TYPE_BLOCKS_FEW_PULL = 0x10, /* 5 */ + STAT_TYPE_ISSUED_PUSH_SEND = 0x20, /* 6 */ + STAT_TYPE_ISSUED_PULL_REQ = 0x40, /* 7 */ + STAT_TYPE_ISSUED_PULL_REP = 0x80, /* 8 */ + STAT_TYPE_SENT_PUSH_SEND = 0x100, /* 9 */ + STAT_TYPE_SENT_PULL_REQ = 0x200, /* 10 */ + STAT_TYPE_SENT_PULL_REP = 0x400, /* 11 */ + STAT_TYPE_RECV_PUSH_SEND = 0x800, /* 12 */ + STAT_TYPE_RECV_PULL_REQ = 0x1000, /* 13 */ + STAT_TYPE_RECV_PULL_REP = 0x2000, /* 14 */ + STAT_TYPE_MAX = 0x80000000, /* 32 */ +}; + +struct STATcls +{ + struct RPSPeer *rps_peer; + enum STAT_TYPE stat_type; }; @@ -155,11 +286,26 @@ struct RPSPeer */ 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. */ @@ -170,34 +316,100 @@ static unsigned int num_peers_online; */ static int ok; +/** + * 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); /** - * Called directly after connecting to the service + * @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) (void *cls, struct GNUNET_RPS_Handle *h); +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 () - * Executes functions to test the api/service + * 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) (void *cls, struct GNUNET_RPS_Handle *h); +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, +}; /** * Structure to define a single test @@ -209,16 +421,26 @@ struct SingleTestRun */ 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; /** - * Function to execute the functions to be tested + * 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 */ @@ -228,32 +450,163 @@ struct SingleTestRun * 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; + + /** + * @brief Mark which values from the statistics service to collect at the end + * of the run + */ + uint32_t stat_collect_flags; } cur_test_run; +/** + * 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 (struct RPSPeer *loc_rps_peers, - unsigned int num_loc_rps_peers, - unsigned int expected_recv) +evaluate (void) { unsigned int i; int tmp_ok; - tmp_ok = (1 == loc_rps_peers[0].num_rec_ids); + tmp_ok = 1; - for (i = 0 ; i < num_loc_rps_peers ; i++) + 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 (loc_rps_peers[i].peer_id), - loc_rps_peers[i].num_rec_ids, - expected_recv, - (1 == loc_rps_peers[i].num_rec_ids)); - tmp_ok &= (1 == loc_rps_peers[i].num_rec_ids); + "%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; } @@ -274,161 +627,143 @@ make_oplist_entry () /** - * Callback to be called when NSE service is started or stopped at peers + * @brief Checks if given peer already received its statistics value from the + * statistics service. * - * @param cls NULL - * @param op the operation handle - * @param emsg NULL on success; otherwise an error description + * @param rps_peer the peer to check for + * + * @return #GNUNET_YES if so + * #GNUNET_NO otherwise */ -static void -churn_cb (void *cls, - struct GNUNET_TESTBED_Operation *op, - const char *emsg) +static int check_statistics_collect_completed_single_peer ( + const struct RPSPeer *rps_peer) { - // FIXME - struct OpListEntry *entry = cls; - - GNUNET_TESTBED_operation_done (entry->op); - if (NULL != emsg) + if (cur_test_run.stat_collect_flags != + (cur_test_run.stat_collect_flags & + rps_peer->stat_collected_flags)) { - GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Failed to start/stop RPS at a peer\n"); - GNUNET_SCHEDULER_shutdown (); - return; - } - GNUNET_assert (0 != entry->delta); - - num_peers_online += entry->delta; - - if (0 > entry->delta) - { /* Peer hopefully just went offline */ - if (GNUNET_YES != rps_peers[entry->index].online) - { - GNUNET_log (GNUNET_ERROR_TYPE_WARNING, - "peer %s was expected to go offline but is still marked as online\n", - GNUNET_i2s (rps_peers[entry->index].peer_id)); - GNUNET_break (0); - } - else - { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "peer %s probably went offline as expected\n", - GNUNET_i2s (rps_peers[entry->index].peer_id)); - } - rps_peers[entry->index].online = GNUNET_NO; + 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; - else if (0 < entry->delta) - { /* Peer hopefully just went online */ - if (GNUNET_NO != rps_peers[entry->index].online) - { - GNUNET_log (GNUNET_ERROR_TYPE_WARNING, - "peer %s was expected to go online but is still marked as offline\n", - GNUNET_i2s (rps_peers[entry->index].peer_id)); - GNUNET_break (0); - } - else + 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, - "peer %s probably went online as expected\n", - GNUNET_i2s (rps_peers[entry->index].peer_id)); + "At least Peer %" PRIu32 " did not yet receive all statistics values\n", + i); + return GNUNET_NO; } - rps_peers[entry->index].online = GNUNET_YES; } - - GNUNET_CONTAINER_DLL_remove (oplist_head, oplist_tail, entry); - GNUNET_free (entry); - //if (num_peers_in_round[current_round] == peers_running) - // run_round (); + 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_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) +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 != churn_task) + { GNUNET_SCHEDULER_cancel (churn_task); - - for (i = 0 ; i < num_peers ; i++) - GNUNET_TESTBED_operation_done (rps_peers[i].op); - GNUNET_SCHEDULER_shutdown (); + churn_task = NULL; + } + for (i = 0; i < num_peers; i++) + { + if (NULL != rps_peers[i].op) + GNUNET_TESTBED_operation_done (rps_peers[i].op); + 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 (); + } } /** - * Callback to call on receipt of a reply - * - * @param cls closure - * @param n number of peers - * @param recv_peers the received peers + * Seed peers. */ static void -handle_reply (void *cls, uint64_t n, const struct GNUNET_PeerIdentity *recv_peers) +seed_peers (void *cls) { - struct RPSPeer *rps_peer = (struct RPSPeer *) cls; + struct RPSPeer *peer = cls; + unsigned int amount; unsigned int i; - 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])); - - GNUNET_array_append (rps_peer->rec_ids, rps_peer->num_rec_ids, recv_peers[i]); - } -} - - -/** - * Request random peers. - */ - void -request_peers (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc) -{ - struct RPSPeer *rps_peer = (struct RPSPeer *) cls; + // TODO if malicious don't seed mal peers + amount = round (.5 * num_peers); - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Requesting one peer\n"); + 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])); - (void) GNUNET_RPS_request_peers (rps_peer->rps_handle, 1, handle_reply, rps_peer); - //rps_peer->req_handle = GNUNET_RPS_request_peers (rps_peer->rps_handle, 1, handle_reply, rps_peer); + GNUNET_RPS_seed_ids (peer->rps_handle, amount, rps_peer_ids); } /** * Seed peers. */ - void -seed_peers (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) +static void +seed_peers_big (void *cls) { + struct RPSPeer *peer = cls; + unsigned int seed_msg_size; + uint32_t num_peers_max; unsigned int amount; - struct RPSPeer *peer = (struct RPSPeer *) cls; 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++) + seed_msg_size = 8; /* sizeof (struct GNUNET_RPS_CS_SeedMessage) */ + num_peers_max = (GNUNET_MAX_MESSAGE_SIZE - seed_msg_size) / + sizeof (struct GNUNET_PeerIdentity); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Peers that fit in one seed msg; %u\n", + num_peers_max); + amount = num_peers_max + (0.5 * num_peers_max); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Seeding many (%u) peers:\n", + amount); + struct GNUNET_PeerIdentity ids_to_seed[amount]; + for (i = 0; i < amount; i++) + { + ids_to_seed[i] = *peer->peer_id; GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Seeding %u. peer: %s\n", i, - GNUNET_i2s (&rps_peer_ids[i])); + GNUNET_i2s (&ids_to_seed[i])); + } - GNUNET_RPS_seed_ids (peer->rps_handle, amount, rps_peer_ids); + GNUNET_RPS_seed_ids (peer->rps_handle, amount, ids_to_seed); } - /** * Get the id of peer i. */ @@ -440,9 +775,15 @@ info_cb (void *cb_cls, { struct OpListEntry *entry = (struct OpListEntry *) cb_cls; + if (GNUNET_YES == in_shutdown) + { + return; + } + if (NULL == pinfo || NULL != emsg) { GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Got Error: %s\n", emsg); + GNUNET_TESTBED_operation_done (entry->op); return; } @@ -453,12 +794,19 @@ info_cb (void *cb_cls, rps_peer_ids[entry->index] = *(pinfo->result.id); rps_peers[entry->index].peer_id = &rps_peer_ids[entry->index]; - rps_peers[entry->index].rec_ids = NULL; - rps_peers[entry->index].num_rec_ids = 0; - GNUNET_TESTBED_operation_done (entry->op); + 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); } @@ -481,6 +829,11 @@ rps_connect_complete_cb (void *cls, struct RPSPeer *rps_peer = cls; struct GNUNET_RPS_Handle *rps = ca_result; + if (GNUNET_YES == in_shutdown) + { + return; + } + rps_peer->rps_handle = rps; rps_peer->online = GNUNET_YES; num_peers_online++; @@ -526,6 +879,76 @@ rps_connect_adapter (void *cls, 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 @@ -538,8 +961,11 @@ static void rps_disconnect_adapter (void *cls, void *op_result) { + struct RPSPeer *peer = cls; struct GNUNET_RPS_Handle *h = op_result; + GNUNET_assert (NULL != peer); GNUNET_RPS_disconnect (h); + peer->rps_handle = NULL; } @@ -547,30 +973,230 @@ rps_disconnect_adapter (void *cls, * Definition of tests ***********************************************************************/ +// TODO check whether tests can be stopped earlier static int default_eval_cb (void) { - return evaluate (rps_peers, num_peers, 1); + return evaluate (); } static int no_eval (void) { - return 1; + return 0; +} + +/** + * Initialise given RPSPeer + */ +static void default_init_peer (struct RPSPeer *rps_peer) +{ + rps_peer->num_ids_to_request = 1; +} + +/** + * 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 != shutdown_task); + GNUNET_SCHEDULER_cancel (shutdown_task); + shutdown_task = GNUNET_SCHEDULER_add_now (&shutdown_op, NULL); + GNUNET_assert (NULL!= shutdown_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) + return; + rps_peer = pending_req->rps_peer; + GNUNET_assert (1 <= rps_peer->num_pending_reqs); + GNUNET_CONTAINER_DLL_remove (rps_peer->pending_req_head, + rps_peer->pending_req_tail, + pending_req); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Requesting one peer\n"); + pending_rep = GNUNET_new (struct PendingReply); + pending_rep->rps_peer = rps_peer; + pending_rep->req_handle = GNUNET_RPS_request_peers (rps_peer->rps_handle, + 1, + cur_test_run.reply_handle, + pending_rep); + GNUNET_CONTAINER_DLL_insert_tail (rps_peer->pending_rep_head, + rps_peer->pending_rep_tail, + pending_rep); + rps_peer->num_pending_reps++; + rps_peer->num_pending_reqs--; +} + +static void +cancel_pending_req (struct PendingRequest *pending_req) +{ + struct RPSPeer *rps_peer; + + rps_peer = pending_req->rps_peer; + GNUNET_CONTAINER_DLL_remove (rps_peer->pending_req_head, + rps_peer->pending_req_tail, + pending_req); + rps_peer->num_pending_reqs--; + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Cancelling pending request\n"); + GNUNET_SCHEDULER_cancel (pending_req->request_task); + GNUNET_free (pending_req); +} + +static void +cancel_request (struct PendingReply *pending_rep) +{ + struct RPSPeer *rps_peer; + + 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, + "Cancelling request\n"); + GNUNET_RPS_request_cancel (pending_rep->req_handle); + GNUNET_free (pending_rep); +} + +/** + * Cancel a request. + */ +static void +cancel_request_cb (void *cls) +{ + struct RPSPeer *rps_peer = cls; + struct PendingReply *pending_rep; + + if (GNUNET_YES == in_shutdown) + return; + pending_rep = rps_peer->pending_rep_head; + GNUNET_assert (1 <= rps_peer->num_pending_reps); + cancel_request (pending_rep); +} + + +/** + * Schedule requests for peer @a rps_peer that have neither been scheduled, nor + * issued, nor replied + */ +void +schedule_missing_requests (struct RPSPeer *rps_peer) +{ + unsigned int i; + struct PendingRequest *pending_req; + + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Scheduling %u - %u missing requests\n", + rps_peer->num_ids_to_request, + rps_peer->num_pending_reqs + rps_peer->num_pending_reps); + GNUNET_assert (rps_peer->num_pending_reqs + rps_peer->num_pending_reps <= + rps_peer->num_ids_to_request); + for (i = rps_peer->num_pending_reqs + rps_peer->num_pending_reps; + i < rps_peer->num_ids_to_request; i++) + { + pending_req = GNUNET_new (struct PendingRequest); + pending_req->rps_peer = rps_peer; + pending_req->request_task = GNUNET_SCHEDULER_add_delayed ( + GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, + cur_test_run.request_interval * i), + request_peers, + pending_req); + GNUNET_CONTAINER_DLL_insert_tail (rps_peer->pending_req_head, + rps_peer->pending_req_tail, + pending_req); + rps_peer->num_pending_reqs++; + } +} + +void +cancel_pending_req_rep (struct RPSPeer *rps_peer) +{ + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Cancelling all (pending) requests.\n"); + while (NULL != rps_peer->pending_req_head) + cancel_pending_req (rps_peer->pending_req_head); + GNUNET_assert (0 == rps_peer->num_pending_reqs); + while (NULL != rps_peer->pending_rep_head) + cancel_request (rps_peer->pending_rep_head); + GNUNET_assert (0 == rps_peer->num_pending_reps); } /*********************************** * MALICIOUS ***********************************/ + +/** + * Initialise only non-mal RPSPeers + */ +static void mal_init_peer (struct RPSPeer *rps_peer) +{ + if (rps_peer->index >= round (portion * num_peers)) + rps_peer->num_ids_to_request = 1; +} + + +/** + * @brief Set peers to (non-)malicious before execution + * + * Of signature #PreTest + * + * @param rps_peer the peer to set (non-) malicious + * @param h the handle to the service + */ static void -mal_pre (void *cls, struct GNUNET_RPS_Handle *h) +mal_pre (struct RPSPeer *rps_peer, struct GNUNET_RPS_Handle *h) { #ifdef ENABLE_MALICIOUS uint32_t num_mal_peers; - struct RPSPeer *rps_peer = (struct RPSPeer *) cls; - GNUNET_assert (1 >= portion - && 0 < portion); + GNUNET_assert ( (1 >= portion) && + (0 < portion) ); num_mal_peers = round (portion * num_peers); if (rps_peer->index < num_mal_peers) @@ -581,7 +1207,8 @@ mal_pre (void *cls, struct GNUNET_RPS_Handle *h) GNUNET_i2s (rps_peer->peer_id), num_mal_peers); - GNUNET_RPS_act_malicious (h, mal_type, num_mal_peers, rps_peer_ids); + GNUNET_RPS_act_malicious (h, mal_type, num_mal_peers, + rps_peer_ids, target_peer); } #endif /* ENABLE_MALICIOUS */ } @@ -591,9 +1218,14 @@ mal_cb (struct RPSPeer *rps_peer) { uint32_t num_mal_peers; + if (GNUNET_YES == in_shutdown) + { + return; + } + #ifdef ENABLE_MALICIOUS - GNUNET_assert (1 >= portion - && 0 < portion); + GNUNET_assert ( (1 >= portion) && + (0 < portion) ); num_mal_peers = round (portion * num_peers); if (rps_peer->index >= num_mal_peers) @@ -601,23 +1233,11 @@ mal_cb (struct RPSPeer *rps_peer) it's not sampling */ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2), seed_peers, rps_peer); - GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10), - request_peers, rps_peer); + schedule_missing_requests (rps_peer); } #endif /* ENABLE_MALICIOUS */ } -static int -mal_eval (void) -{ - unsigned int num_mal_peers; - - num_mal_peers = round (num_peers * portion); - return evaluate (&rps_peers[num_mal_peers], - num_peers - (num_mal_peers), - 1); -} - /*********************************** * SINGLE_REQUEST @@ -625,8 +1245,12 @@ mal_eval (void) static void single_req_cb (struct RPSPeer *rps_peer) { - GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5), - request_peers, rps_peer); + if (GNUNET_YES == in_shutdown) + { + return; + } + + schedule_missing_requests (rps_peer); } /*********************************** @@ -635,10 +1259,12 @@ single_req_cb (struct RPSPeer *rps_peer) static void delay_req_cb (struct RPSPeer *rps_peer) { - GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5), - request_peers, rps_peer); - GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10), - request_peers, rps_peer); + if (GNUNET_YES == in_shutdown) + { + return; + } + + schedule_missing_requests (rps_peer); } /*********************************** @@ -647,8 +1273,14 @@ delay_req_cb (struct RPSPeer *rps_peer) static void seed_cb (struct RPSPeer *rps_peer) { - GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10), - seed_peers, rps_peer); + if (GNUNET_YES == in_shutdown) + { + return; + } + + GNUNET_SCHEDULER_add_delayed ( + GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10), + seed_peers, rps_peer); } /*********************************** @@ -657,7 +1289,15 @@ seed_cb (struct RPSPeer *rps_peer) static void seed_big_cb (struct RPSPeer *rps_peer) { - // TODO test seeding > GNUNET_SERVER_MAX_MESSAGE_SIZE peers + if (GNUNET_YES == in_shutdown) + { + return; + } + + // TODO test seeding > GNUNET_MAX_MESSAGE_SIZE peers + GNUNET_SCHEDULER_add_delayed ( + GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2), + seed_peers_big, rps_peer); } /*********************************** @@ -675,10 +1315,15 @@ single_peer_seed_cb (struct RPSPeer *rps_peer) static void seed_req_cb (struct RPSPeer *rps_peer) { - GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2), - seed_peers, rps_peer); - GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 15), - request_peers, rps_peer); + if (GNUNET_YES == in_shutdown) + { + return; + } + + GNUNET_SCHEDULER_add_delayed ( + GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2), + seed_peers, rps_peer); + schedule_missing_requests (rps_peer); } //TODO start big mal @@ -689,16 +1334,207 @@ seed_req_cb (struct RPSPeer *rps_peer) static void req_cancel_cb (struct RPSPeer *rps_peer) { - // TODO + if (GNUNET_YES == in_shutdown) + { + return; + } + + schedule_missing_requests (rps_peer); + GNUNET_SCHEDULER_add_delayed ( + GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, + (cur_test_run.request_interval + 1)), + cancel_request_cb, rps_peer); +} + +/*********************************** + * CHURN +***********************************/ + +static void +churn (void *cls); + +/** + * @brief Starts churn + * + * Has signature of #MainTest + * + * This is not implemented too nicely as this is called for each peer, but we + * only need to call it once. (Yes we check that we only schedule the task + * once.) + * + * @param rps_peer The peer it's called for + */ +static void +churn_test_cb (struct RPSPeer *rps_peer) +{ + if (GNUNET_YES == in_shutdown) + { + return; + } + + /* Start churn */ + if (HAVE_CHURN == cur_test_run.have_churn && NULL == churn_task) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Starting churn task\n"); + churn_task = GNUNET_SCHEDULER_add_delayed ( + GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5), + churn, + NULL); + } else { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Not starting churn task\n"); + } + + schedule_missing_requests (rps_peer); } /*********************************** * PROFILER ***********************************/ + +/** + * Callback to be called when RPS service is started or stopped at peers + * + * @param cls NULL + * @param op the operation handle + * @param emsg NULL on success; otherwise an error description + */ static void -churn (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) +churn_cb (void *cls, + struct GNUNET_TESTBED_Operation *op, + const char *emsg) +{ + // FIXME + struct OpListEntry *entry = cls; + + if (GNUNET_YES == in_shutdown) + { + return; + } + + GNUNET_TESTBED_operation_done (entry->op); + if (NULL != emsg) + { + GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Failed to start/stop RPS at a peer\n"); + GNUNET_SCHEDULER_shutdown (); + return; + } + GNUNET_assert (0 != entry->delta); + + num_peers_online += entry->delta; + + if (PEER_GO_OFFLINE == entry->delta) + { /* Peer hopefully just went offline */ + if (GNUNET_YES != rps_peers[entry->index].online) + { + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "peer %s was expected to go offline but is still marked as online\n", + GNUNET_i2s (rps_peers[entry->index].peer_id)); + GNUNET_break (0); + } + else + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "peer %s probably went offline as expected\n", + GNUNET_i2s (rps_peers[entry->index].peer_id)); + } + rps_peers[entry->index].online = GNUNET_NO; + } + + else if (PEER_GO_ONLINE < entry->delta) + { /* Peer hopefully just went online */ + if (GNUNET_NO != rps_peers[entry->index].online) + { + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "peer %s was expected to go online but is still marked as offline\n", + GNUNET_i2s (rps_peers[entry->index].peer_id)); + GNUNET_break (0); + } + else + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "peer %s probably went online as expected\n", + GNUNET_i2s (rps_peers[entry->index].peer_id)); + if (NULL != cur_test_run.pre_test) + { + cur_test_run.pre_test (&rps_peers[entry->index], + rps_peers[entry->index].rps_handle); + schedule_missing_requests (&rps_peers[entry->index]); + } + } + rps_peers[entry->index].online = GNUNET_YES; + } + else + { + GNUNET_log (GNUNET_ERROR_TYPE_ERROR, + "Invalid value for delta: %i\n", entry->delta); + GNUNET_break (0); + } + + GNUNET_CONTAINER_DLL_remove (oplist_head, oplist_tail, entry); + rps_peers[entry->index].entry_op_manage = NULL; + GNUNET_free (entry); + //if (num_peers_in_round[current_round] == peers_running) + // run_round (); +} + +/** + * @brief Set the rps-service up or down for a specific peer + * + * @param i index of action + * @param j index of peer + * @param delta (#PEER_ONLINE_DELTA) down (-1) or up (1) + * @param prob_go_on_off the probability of the action + */ +static void +manage_service_wrapper (unsigned int i, unsigned int j, + enum PEER_ONLINE_DELTA delta, + double prob_go_on_off) +{ + struct OpListEntry *entry = NULL; + uint32_t prob; + + /* make sure that management operation is not already scheduled */ + if (NULL != rps_peers[j].entry_op_manage) + { + return; + } + + prob = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, + UINT32_MAX); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "%u. selected peer (%u: %s) is %s.\n", + i, + j, + GNUNET_i2s (rps_peers[j].peer_id), + (PEER_GO_ONLINE == delta) ? "online" : "offline"); + if (prob < prob_go_on_off * UINT32_MAX) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "%s goes %s\n", + GNUNET_i2s (rps_peers[j].peer_id), + (PEER_GO_OFFLINE == delta) ? "offline" : "online"); + + if (PEER_GO_OFFLINE == delta) + cancel_pending_req_rep (&rps_peers[j]); + entry = make_oplist_entry (); + entry->delta = delta; + entry->index = j; + entry->op = GNUNET_TESTBED_peer_manage_service (NULL, + testbed_peers[j], + "rps", + &churn_cb, + entry, + (PEER_GO_OFFLINE == delta) ? 0 : 1); + } + rps_peers[j].entry_op_manage = entry; +} + + +static void +churn (void *cls) { - struct OpListEntry *entry; unsigned int i; unsigned int j; double portion_online; @@ -706,7 +1542,15 @@ churn (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) double prob_go_offline; double portion_go_online; double portion_go_offline; - uint32_t prob; + + if (GNUNET_YES == in_shutdown) + { + return; + } + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Churn function executing\n"); + + churn_task = NULL; /* Should be invalid by now */ /* Compute the probability for an online peer to go offline * this round */ @@ -731,67 +1575,25 @@ churn (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) (unsigned int) num_peers); /* Go over 50% randomly chosen peers */ - for (i = 0 ; i < .5 * num_peers ; i++) + for (i = 0; i < .5 * num_peers; i++) { j = permut[i]; /* If online, shut down with certain probability */ if (GNUNET_YES == rps_peers[j].online) { - prob = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, - UINT32_MAX); - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%u. selected peer (%u: %s) is online.\n", - i, - j, - GNUNET_i2s (rps_peers[j].peer_id)); - if (prob < prob_go_offline * UINT32_MAX) - { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%s goes offline\n", - GNUNET_i2s (rps_peers[j].peer_id)); - - entry = make_oplist_entry (); - entry->delta = -1; - entry->index = j; - entry->op = GNUNET_TESTBED_peer_manage_service (NULL, - testbed_peers[j], - "rps", - &churn_cb, - entry, - 0); - } - } - - /* If offline, restart with certain probability */ - else if (GNUNET_NO == rps_peers[j].online) - { - prob = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, - UINT32_MAX); - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%u. selected peer (%u: %s) is offline.\n", - i, - j, - GNUNET_i2s (rps_peers[j].peer_id)); - if (prob < .66 * UINT32_MAX) - { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%s goes online\n", - GNUNET_i2s (rps_peers[j].peer_id)); - - entry = make_oplist_entry (); - entry->delta = 1; - entry->index = j; - entry->op = GNUNET_TESTBED_peer_manage_service (NULL, - testbed_peers[j], - "rps", - &churn_cb, - entry, - 1); - } - } + manage_service_wrapper (i, j, -1, prob_go_offline); + } + + /* If offline, restart with certain probability */ + else if (GNUNET_NO == rps_peers[j].online) + { + manage_service_wrapper (i, j, 1, 0.66); + } } + GNUNET_free (permut); + churn_task = GNUNET_SCHEDULER_add_delayed ( GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2), churn, @@ -799,28 +1601,293 @@ churn (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) } +/** + * Initialise given RPSPeer + */ +static void profiler_init_peer (struct RPSPeer *rps_peer) +{ + if (num_peers - 1 == rps_peer->index) + rps_peer->num_ids_to_request = cur_test_run.num_requests; +} + + +/** + * Callback to call on receipt of a reply + * + * @param cls closure + * @param n number of peers + * @param recv_peers the received peers + */ +static void +profiler_reply_handle (void *cls, + uint64_t n, + const struct GNUNET_PeerIdentity *recv_peers) +{ + struct RPSPeer *rps_peer; + struct RPSPeer *rcv_rps_peer; + char *file_name; + char *file_name_dh; + unsigned int i; + struct PendingReply *pending_rep = (struct PendingReply *) cls; + + rps_peer = pending_rep->rps_peer; + file_name = "/tmp/rps/received_ids"; + file_name_dh = "/tmp/rps/diehard_input"; + 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])); + tofile (file_name, + "%s\n", + GNUNET_i2s_full (&recv_peers[i])); + rcv_rps_peer = GNUNET_CONTAINER_multipeermap_get (peer_map, &recv_peers[i]); + GNUNET_assert (NULL != rcv_rps_peer); + tofile (file_name_dh, + "%" PRIu32 "\n", + (uint32_t) rcv_rps_peer->index); + } + default_reply_handle (cls, n, recv_peers); +} + + static void -profiler_pre (void *cls, struct GNUNET_RPS_Handle *h) +profiler_cb (struct RPSPeer *rps_peer) { - //churn_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, - // 10), - // churn, NULL); - mal_pre (cls, h); + if (GNUNET_YES == in_shutdown) + { + return; + } - if (NULL == churn_task) + /* Start churn */ + if (HAVE_CHURN == cur_test_run.have_churn && NULL == churn_task) { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Starting churn task\n"); churn_task = GNUNET_SCHEDULER_add_delayed ( - GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10), + GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5), churn, NULL); + } else { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Not starting churn task\n"); } + + /* Only request peer ids at one peer. + * (It's the before-last because last one is target of the focussed attack.) + */ + if (eval_peer == rps_peer) + schedule_missing_requests (rps_peer); } -static void -profiler_cb (struct RPSPeer *rps_peer) +/** + * Function called from #profiler_eval with a filename. + * + * @param cls closure + * @param filename complete filename (absolute path) + * @return #GNUNET_OK to continue to iterate, + * #GNUNET_NO to stop iteration with no error, + * #GNUNET_SYSERR to abort iteration with error! + */ +int +file_name_cb (void *cls, const char *filename) +{ + if (NULL != strstr (filename, "sampler_el")) + { + struct RPS_SamplerElement *s_elem; + struct GNUNET_CRYPTO_AuthKey auth_key; + const char *key_char; + uint32_t i; + + key_char = filename + 20; /* Length of "/tmp/rps/sampler_el-" */ + tofile (filename, "--------------------------\n"); + + auth_key = string_to_auth_key (key_char); + s_elem = RPS_sampler_elem_create (); + RPS_sampler_elem_set (s_elem, auth_key); + + for (i = 0; i < num_peers; i++) + { + RPS_sampler_elem_next (s_elem, &rps_peer_ids[i]); + } + RPS_sampler_elem_destroy (s_elem); + } + return GNUNET_OK; +} + +/** + * This is run after the test finished. + * + * Compute all perfect samples. + */ +int +profiler_eval (void) { - // We're not requesting peers - // TODO maybe seed + /* Compute perfect sample for each sampler element */ + if (-1 == GNUNET_DISK_directory_scan ("/tmp/rps/", file_name_cb, NULL)) + { + GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Scan of directory failed\n"); + } + + return evaluate (); +} + +/** + * Continuation called by #GNUNET_STATISTICS_get() functions. + * + * Remembers that this specific statistics value was received for this peer. + * Checks whether all peers received their statistics yet. + * Issues the shutdown. + * + * @param cls closure + * @param success #GNUNET_OK if statistics were + * successfully obtained, #GNUNET_SYSERR if not. + */ +void +post_test_shutdown_ready_cb (void *cls, + int success) +{ + struct STATcls *stat_cls = (struct STATcls *) cls; + struct RPSPeer *rps_peer = stat_cls->rps_peer; + if (GNUNET_OK == success) + { + /* set flag that we we got the value */ + rps_peer->stat_collected_flags |= stat_cls->stat_type; + } else { + GNUNET_log (GNUNET_ERROR_TYPE_ERROR, + "Peer %u did not receive statistics value\n", + rps_peer->index); + GNUNET_free (stat_cls); + GNUNET_break (0); + } + + if (NULL != rps_peer->stat_op && + GNUNET_YES == check_statistics_collect_completed_single_peer (rps_peer)) + { + GNUNET_TESTBED_operation_done (rps_peer->stat_op); + } + + if (GNUNET_YES == check_statistics_collect_completed()) + { + GNUNET_free (stat_cls); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Shutting down\n"); + GNUNET_SCHEDULER_shutdown (); + } else { + GNUNET_free (stat_cls); + } +} + +/** + * @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_FEW_PUSH: + return "# rounds blocked - no pushes"; + case STAT_TYPE_BLOCKS_FEW_PULL: + return "# rounds blocked - 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"; + ; + } +} + +/** + * Callback function to process statistic values. + * + * @param cls closure + * @param subsystem name of subsystem that created the statistic + * @param name the name of the datum + * @param value the current value + * @param is_persistent #GNUNET_YES if the value is persistent, #GNUNET_NO if not + * @return #GNUNET_OK to continue, #GNUNET_SYSERR to abort iteration + */ +int +stat_iterator (void *cls, + const char *subsystem, + const char *name, + uint64_t value, + int is_persistent) +{ + const struct STATcls *stat_cls = (const struct STATcls *) cls; + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got stat value: %s - %" PRIu64 "\n", + stat_type_2_str (stat_cls->stat_type), + value); + return GNUNET_OK; +} + +void post_profiler (struct RPSPeer *rps_peer) +{ + if (COLLECT_STATISTICS != cur_test_run.have_collect_statistics) + { + return; + } + + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Going to request statistic values with mask 0x%" PRIx32 "\n", + cur_test_run.stat_collect_flags); + + struct STATcls *stat_cls; + uint32_t stat_type; + for (stat_type = STAT_TYPE_ROUNDS; + stat_type < STAT_TYPE_MAX; + stat_type = stat_type <<1) + { + if (stat_type & cur_test_run.stat_collect_flags) + { + stat_cls = GNUNET_malloc (sizeof (struct STATcls)); + stat_cls->rps_peer = rps_peer; + stat_cls->stat_type = stat_type; + GNUNET_STATISTICS_get (rps_peer->stats_h, + "rps", + stat_type_2_str (stat_type), + post_test_shutdown_ready_cb, + stat_iterator, + (struct STATcls *) stat_cls); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Requested statistics for %s (peer %" PRIu32 ")\n", + stat_type_2_str (stat_type), + rps_peer->index); + } + } } @@ -851,50 +1918,78 @@ run (void *cls, { unsigned int i; struct OpListEntry *entry; + uint32_t num_mal_peers; + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "RUN was called\n"); + + /* Check whether we timed out */ + if (n_peers != num_peers || + NULL == peers || + 0 == links_succeeded) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Going down due to args (eg. timeout)\n"); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\tn_peers: %u\n", n_peers); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\tnum_peers: %" PRIu32 "\n", num_peers); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\tpeers: %p\n", peers); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\tlinks_succeeded: %u\n", links_succeeded); + GNUNET_SCHEDULER_shutdown (); + return; + } + + + /* Initialize peers */ testbed_peers = peers; num_peers_online = 0; - - for (i = 0 ; i < num_peers ; i++) + for (i = 0; i < num_peers; i++) { entry = make_oplist_entry (); entry->index = i; + rps_peers[i].index = i; + if (NULL != cur_test_run.init_peer) + cur_test_run.init_peer (&rps_peers[i]); entry->op = GNUNET_TESTBED_peer_get_information (peers[i], GNUNET_TESTBED_PIT_IDENTITY, &info_cb, entry); } - - // This seems not to work - //if (NULL != strstr (cur_test_run.name, "profiler")) - //{ - // churn_task = GNUNET_SCHEDULER_add_delayed ( - // GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5), - // churn, - // NULL); - //} - + /* Bring peers up */ + num_mal_peers = round (portion * num_peers); GNUNET_assert (num_peers == n_peers); - for (i = 0 ; i < n_peers ; i++) + for (i = 0; i < n_peers; i++) { rps_peers[i].index = i; - rps_peers[i].op = - GNUNET_TESTBED_service_connect (&rps_peers[i], - peers[i], - "rps", - &rps_connect_complete_cb, - &rps_peers[i], - &rps_connect_adapter, - &rps_disconnect_adapter, - &rps_peers[i]); + if ( (rps_peers[i].num_recv_ids < rps_peers[i].num_ids_to_request) || + (i < num_mal_peers) ) + { + rps_peers[i].op = + GNUNET_TESTBED_service_connect (&rps_peers[i], + peers[i], + "rps", + &rps_connect_complete_cb, + &rps_peers[i], + &rps_connect_adapter, + &rps_disconnect_adapter, + &rps_peers[i]); + } + /* Connect all peers to statistics service */ + if (COLLECT_STATISTICS == cur_test_run.have_collect_statistics) + { + rps_peers[i].stat_op = + GNUNET_TESTBED_service_connect (NULL, + peers[i], + "statistics", + stat_complete_cb, + &rps_peers[i], + &stat_connect_adapter, + &stat_disconnect_adapter, + &rps_peers[i]); + } } if (NULL != churn_task) GNUNET_SCHEDULER_cancel (churn_task); - - //GNUNET_SCHEDULER_add_delayed (TIMEOUT, &shutdown_task, NULL); - GNUNET_SCHEDULER_add_delayed (timeout, &shutdown_task, NULL); + shutdown_task = GNUNET_SCHEDULER_add_delayed (timeout, &shutdown_op, NULL); } @@ -908,29 +2003,43 @@ run (void *cls, int main (int argc, char *argv[]) { + int ret_value; + + num_peers = 5; + cur_test_run.name = "test-rps-default"; + cur_test_run.init_peer = default_init_peer; cur_test_run.pre_test = NULL; + cur_test_run.reply_handle = default_reply_handle; cur_test_run.eval_cb = default_eval_cb; + cur_test_run.post_test = NULL; + cur_test_run.have_churn = HAVE_CHURN; + cur_test_run.have_collect_statistics = NO_COLLECT_STATISTICS; + cur_test_run.stat_collect_flags = 0; churn_task = NULL; + timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30); if (strstr (argv[0], "malicious") != NULL) { cur_test_run.pre_test = mal_pre; cur_test_run.main_test = mal_cb; - cur_test_run.eval_cb = mal_eval; + cur_test_run.init_peer = mal_init_peer; if (strstr (argv[0], "_1") != NULL) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test malicious peer type 1\n"); + cur_test_run.name = "test-rps-malicious_1"; mal_type = 1; } else if (strstr (argv[0], "_2") != NULL) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test malicious peer type 2\n"); + cur_test_run.name = "test-rps-malicious_2"; mal_type = 2; } else if (strstr (argv[0], "_3") != NULL) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test malicious peer type 3\n"); + cur_test_run.name = "test-rps-malicious_3"; mal_type = 3; } } @@ -938,61 +2047,148 @@ main (int argc, char *argv[]) else if (strstr (argv[0], "_single_req") != NULL) { GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Test single request\n"); + cur_test_run.name = "test-rps-single-req"; cur_test_run.main_test = single_req_cb; + cur_test_run.have_churn = HAVE_NO_CHURN; } + else if (strstr (argv[0], "_delayed_reqs") != NULL) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test delayed requests\n"); + cur_test_run.name = "test-rps-delayed-reqs"; cur_test_run.main_test = delay_req_cb; + cur_test_run.have_churn = HAVE_NO_CHURN; } + else if (strstr (argv[0], "_seed_big") != NULL) { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test seeding (num_peers > GNUNET_SERVER_MAX_MESSAGE_SIZE)\n"); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test seeding (num_peers > GNUNET_MAX_MESSAGE_SIZE)\n"); + num_peers = 1; + cur_test_run.name = "test-rps-seed-big"; cur_test_run.main_test = seed_big_cb; + cur_test_run.eval_cb = no_eval; + cur_test_run.have_churn = HAVE_NO_CHURN; + timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10); } + else if (strstr (argv[0], "_single_peer_seed") != NULL) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test seeding and requesting on a single peer\n"); + cur_test_run.name = "test-rps-single-peer-seed"; cur_test_run.main_test = single_peer_seed_cb; + cur_test_run.have_churn = HAVE_NO_CHURN; } + else if (strstr (argv[0], "_seed_request") != NULL) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test seeding and requesting on multiple peers\n"); + cur_test_run.name = "test-rps-seed-request"; cur_test_run.main_test = seed_req_cb; + cur_test_run.have_churn = HAVE_NO_CHURN; } + else if (strstr (argv[0], "_seed") != NULL) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test seeding\n"); + cur_test_run.name = "test-rps-seed"; cur_test_run.main_test = seed_cb; - cur_test_run.eval_cb = seed_eval; + cur_test_run.eval_cb = no_eval; + cur_test_run.have_churn = HAVE_NO_CHURN; } + else if (strstr (argv[0], "_req_cancel") != NULL) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test cancelling a request\n"); + cur_test_run.name = "test-rps-req-cancel"; + num_peers = 1; cur_test_run.main_test = req_cancel_cb; + cur_test_run.eval_cb = no_eval; + cur_test_run.have_churn = HAVE_NO_CHURN; + timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10); } + + else if (strstr (argv[0], "_churn") != NULL) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test churn\n"); + cur_test_run.name = "test-rps-churn"; + num_peers = 5; + cur_test_run.init_peer = default_init_peer; + cur_test_run.main_test = churn_test_cb; + cur_test_run.reply_handle = default_reply_handle; + cur_test_run.eval_cb = default_eval_cb; + cur_test_run.have_churn = HAVE_CHURN; + cur_test_run.have_quick_quit = HAVE_NO_QUICK_QUIT; + timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10); + } + else if (strstr (argv[0], "profiler") != NULL) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "This is the profiler\n"); + cur_test_run.name = "test-rps-profiler"; + num_peers = 10; mal_type = 3; - cur_test_run.pre_test = profiler_pre; + cur_test_run.init_peer = profiler_init_peer; + cur_test_run.pre_test = mal_pre; cur_test_run.main_test = profiler_cb; - churn_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, - 10), - churn, NULL); + cur_test_run.reply_handle = profiler_reply_handle; + cur_test_run.eval_cb = profiler_eval; + cur_test_run.post_test = post_profiler; + cur_test_run.request_interval = 2; + cur_test_run.num_requests = 5; + cur_test_run.have_churn = HAVE_CHURN; + cur_test_run.have_quick_quit = HAVE_NO_QUICK_QUIT; + cur_test_run.have_collect_statistics = COLLECT_STATISTICS; + cur_test_run.stat_collect_flags = STAT_TYPE_ROUNDS | + STAT_TYPE_BLOCKS | + STAT_TYPE_BLOCKS_MANY_PUSH | + STAT_TYPE_BLOCKS_FEW_PUSH | + STAT_TYPE_BLOCKS_FEW_PULL | + STAT_TYPE_ISSUED_PUSH_SEND | + STAT_TYPE_ISSUED_PULL_REQ | + STAT_TYPE_ISSUED_PULL_REP | + STAT_TYPE_SENT_PUSH_SEND | + STAT_TYPE_SENT_PULL_REQ | + STAT_TYPE_SENT_PULL_REP | + STAT_TYPE_RECV_PUSH_SEND | + STAT_TYPE_RECV_PULL_REQ | + STAT_TYPE_RECV_PULL_REP; + timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 300); + + /* 'Clean' directory */ + (void) GNUNET_DISK_directory_remove ("/tmp/rps/"); + GNUNET_DISK_directory_create ("/tmp/rps/"); } - ok = 1; - (void) GNUNET_TESTBED_test_run ("test-rps-multipeer", - "test_rps.conf", - NUM_PEERS, - 0, NULL, NULL, - &run, NULL); + rps_peers = GNUNET_new_array (num_peers, struct RPSPeer); + peer_map = GNUNET_CONTAINER_multipeermap_create (num_peers, GNUNET_NO); + rps_peer_ids = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity); + if ( (2 == mal_type) || + (3 == mal_type)) + target_peer = &rps_peer_ids[num_peers - 2]; + if (profiler_eval == cur_test_run.eval_cb) + eval_peer = &rps_peers[num_peers - 1]; /* FIXME: eval_peer could be a + malicious peer if not careful + with the malicious portion */ - if (NULL != churn_task) - GNUNET_SCHEDULER_cancel (churn_task); + ok = 1; + ret_value = GNUNET_TESTBED_test_run (cur_test_run.name, + "test_rps.conf", + num_peers, + 0, NULL, NULL, + &run, NULL); + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "_test_run returned.\n"); + if (GNUNET_OK != ret_value) + { + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "Test did not run successfully!\n"); + } - return cur_test_run.eval_cb(); + ret_value = cur_test_run.eval_cb(); + GNUNET_free (rps_peers); + GNUNET_free (rps_peer_ids); + GNUNET_CONTAINER_multipeermap_destroy (peer_map); + return ret_value; } -/* end of test_rps_multipeer.c */ +/* end of test_rps.c */