X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/f8c559b682d138281eedddfbcb363a2b051c4d13..69d66050bef9b3b9ec134c84f71f303194d5b13e:/examples/msg/kadeploy/kadeploy.c

diff --git a/examples/msg/kadeploy/kadeploy.c b/examples/msg/kadeploy/kadeploy.c
index 4b27de36ef..93e69987fc 100644
--- a/examples/msg/kadeploy/kadeploy.c
+++ b/examples/msg/kadeploy/kadeploy.c
@@ -5,7 +5,8 @@
 /* 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 <stdio.h>
+#include <stdlib.h>
 
 #include "msg/msg.h"            /* Yeah! If you want to use msg, you need to include msg/msg.h */
 #include "xbt/sysdep.h"         /* calloc */
@@ -23,6 +24,12 @@
 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_kadeploy,
                              "Messages specific for kadeploy");
 
+#define MESSAGE_SIZE 1
+#define PIECE_COUNT 100
+#define HOSTNAME_LENGTH 20
+
+#define PEER_SHUTDOWN_DEADLINE 600
+
 /*
  Data structures
  */
@@ -31,43 +38,87 @@ XBT_LOG_NEW_DEFAULT_CATEGORY(msg_kadeploy,
 typedef struct xbt_dynar_iterator_struct {
   xbt_dynar_t list;
   xbt_dynar_t indices_list;
-  unsigned int current;
+  int current;
   unsigned long length;
-  unsigned int (*criteria_fn)(void* it);
+  int (*criteria_fn)(void* it);
 } *xbt_dynar_iterator_t;
 typedef struct xbt_dynar_iterator_struct xbt_dynar_iterator_s;
 
-
-xbt_dynar_iterator_t xbt_dynar_iterator_new(xbt_dynar_t list, unsigned int (*criteria_fn)(void*));
+/* Messages enum */
+typedef enum {
+  MESSAGE_BUILD_CHAIN = 0,
+  MESSAGE_SEND_DATA,
+  MESSAGE_END_DATA
+} e_message_type;
+
+/* Message struct */
+typedef struct s_message {
+  e_message_type type;
+  const char *issuer_hostname;
+  const char *mailbox;
+  const char *prev_hostname;
+  const char *next_hostname;
+  const char *data_block;
+  unsigned int data_length;
+} s_message_t, *message_t;
+
+/* Peer struct */
+typedef struct s_peer {
+  int init;
+  const char *prev;
+  const char *next;
+  const char *me;
+  int pieces;
+  xbt_dynar_t pending_sends;
+  int close_asap; /* TODO: unused */
+} s_peer_t, *peer_t;
+
+/* Iterator methods */
+xbt_dynar_iterator_t xbt_dynar_iterator_new(xbt_dynar_t list, int (*criteria_fn)(void*));
 void *xbt_dynar_iterator_next(xbt_dynar_iterator_t it);
 void xbt_dynar_iterator_delete(xbt_dynar_iterator_t it);
-unsigned int xbt_dynar_iterator_forward_criteria(void *p);
+int xbt_dynar_iterator_forward_criteria(void *p);
+
+/* Message methods */
+msg_task_t task_message_new(e_message_type type, const char *issuer_hostname, const char *mailbox);
+msg_task_t task_message_chain_new(const char *issuer_hostname, const char *mailbox, const char* prev, const char *next);
+msg_task_t task_message_data_new(const char *issuer_hostname, const char *mailbox, const char *block, unsigned int len);
+msg_task_t task_message_end_data_new(const char *issuer_hostname, const char *mailbox);
+void task_message_delete(void *);
 
+/* Tasks */
 int broadcaster(int argc, char *argv[]);
 int peer(int argc, char *argv[]);
 
-void check_hosts(const int count, char **list);
-xbt_dynar_t build_hostlist_from_argv(int argc, char *argv[]);
-void build_chain(xbt_dynar_t host_list);
+xbt_dynar_t build_hostlist_from_hostcount(int hostcount); 
+/*xbt_dynar_t build_hostlist_from_argv(int argc, char *argv[]);*/
 
-int peer_wait_for_init();
+/* Broadcaster: helper functions */
+int broadcaster_build_chain(const char **first, xbt_dynar_t host_list);
+int broadcaster_send_file(const char *first);
+int broadcaster_finish(xbt_dynar_t host_list);
 
+/* Peer: helper functions */
+msg_error_t peer_wait_for_message(peer_t peer);
+int peer_execute_task(peer_t peer, msg_task_t task);
+void peer_init_chain(peer_t peer, message_t msg);
+void peer_shutdown(peer_t p);
+void peer_init(peer_t p);
+
+/* Initialization stuff */
 msg_error_t test_all(const char *platform_file,
                      const char *application_file);
 
-double task_comm_size_lat = 10e0;
-double task_comm_size_bw = 10e8;
-
 /* Allocates and initializes a new xbt_dynar iterator for list, using criteria_fn as iteration criteria
    criteria_fn: given an iterator, it must update the iterator and give the next element's index, 
    less than 0 otherwise*/
-xbt_dynar_iterator_t xbt_dynar_iterator_new(xbt_dynar_t list, unsigned int (*criteria_fn)(void*))
+xbt_dynar_iterator_t xbt_dynar_iterator_new(xbt_dynar_t list, int (*criteria_fn)(void*))
 {
   xbt_dynar_iterator_t it = xbt_new(xbt_dynar_iterator_s, 1);
   
   it->list = list;
   it->length = xbt_dynar_length(list);
-  it->indices_list = xbt_dynar_new(sizeof(unsigned int), NULL);
+  it->indices_list = xbt_dynar_new(sizeof(int), NULL);
   it->criteria_fn = criteria_fn;
   it->current = -1;
 }
@@ -75,11 +126,12 @@ xbt_dynar_iterator_t xbt_dynar_iterator_new(xbt_dynar_t list, unsigned int (*cri
 /* Returns the next element iterated by iterator it, NULL if there are no more elements */
 void *xbt_dynar_iterator_next(xbt_dynar_iterator_t it)
 {
-  unsigned int next = it->criteria_fn((xbt_dynar_iterator_t)it);
-  XBT_INFO("%d current\n", next);
-  if (next < 0)
+  int next = it->criteria_fn((xbt_dynar_iterator_t)it);
+  //XBT_INFO("%d current\n", next);
+  if (next < 0) {
+    //XBT_INFO("Nothing to return!\n");
     return NULL;
-  else {
+  } else {
     xbt_dynar_push(it->indices_list, &next);
     return xbt_dynar_get_ptr(it->list, next);
   }
@@ -91,10 +143,10 @@ void xbt_dynar_iterator_delete(xbt_dynar_iterator_t it)
   xbt_free_ref(&it);
 }
 
-unsigned int xbt_dynar_iterator_forward_criteria(void *p)
+int xbt_dynar_iterator_forward_criteria(void *p)
 {
   xbt_dynar_iterator_t it = (xbt_dynar_iterator_t)p;
-  unsigned int r = -1;
+  int r = -1;
   if (it->current == -1) {
     /* iterator initialization */
     it->current = 0;
@@ -107,7 +159,98 @@ unsigned int xbt_dynar_iterator_forward_criteria(void *p)
   return r;
 }
 
-xbt_dynar_t build_hostlist_from_argv(int argc, char *argv[])
+msg_task_t task_message_new(e_message_type type, const char *issuer_hostname, const char *mailbox)
+{
+  message_t msg = xbt_new(s_message_t, 1);
+  msg->type = type;
+  msg->issuer_hostname = issuer_hostname;
+  msg->mailbox = mailbox;
+  msg_task_t task = MSG_task_create(NULL, 0, MESSAGE_SIZE, msg); 
+
+  return task;
+}
+
+msg_task_t task_message_chain_new(const char *issuer_hostname, const char *mailbox, const char* prev, const char *next)
+{
+  msg_task_t task = task_message_new(MESSAGE_BUILD_CHAIN, issuer_hostname, mailbox);
+  message_t msg = MSG_task_get_data(task);
+  msg->prev_hostname = prev;
+  msg->next_hostname = next;
+
+  return task;
+}
+
+msg_task_t task_message_data_new(const char *issuer_hostname, const char *mailbox, const char *block, unsigned int len)
+{
+  msg_task_t task = task_message_new(MESSAGE_SEND_DATA, issuer_hostname, mailbox);
+  if (strcmp(mailbox, "host4") == 0) MSG_task_set_category(task, mailbox);
+  message_t msg = MSG_task_get_data(task);
+  msg->data_block = block;
+  msg->data_length = len;
+
+  return task;
+}
+
+msg_task_t task_message_end_data_new(const char *issuer_hostname, const char *mailbox)
+{
+  return task_message_new(MESSAGE_END_DATA, issuer_hostname, mailbox);
+}
+
+void task_message_delete(void *task)
+{
+  message_t msg = MSG_task_get_data(task);
+  xbt_free(msg);
+  MSG_task_destroy(task);
+}
+
+void queue_pending_connection(msg_comm_t comm, xbt_dynar_t q)
+{
+  xbt_dynar_push(q, &comm);
+}
+
+int process_pending_connections(xbt_dynar_t q)
+{
+  unsigned int iter;
+  int status;
+  int empty = 0;
+  msg_comm_t comm;
+
+  xbt_dynar_foreach(q, iter, comm) {
+    empty = 1;
+    if (MSG_comm_test(comm)) {
+      MSG_comm_destroy(comm);
+      status = MSG_comm_get_status(comm);
+      xbt_assert(status == MSG_OK, __FILE__ ": process_pending_connections() failed");
+      xbt_dynar_cursor_rm(q, &iter);
+      empty = 0;
+    }
+  }
+  return empty;
+}
+
+xbt_dynar_t build_hostlist_from_hostcount(int hostcount)
+{
+  xbt_dynar_t host_list = xbt_dynar_new(sizeof(char*), NULL);
+  char *hostname = NULL;
+  msg_host_t h = NULL;
+  int i = 1;
+  
+  for (; i < hostcount+1; i++) {
+    hostname = xbt_new(char, HOSTNAME_LENGTH);
+    snprintf(hostname, HOSTNAME_LENGTH, "host%d", i);
+    //XBT_INFO("%s", hostname);
+    h = MSG_get_host_by_name(hostname);
+    if (h == NULL) {
+      XBT_INFO("Unknown host %s. Stopping Now! ", hostname);
+      abort();
+    } else {
+      xbt_dynar_push(host_list, &hostname);
+    }
+  }
+  return host_list;
+}
+
+/*xbt_dynar_t build_hostlist_from_argv(int argc, char *argv[])
 {
   xbt_dynar_t host_list = xbt_dynar_new(sizeof(char*), NULL);
   msg_host_t h = NULL;
@@ -124,130 +267,246 @@ xbt_dynar_t build_hostlist_from_argv(int argc, char *argv[])
     }
   }
   return host_list;
-}
+}*/
 
 void delete_hostlist(xbt_dynar_t h)
 {
-  xbt_dynar_free_container(&h);
+  xbt_dynar_free(&h);
 }
 
-void build_chain(xbt_dynar_t host_list)
+int broadcaster_build_chain(const char **first, xbt_dynar_t host_list)
 {
   xbt_dynar_iterator_t it = xbt_dynar_iterator_new(host_list, xbt_dynar_iterator_forward_criteria);
-  char **cur = NULL;
+  msg_task_t task = NULL;
+  char **cur = (char**)xbt_dynar_iterator_next(it);
+  const char *me = MSG_host_get_name(MSG_host_self());
+  const char *current_host = NULL;
+  const char *prev = NULL;
+  const char *next = NULL;
+  const char *last = NULL;
+
+  /* Build the chain if there's at least one peer */
+  if (cur != NULL) {
+    /* init: prev=NULL, host=current cur, next=next cur */
+    next = *cur;
+    *first = next;
+
+    /* This iterator iterates one step ahead: cur is current iterated element, 
+       but it's actually the next one in the chain */
+    do {
+      /* following steps: prev=last, host=next, next=cur */
+      cur = (char**)xbt_dynar_iterator_next(it);
+      prev = last;
+      current_host = next;
+      if (cur != NULL)
+        next = *cur;
+      else
+        next = NULL;
+      //XBT_INFO("Building chain -- broadcaster:\"%s\" dest:\"%s\" prev:\"%s\" next:\"%s\"", me, current_host, prev, next);
+    
+      /* Send message to current peer */
+      task = task_message_chain_new(me, current_host, prev, next);
+      //MSG_task_set_category(task, current_host);
+      MSG_task_send(task, current_host);
 
-  for (cur = (char**)xbt_dynar_iterator_next(it); cur != NULL; cur = (char**)xbt_dynar_iterator_next(it)) {
-    XBT_INFO("iterating host = %s", *cur);
+      last = current_host;
+    } while (cur != NULL);
   }
+  xbt_dynar_iterator_delete(it);
+
+  return MSG_OK;
 }
 
-/*void setup_chain_criteria(chain_criteria_t c, char *(*fn)(void))
+int broadcaster_send_file(const char *first)
 {
-  
+  const char *me = MSG_host_get_name(MSG_host_self());
+  msg_task_t task = NULL;
+  msg_comm_t comm = NULL;
+  int status;
+
+  int piece_count = PIECE_COUNT;
+  int cur = 0;
+
+  for (; cur < piece_count; cur++) {
+    /* TODO: stub */
+    task = task_message_data_new(me, first, NULL, 0);
+    //XBT_INFO("Sending (isend) from %s into mailbox %s", me, first);
+    //comm = MSG_task_isend(task, first);
+    status = MSG_task_send(task, first);
+    //MSG_task_dsend(task, first, task_message_delete);
+   
+    //status = MSG_comm_wait(comm, -1);
+    xbt_assert(status == MSG_OK, __FILE__ ": broadcaster_send_file() failed");
+    //MSG_comm_destroy(comm);
+  }
+
+  return MSG_OK;
 }
 
-void build_chain(const int hostcount, char **hostlist)
+int broadcaster_finish(xbt_dynar_t host_list)
 {
-  int i;
-  for (i = 0; i < hostcount; i++) {
-    
+  xbt_dynar_iterator_t it = xbt_dynar_iterator_new(host_list, xbt_dynar_iterator_forward_criteria);
+  msg_task_t task = NULL;
+  const char *me = MSG_host_get_name(MSG_host_self());
+  const char *current_host = NULL;
+  char **cur = NULL;
+
+  /* Send goodbye message to every peer */
+  for (cur = (char**)xbt_dynar_iterator_next(it); cur != NULL; cur = (char**)xbt_dynar_iterator_next(it)) {
+    /* Send message to current peer */
+    current_host = *cur;
+    task = task_message_end_data_new(me, current_host);
+    //MSG_task_set_category(task, current_host);
+    MSG_task_send(task, current_host);
   }
-}*/
+
+  return MSG_OK;
+}
+
 
 /** Emitter function  */
 int broadcaster(int argc, char *argv[])
 {
-  double time;
   xbt_dynar_t host_list = NULL;
-  msg_task_t task_la = NULL;
-  msg_task_t task_bw = NULL;
-  char sprintf_buffer_la[64];
-  char sprintf_buffer_bw[64];
+  const char *first = NULL;
+  int status = !MSG_OK;
 
   XBT_INFO("broadcaster");
 
-  /* Check that every host in the command line actually exists and add it to a dynamic array */
-  host_list = build_hostlist_from_argv(argc, argv);
+  /* Check that every host given by the hostcount in argv[1] exists and add it
+     to a dynamic array */
+  host_list = build_hostlist_from_hostcount(atoi(argv[1]));
+  /*host_list = build_hostlist_from_argv(argc, argv);*/
   
-  build_chain(host_list);
-
-  /* Latency */
-  /*time = MSG_get_clock();
-  sprintf(sprintf_buffer_la, "latency task");
-  task_la =
-      MSG_task_create(sprintf_buffer_la, 0.0, task_comm_size_lat, NULL);
-  task_la->data = xbt_new(double, 1);
-  *(double *) task_la->data = time;
-  XBT_INFO("task_la->data = %le", *((double *) task_la->data));
-  MSG_task_send(task_la, argv[1]);*/
-
-  /* Bandwidth */
-  /*time = MSG_get_clock();
-  sprintf(sprintf_buffer_bw, "bandwidth task");
-  task_bw =
-      MSG_task_create(sprintf_buffer_bw, 0.0, task_comm_size_bw, NULL);
-  task_bw->data = xbt_new(double, 1);
-  *(double *) task_bw->data = time;
-  XBT_INFO("task_bw->data = %le", *((double *) task_bw->data));
-  MSG_task_send(task_bw, argv[1]);
-  */
-  return 0;
-}                               /* end_of_client */
-
-int peer_wait_for_init()
+  /* TODO: Error checking */
+  status = broadcaster_build_chain(&first, host_list);
+  status = broadcaster_send_file(first);
+  status = broadcaster_finish(host_list);
+
+  delete_hostlist(host_list);
+
+  return status;
+}
+
+/*******************************************************
+ *                     Peer                            *
+ *******************************************************/
+
+void peer_init_chain(peer_t peer, message_t msg)
 {
-  return MSG_OK;
+  peer->prev = msg->prev_hostname;
+  peer->next = msg->next_hostname;
+  peer->init = 1;
 }
 
-/** Peer function  */
-int peer(int argc, char *argv[])
+void peer_forward_msg(peer_t peer, message_t msg)
 {
-  double time, time1, sender_time;
-  msg_task_t task_la = NULL;
-  msg_task_t task_bw = NULL;
-  int a;
-  double communication_time = 0;
+  int status;
+  msg_task_t task = task_message_data_new(peer->me, peer->next, NULL, 0);
+  msg_comm_t comm = NULL;
+  //XBT_INFO("Sending (isend) from %s into mailbox %s", peer->me, peer->next);
+  comm = MSG_task_isend(task, peer->next);
+  queue_pending_connection(comm, peer->pending_sends);
+}
 
-  XBT_INFO("peer");
+int peer_execute_task(peer_t peer, msg_task_t task)
+{
+  int done = 0;
+  message_t msg = MSG_task_get_data(task);
+  
+  //XBT_INFO("Peer %s got message of type %d\n", peer->me, msg->type);
+  switch (msg->type) {
+    case MESSAGE_BUILD_CHAIN:
+      peer_init_chain(peer, msg);
+      break;
+    case MESSAGE_SEND_DATA:
+      xbt_assert(peer->init, __FILE__ ": peer_execute_task() failed: got msg_type %d before initialization", msg->type);
+      if (peer->next != NULL)
+        peer_forward_msg(peer, msg);
+      peer->pieces++;
+      break;
+    case MESSAGE_END_DATA:
+      xbt_assert(peer->init, __FILE__ ": peer_execute_task() failed: got msg_type %d before initialization", msg->type);
+      done = 1;
+      XBT_INFO("%d pieces receieved", peer->pieces);
+      break;
+  }
 
-  time = MSG_get_clock();
-
-  a = peer_wait_for_init();
-  /* Get Latency */
-  /*a = MSG_task_receive(&task_la,MSG_host_get_name(MSG_host_self()));
-  if (a == MSG_OK) {
-    time1 = MSG_get_clock();
-    sender_time = *((double *) (task_la->data));
-    time = sender_time;
-    communication_time = time1 - time;
-    XBT_INFO("Task received : %s", task_la->name);
-    xbt_free(task_la->data);
-    MSG_task_destroy(task_la);
-    XBT_INFO("Communic. time %le", communication_time);
-    XBT_INFO("--- la %f ----", communication_time);
-  } else {
-    xbt_die("Unexpected behavior");
-  }*/
+  MSG_task_execute(task);
 
+  return done;
+}
 
-  /* Get Bandwidth */
-  /*a = MSG_task_receive(&task_bw,MSG_host_get_name(MSG_host_self()));
-  if (a == MSG_OK) {
-    time1 = MSG_get_clock();
-    sender_time = *((double *) (task_bw->data));
-    time = sender_time;
-    communication_time = time1 - time;
-    XBT_INFO("Task received : %s", task_bw->name);
-    xbt_free(task_bw->data);
-    MSG_task_destroy(task_bw);
-    XBT_INFO("Communic. time %le", communication_time);
-    XBT_INFO("--- bw %f ----", task_comm_size_bw / communication_time);
-  } else {
-    xbt_die("Unexpected behavior");
-  }*/
+msg_error_t peer_wait_for_message(peer_t peer)
+{
+  msg_error_t status;
+  msg_comm_t comm = NULL;
+  msg_task_t task = NULL;
+  int done = 0;
+
+  while (!done) {
+    if (comm == NULL)
+      comm = MSG_task_irecv(&task, peer->me);
+
+    if (MSG_comm_test(comm)) {
+      status = MSG_comm_get_status(comm);
+      //XBT_INFO("peer_wait_for_message: error code = %d", status);
+      xbt_assert(status == MSG_OK, __FILE__ ": peer_wait_for_message() failed");
+      MSG_comm_destroy(comm);
+      comm = NULL;
+      done = peer_execute_task(peer, task);
+      task_message_delete(task);
+      task = NULL;
+    } else {
+      process_pending_connections(peer->pending_sends);
+      MSG_process_sleep(0.01);
+    }
+  }
+
+  return status;
+}
+
+void peer_init(peer_t p)
+{
+  p->init = 0;
+  p->prev = NULL;
+  p->next = NULL;
+  p->pieces = 0;
+  p->close_asap = 0;
+  p->pending_sends = xbt_dynar_new(sizeof(msg_comm_t), NULL);
+  p->me = MSG_host_get_name(MSG_host_self());
+}
+
+void peer_shutdown(peer_t p)
+{
+  float start_time = MSG_get_clock();
+  float end_time = start_time + PEER_SHUTDOWN_DEADLINE;
+
+  XBT_INFO("Waiting for sends to finish before shutdown...");
+  while (xbt_dynar_length(p->pending_sends) && MSG_get_clock() < end_time) {
+    process_pending_connections(p->pending_sends);
+    MSG_process_sleep(0.1);
+  }
 
+  xbt_assert(xbt_dynar_length(p->pending_sends) == 0, "Shutdown failed, sends still pending after deadline");
+  xbt_dynar_free(&p->pending_sends);
 
-  return 0;
+  xbt_free(p);
+}
+
+/** Peer function  */
+int peer(int argc, char *argv[])
+{
+  peer_t p = xbt_new(s_peer_t, 1);
+  msg_error_t status;
+
+  XBT_INFO("peer");
+
+  peer_init(p);
+  status = peer_wait_for_message(p);
+  peer_shutdown(p);
+
+  return MSG_OK;
 }                               /* end_of_receiver */
 
 
@@ -265,6 +524,14 @@ msg_error_t test_all(const char *platform_file,
   /*  Simulation setting */
   MSG_create_environment(platform_file);
 
+  /* Trace categories */
+  TRACE_category_with_color("host0", "0 0 1");
+  TRACE_category_with_color("host1", "0 1 0");
+  TRACE_category_with_color("host2", "0 1 1");
+  TRACE_category_with_color("host3", "1 0 0");
+  TRACE_category_with_color("host4", "1 0 1");
+  TRACE_category_with_color("host5", "1 1 0");
+
   /*   Application deployment */
   MSG_function_register("broadcaster", broadcaster);
   MSG_function_register("peer", peer);
@@ -289,15 +556,14 @@ int main(int argc, char *argv[])
 
   MSG_init(&argc, argv);
 
-
-  if (argc != 3) {
+  /*if (argc <= 3) {
     XBT_CRITICAL("Usage: %s platform_file deployment_file <model>\n",
               argv[0]);
     XBT_CRITICAL
         ("example: %s msg_platform.xml msg_deployment.xml KCCFLN05_Vegas\n",
          argv[0]);
     exit(1);
-  }
+  }*/
 
   /* Options for the workstation/model: