-
-/* Copyright (c) 2010. The SimGrid Team.
+/* Copyright (c) 2010-2015. The SimGrid Team.
* All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
#include <stdio.h>
-#include "msg/msg.h"
+#include "simgrid/msg.h"
#include "xbt/log.h"
#include "xbt/asserts.h"
#include "simgrid/modelchecker.h"
+#include <xbt/RngStream.h>
+#include "src/mc/mc_replay.h" // FIXME: this is an internal header
/** @addtogroup MSG_examples
*
* - <b>chord/chord.c: Classical Chord P2P protocol</b>
* This example implements the well known Chord P2P protocol. Its
* main advantage is that it constitute a fully working non-trivial
- * example. In addition, its implementation is rather efficient, as
+ * example. In addition, its implementation is rather efficient, as
* demonstrated in http://hal.inria.fr/inria-00602216/
*/
-
- XBT_LOG_NEW_DEFAULT_CATEGORY(msg_chord,
+
+XBT_LOG_NEW_DEFAULT_CATEGORY(msg_chord,
"Messages specific for this msg example");
#define COMM_SIZE 10
static int periodic_check_predecessor_delay = 120;
static int periodic_lookup_delay = 10;
+static const double sleep_delay = 4.9999;
+
extern long int smx_total_comms;
/*
int next_finger_to_fix; // index of the next finger to fix in fix_fingers()
msg_comm_t comm_receive; // current communication to receive
double last_change_date; // last time I changed a finger or my predecessor
+ RngStream stream; //RngStream for
} s_node_t, *node_t;
/**
TASK_GET_PREDECESSOR_ANSWER,
TASK_NOTIFY,
TASK_SUCCESSOR_LEAVING,
- TASK_PREDECESSOR_LEAVING
+ TASK_PREDECESSOR_LEAVING,
+ TASK_PREDECESSOR_ALIVE,
+ TASK_PREDECESSOR_ALIVE_ANSWER
} e_task_type_t;
/*
} s_task_data_t, *task_data_t;
static int *powers2;
+static xbt_dynar_t host_list;
// utility functions
static void chord_initialize(void);
// compute the powers of 2 once for all
powers2 = xbt_new(int, nb_bits);
int pow = 1;
- int i;
+ unsigned i;
for (i = 0; i < nb_bits; i++) {
powers2[i] = pow;
pow = pow << 1;
}
nb_keys = pow;
XBT_DEBUG("Sets nb_keys to %d", nb_keys);
+
+ msg_host_t host;
+ host_list = MSG_hosts_as_dynar();
+ xbt_dynar_foreach(host_list, i, host) {
+ char descr[512];
+ RngStream stream;
+ snprintf(descr, sizeof descr, "RngSream<%s>", MSG_host_get_name(host));
+ stream = RngStream_CreateStream(descr);
+ MSG_host_set_property_value(host, "stream", (char*)stream, NULL);
+ }
}
static void chord_exit(void)
{
+ msg_host_t host;
+ unsigned i;
+ xbt_dynar_foreach(host_list, i, host) {
+ RngStream stream = (RngStream)MSG_host_get_property_value(host, "stream");
+ RngStream_DeleteStream(&stream);
+ }
+ xbt_dynar_free(&host_list);
+
xbt_free(powers2);
}
static void task_free(void* task)
{
// TODO add a parameter data_free_function to MSG_task_create?
- xbt_free(MSG_task_get_data(task));
- MSG_task_destroy(task);
+ if(task != NULL){
+ xbt_free(MSG_task_get_data(task));
+ MSG_task_destroy(task);
+ }
}
/**
if (XBT_LOG_ISENABLED(msg_chord, xbt_log_priority_verbose)) {
int i;
XBT_VERB("My finger table:");
- XBT_VERB("Start | Succ ");
+ XBT_VERB("Start | Succ");
for (i = 0; i < nb_bits; i++) {
- XBT_VERB(" %3d | %3d ", (node->id + powers2[i]) % nb_keys, node->fingers[i].id);
+ XBT_VERB(" %3d | %3d", (node->id + powers2[i]) % nb_keys, node->fingers[i].id);
}
XBT_VERB("Predecessor: %d", node->pred_id);
}
*/
int node(int argc, char *argv[])
{
+
/* Reduce the run size for the MC */
- if(MC_IS_ENABLED){
+ if(MC_is_active() || MC_record_replay_is_active()){
periodic_stabilize_delay = 8;
periodic_fix_fingers_delay = 8;
periodic_check_predecessor_delay = 8;
double next_check_predecessor_date = init_time + periodic_check_predecessor_delay;
double next_lookup_date = init_time + periodic_lookup_delay;
+ int listen = 0;
+ int no_op = 0;
+ int sub_protocol = 0;
+
xbt_assert(argc == 3 || argc == 5, "Wrong number of arguments for this node");
// initialize my node
s_node_t node = {0};
- node.id = atoi(argv[1]);
+ node.id = xbt_str_parse_int(argv[1],"Invalid ID: %s");
+ node.stream = (RngStream)MSG_host_get_property_value(MSG_host_self(), "stream");
get_mailbox(node.id, node.mailbox);
node.next_finger_to_fix = 0;
node.fingers = xbt_new0(s_finger_t, nb_bits);
}
if (argc == 3) { // first ring
- deadline = atof(argv[2]);
+ deadline = xbt_str_parse_double(argv[2],"Invalid deadline: %s");
create(&node);
join_success = 1;
- }
- else {
- int known_id = atoi(argv[2]);
+
+ } else {
+ int known_id = xbt_str_parse_int(argv[2],"Invalid root ID: %s");
//double sleep_time = atof(argv[3]);
- deadline = atof(argv[4]);
+ deadline = xbt_str_parse_double(argv[4],"Invalid deadline: %s");
/*
// sleep before starting
// FIXME: do not make MSG_task_irecv() calls from several functions
}
+ //XBT_INFO("Node %d is ring member : %d", node.id, is_ring_member(known_id, node.id) != -1);
+
if (!MSG_comm_test(node.comm_receive)) {
// no task was received: make some periodic calls
- if (MSG_get_clock() >= next_stabilize_date) {
- stabilize(&node);
- next_stabilize_date = MSG_get_clock() + periodic_stabilize_delay;
- }
- else if (MSG_get_clock() >= next_fix_fingers_date) {
- fix_fingers(&node);
- next_fix_fingers_date = MSG_get_clock() + periodic_fix_fingers_delay;
- }
- else if (MSG_get_clock() >= next_check_predecessor_date) {
- check_predecessor(&node);
- next_check_predecessor_date = MSG_get_clock() + periodic_check_predecessor_delay;
- }
- else if (MSG_get_clock() >= next_lookup_date) {
- random_lookup(&node);
- next_lookup_date = MSG_get_clock() + periodic_lookup_delay;
- }
- else {
- // nothing to do: sleep for a while
- MSG_process_sleep(5);
- }
- }
- if (node.comm_receive && MSG_comm_test(node.comm_receive)) {
+ if(MC_is_active() || MC_record_replay_is_active()){
+ if(MC_is_active() && !MC_visited_reduction() && no_op){
+ MC_cut();
+ }
+ if(listen == 0 && (sub_protocol = MC_random(0, 4)) > 0){
+ if(sub_protocol == 1)
+ stabilize(&node);
+ else if(sub_protocol == 2)
+ fix_fingers(&node);
+ else if(sub_protocol == 3)
+ check_predecessor(&node);
+ else
+ random_lookup(&node);
+ listen = 1;
+ }else{
+ MSG_process_sleep(sleep_delay);
+ if(!MC_visited_reduction())
+ no_op = 1;
+ }
+ }else{
+ if (MSG_get_clock() >= next_stabilize_date) {
+ stabilize(&node);
+ next_stabilize_date = MSG_get_clock() + periodic_stabilize_delay;
+ }else if (MSG_get_clock() >= next_fix_fingers_date) {
+ fix_fingers(&node);
+ next_fix_fingers_date = MSG_get_clock() + periodic_fix_fingers_delay;
+ }else if (MSG_get_clock() >= next_check_predecessor_date) {
+ check_predecessor(&node);
+ next_check_predecessor_date = MSG_get_clock() + periodic_check_predecessor_delay;
+ }else if (MSG_get_clock() >= next_lookup_date) {
+ random_lookup(&node);
+ next_lookup_date = MSG_get_clock() + periodic_lookup_delay;
+ }else {
+ // nothing to do: sleep for a while
+ MSG_process_sleep(sleep_delay);
+ }
+ }
- // a transfer has occured
+ } else {
+ // a transfer has occurred
msg_error_t status = MSG_comm_get_status(node.comm_receive);
}
if (node.comm_receive) {
+ /* handle last task if any */
+ if (MSG_comm_wait(node.comm_receive, 0) == MSG_OK)
+ task_free(task_received);
MSG_comm_destroy(node.comm_receive);
node.comm_receive = NULL;
}
switch (type) {
- case TASK_FIND_SUCCESSOR:
- XBT_DEBUG("Receiving a 'Find Successor' request from %s for id %d",
- task_data->issuer_host_name, task_data->request_id);
- // is my successor the successor?
- if (is_in_interval(task_data->request_id, node->id + 1, node->fingers[0].id)) {
- task_data->type = TASK_FIND_SUCCESSOR_ANSWER;
- task_data->answer_id = node->fingers[0].id;
- XBT_DEBUG("Sending back a 'Find Successor Answer' to %s (mailbox %s): the successor of %d is %d",
- task_data->issuer_host_name,
- task_data->answer_to,
- task_data->request_id, task_data->answer_id);
- MSG_task_dsend(task, task_data->answer_to, task_free);
- }
- else {
- // otherwise, forward the request to the closest preceding finger in my table
- int closest = closest_preceding_node(node, task_data->request_id);
- XBT_DEBUG("Forwarding the 'Find Successor' request for id %d to my closest preceding finger %d",
- task_data->request_id, closest);
- get_mailbox(closest, mailbox);
- MSG_task_dsend(task, mailbox, task_free);
- }
- break;
-
- case TASK_GET_PREDECESSOR:
- XBT_DEBUG("Receiving a 'Get Predecessor' request from %s", task_data->issuer_host_name);
- task_data->type = TASK_GET_PREDECESSOR_ANSWER;
- task_data->answer_id = node->pred_id;
- XBT_DEBUG("Sending back a 'Get Predecessor Answer' to %s via mailbox '%s': my predecessor is %d",
- task_data->issuer_host_name,
- task_data->answer_to, task_data->answer_id);
+ case TASK_FIND_SUCCESSOR:
+ XBT_DEBUG("Receiving a 'Find Successor' request from %s for id %d",
+ task_data->issuer_host_name, task_data->request_id);
+ // is my successor the successor?
+ if (is_in_interval(task_data->request_id, node->id + 1, node->fingers[0].id)) {
+ task_data->type = TASK_FIND_SUCCESSOR_ANSWER;
+ task_data->answer_id = node->fingers[0].id;
+ XBT_DEBUG("Sending back a 'Find Successor Answer' to %s (mailbox %s): the successor of %d is %d",
+ task_data->issuer_host_name,
+ task_data->answer_to,
+ task_data->request_id, task_data->answer_id);
MSG_task_dsend(task, task_data->answer_to, task_free);
- break;
-
- case TASK_NOTIFY:
- // someone is telling me that he may be my new predecessor
- XBT_DEBUG("Receiving a 'Notify' request from %s", task_data->issuer_host_name);
- notify(node, task_data->request_id);
- task_free(task);
- break;
-
- case TASK_PREDECESSOR_LEAVING:
- // my predecessor is about to quit
- XBT_DEBUG("Receiving a 'Predecessor Leaving' message from %s", task_data->issuer_host_name);
- // modify my predecessor
- set_predecessor(node, task_data->request_id);
- task_free(task);
- /*TODO :
+ }
+ else {
+ // otherwise, forward the request to the closest preceding finger in my table
+ int closest = closest_preceding_node(node, task_data->request_id);
+ XBT_DEBUG("Forwarding the 'Find Successor' request for id %d to my closest preceding finger %d",
+ task_data->request_id, closest);
+ get_mailbox(closest, mailbox);
+ MSG_task_dsend(task, mailbox, task_free);
+ }
+ break;
+
+ case TASK_GET_PREDECESSOR:
+ XBT_DEBUG("Receiving a 'Get Predecessor' request from %s", task_data->issuer_host_name);
+ task_data->type = TASK_GET_PREDECESSOR_ANSWER;
+ task_data->answer_id = node->pred_id;
+ XBT_DEBUG("Sending back a 'Get Predecessor Answer' to %s via mailbox '%s': my predecessor is %d",
+ task_data->issuer_host_name,
+ task_data->answer_to, task_data->answer_id);
+ MSG_task_dsend(task, task_data->answer_to, task_free);
+ break;
+
+ case TASK_NOTIFY:
+ // someone is telling me that he may be my new predecessor
+ XBT_DEBUG("Receiving a 'Notify' request from %s", task_data->issuer_host_name);
+ notify(node, task_data->request_id);
+ task_free(task);
+ break;
+
+ case TASK_PREDECESSOR_LEAVING:
+ // my predecessor is about to quit
+ XBT_DEBUG("Receiving a 'Predecessor Leaving' message from %s", task_data->issuer_host_name);
+ // modify my predecessor
+ set_predecessor(node, task_data->request_id);
+ task_free(task);
+ /*TODO :
>> notify my new predecessor
>> send a notify_predecessors !!
- */
- break;
-
- case TASK_SUCCESSOR_LEAVING:
- // my successor is about to quit
- XBT_DEBUG("Receiving a 'Successor Leaving' message from %s", task_data->issuer_host_name);
- // modify my successor FIXME : this should be implicit ?
- set_finger(node, 0, task_data->request_id);
- task_free(task);
- /* TODO
- >> notify my new successor
- >> update my table & predecessors table */
- break;
-
- case TASK_FIND_SUCCESSOR_ANSWER:
- case TASK_GET_PREDECESSOR_ANSWER:
- XBT_DEBUG("Ignoring unexpected task of type %d (%p)", (int)type, task);
- task_free(task);
- break;
+ */
+ break;
+
+ case TASK_SUCCESSOR_LEAVING:
+ // my successor is about to quit
+ XBT_DEBUG("Receiving a 'Successor Leaving' message from %s", task_data->issuer_host_name);
+ // modify my successor FIXME : this should be implicit ?
+ set_finger(node, 0, task_data->request_id);
+ task_free(task);
+ /* TODO
+ >> notify my new successor
+ >> update my table & predecessors table */
+ break;
+
+ case TASK_FIND_SUCCESSOR_ANSWER:
+ case TASK_GET_PREDECESSOR_ANSWER:
+ case TASK_PREDECESSOR_ALIVE_ANSWER:
+ XBT_DEBUG("Ignoring unexpected task of type %d (%p)", (int)type, task);
+ task_free(task);
+ break;
+
+ case TASK_PREDECESSOR_ALIVE:
+ XBT_DEBUG("Receiving a 'Predecessor Alive' request from %s", task_data->issuer_host_name);
+ task_data->type = TASK_PREDECESSOR_ALIVE_ANSWER;
+ XBT_DEBUG("Sending back a 'Predecessor Alive Answer' to %s (mailbox %s)",
+ task_data->issuer_host_name,
+ task_data->answer_to);
+ MSG_task_dsend(task, task_data->answer_to, task_free);
+ break;
+
+ default:
+ THROW_IMPOSSIBLE;
}
}
quit_notify(node);
}
-/*
+/**
* \brief Notifies the successor and the predecessor of the current node
* of the departure
* \param node the current node
msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
XBT_DEBUG("Sending a 'PREDECESSOR_LEAVING' to my successor %d",node->fingers[0].id);
- MSG_task_send_with_timeout(task_sent, node->fingers[0].mailbox, timeout);
+ if (MSG_task_send_with_timeout(task_sent, node->fingers[0].mailbox, timeout)==
+ MSG_TIMEOUT) {
+ XBT_DEBUG("Timeout expired when sending a 'PREDECESSOR_LEAVING' to my successor %d",
+ node->fingers[0].id);
+ task_free(task_sent);
+ }
//send the SUCCESSOR_LEAVING to our predecessor
get_mailbox(node->pred_id, mailbox);
msg_task_t task_sent_s = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data_s);
XBT_DEBUG("Sending a 'SUCCESSOR_LEAVING' to my predecessor %d",node->pred_id);
- MSG_task_send_with_timeout(task_sent_s, mailbox, timeout);
+ if (MSG_task_send_with_timeout(task_sent_s, mailbox, timeout)==
+ MSG_TIMEOUT) {
+ XBT_DEBUG("Timeout expired when sending a 'SUCCESSOR_LEAVING' to my predecessor %d",
+ node->pred_id);
+ task_free(task_sent_s);
+ }
}
XBT_DEBUG("Received a task (%p)", task_received);
task_data_t ans_data = MSG_task_get_data(task_received);
- if (MC_IS_ENABLED) {
- // the model-checker is expected to find a counter-example here.
- //
- // As you can see in the test right below, task_received is not always equal to task_sent
- // (as messages from differing round can interleave). But the previous version of this code
- // wrongly assumed that, leading to problems. But this only occured on large platforms,
- // leading to hardly usable traces. So we used the model-checker to track down the issue,
- // and we came down to this test, that explained the bug in a snap.
- MC_assert(task_received == task_sent);
- }
-
- if (task_received != task_sent) {
+ // Once upon a time, our code assumed that here, task_received != task_sent all the time
+ //
+ // This assumption is wrong (as messages from differing round can interleave), leading to a bug in our code.
+ // We failed to find this bug directly, as it only occured on large platforms, leading to hardly usable traces.
+ // Instead, we used the model-checker to track down the issue by adding the following test here in the code:
+ // if (MC_is_active()) {
+ // MC_assert(task_received == task_sent);
+ // }
+ // That explained the bug in a snap, with a very cool example and everything.
+ //
+ // This MC_assert is now desactivated as the case is now properly handled in our code and we don't want the
+ // MC to fail any further under that condition, but this comment is here to as a memorial for this first
+ // brillant victory of the model-checking in the SimGrid community :)
+
+ if (task_received != task_sent ||
+ ans_data->type != TASK_FIND_SUCCESSOR_ANSWER) {
// this is not the expected answer
MSG_comm_destroy(node->comm_receive);
node->comm_receive = NULL;
msg_task_t task_received = MSG_comm_get_task(node->comm_receive);
task_data_t ans_data = MSG_task_get_data(task_received);
- if (MC_IS_ENABLED) {
+ /*if (MC_is_active()) {
MC_assert(task_received == task_sent);
- }
+ }*/
- if (task_received != task_sent) {
+ if (task_received != task_sent ||
+ ans_data->type != TASK_GET_PREDECESSOR_ANSWER) {
MSG_comm_destroy(node->comm_receive);
node->comm_receive = NULL;
handle_task(node, task_received);
static void check_predecessor(node_t node)
{
XBT_DEBUG("Checking whether my predecessor is alive");
- // TODO
+
+ if(node->pred_id == -1)
+ return;
+
+ int stop = 0;
+
+ char mailbox[MAILBOX_NAME_SIZE];
+ get_mailbox(node->pred_id, mailbox);
+ task_data_t req_data = xbt_new0(s_task_data_t,1);
+ req_data->type = TASK_PREDECESSOR_ALIVE;
+ req_data->request_id = node->pred_id;
+ get_mailbox(node->id, req_data->answer_to);
+ req_data->issuer_host_name = MSG_host_get_name(MSG_host_self());
+
+ msg_task_t task_sent = MSG_task_create(NULL, COMP_SIZE, COMM_SIZE, req_data);
+ XBT_DEBUG("Sending a 'Predecessor Alive' request to my predecessor %d", node->pred_id);
+
+ msg_error_t res = MSG_task_send_with_timeout(task_sent, mailbox, timeout);
+
+ if (res != MSG_OK) {
+ XBT_DEBUG("Failed to send the 'Predecessor Alive' request (task %p) to %d", task_sent, node->pred_id);
+ task_free(task_sent);
+ }else{
+
+ // receive the answer
+ XBT_DEBUG("Sent 'Predecessor Alive' request (task %p) to %d, waiting for the answer on my mailbox '%s'",
+ task_sent, node->pred_id, req_data->answer_to);
+
+ do {
+ if (node->comm_receive == NULL) { // FIXME simplify this
+ msg_task_t task_received = NULL;
+ node->comm_receive = MSG_task_irecv(&task_received, node->mailbox);
+ }
+
+ res = MSG_comm_wait(node->comm_receive, timeout);
+
+ if (res != MSG_OK) {
+ XBT_DEBUG("Failed to receive the answer to my 'Predecessor Alive' request (task %p): %d",
+ task_sent, (int)res);
+ stop = 1;
+ MSG_comm_destroy(node->comm_receive);
+ node->comm_receive = NULL;
+ node->pred_id = -1;
+ }else {
+ msg_task_t task_received = MSG_comm_get_task(node->comm_receive);
+ if (task_received != task_sent) {
+ MSG_comm_destroy(node->comm_receive);
+ node->comm_receive = NULL;
+ handle_task(node, task_received);
+ }else{
+ XBT_DEBUG("Received the answer to my 'Predecessor Alive' request (task %p) : my predecessor %d is alive", task_received, node->pred_id);
+ stop = 1;
+ MSG_comm_destroy(node->comm_receive);
+ node->comm_receive = NULL;
+ task_free(task_received);
+ }
+ }
+ } while (!stop);
+ }
}
/**
*/
static void random_lookup(node_t node)
{
- int id = 1337; // TODO pick a pseudorandom id
- XBT_DEBUG("Making a lookup request for id %d", id);
- find_successor(node, id);
+ int random_index = RngStream_RandInt (node->stream, 0, nb_bits - 1);
+ int random_id = node->fingers[random_index].id;
+ XBT_DEBUG("Making a lookup request for id %d", random_id);
+ int res = find_successor(node, random_id);
+ XBT_DEBUG("The successor of node %d is %d", random_id, res);
+
}
/**
int main(int argc, char *argv[])
{
MSG_init(&argc, argv);
- if (argc < 3) {
- printf("Usage: %s [-nb_bits=n] [-timeout=t] platform_file deployment_file\n", argv[0]);
- printf("example: %s ../msg_platform.xml chord.xml\n", argv[0]);
- exit(1);
- }
+ xbt_assert(argc > 2,
+ "Usage: %s [-nb_bits=n] [-timeout=t] platform_file deployment_file\n"
+ "\tExample: %s ../msg_platform.xml chord.xml\n",
+ argv[0], argv[0]);
char **options = &argv[1];
while (!strncmp(options[0], "-", 1)) {
int length = strlen("-nb_bits=");
if (!strncmp(options[0], "-nb_bits=", length) && strlen(options[0]) > length) {
- nb_bits = atoi(options[0] + length);
+ nb_bits = xbt_str_parse_int(options[0] + length, "Invalid nb_bits parameter: %s");
XBT_DEBUG("Set nb_bits to %d", nb_bits);
}
else {
length = strlen("-timeout=");
if (!strncmp(options[0], "-timeout=", length) && strlen(options[0]) > length) {
- timeout = atoi(options[0] + length);
+ timeout = xbt_str_parse_int(options[0] + length, "Invalid timeout parameter: %s");
XBT_DEBUG("Set timeout to %d", timeout);
}
else {
const char* platform_file = options[0];
const char* application_file = options[1];
- chord_initialize();
-
MSG_create_environment(platform_file);
+ chord_initialize();
+
MSG_function_register("node", node);
MSG_launch_application(application_file);
XBT_CRITICAL("Messages created: %ld", smx_total_comms);
XBT_INFO("Simulated time: %g", MSG_get_clock());
- MSG_clean();
chord_exit();
- if (res == MSG_OK)
- return 0;
- else
- return 1;
+ return res != MSG_OK;
}
