Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Completely eewrite the msg::ptask example
authorMartin Quinson <martin.quinson@loria.fr>
Tue, 9 Feb 2016 21:43:38 +0000 (22:43 +0100)
committerMartin Quinson <martin.quinson@loria.fr>
Tue, 9 Feb 2016 21:47:58 +0000 (22:47 +0100)
- Examplify the parallel execution without communication
- Examplify the remote execution trick
- Kill test_ptask binary that was not any clearer and somehow redundent

Fixes #55, even if it rather reimplement what was proposed.
Sorry @glesserd and thanks for the PR anyway. Your feedback is precious.

examples/msg/parallel_task/CMakeLists.txt
examples/msg/parallel_task/parallel_task.c
examples/msg/parallel_task/parallel_task.tesh
examples/msg/parallel_task/test_ptask.c [deleted file]
examples/msg/parallel_task/test_ptask_deployment.xml [deleted file]

index 5d7e81a..c8ad430 100644 (file)
@@ -1,11 +1,9 @@
 set(EXECUTABLE_OUTPUT_PATH "${CMAKE_CURRENT_BINARY_DIR}")
 
 add_executable(parallel_task parallel_task.c)
-add_executable(test_ptask test_ptask.c)
 
 ### Add definitions for compile
 target_link_libraries(parallel_task simgrid )
-target_link_libraries(test_ptask simgrid )
 
 set(tesh_files
   ${tesh_files}
@@ -14,13 +12,11 @@ set(tesh_files
   )
 set(xml_files
   ${xml_files}
-  ${CMAKE_CURRENT_SOURCE_DIR}/test_ptask_deployment.xml
   PARENT_SCOPE
   )
 set(examples_src
   ${examples_src}
   ${CMAKE_CURRENT_SOURCE_DIR}/parallel_task.c
-  ${CMAKE_CURRENT_SOURCE_DIR}/test_ptask.c
   PARENT_SCOPE
   )
 set(bin_files
index 7035cba..13866dd 100644 (file)
@@ -24,80 +24,73 @@ XBT_LOG_NEW_DEFAULT_CATEGORY(msg_test,
  *   to mix interfaces, but it's not possible ATM).
  */
 
-int test(int argc, char *argv[]);
-msg_error_t test_all(const char *platform_file);
 
-/** Emitter function  */
-int test(int argc, char *argv[])
+/** Function in charge of running the example (that's a simgrid process) */
+static int runner(int argc, char *argv[])
 {
-  xbt_dynar_t slaves_dynar;
-  int slaves_count = 0;
-  msg_host_t *slaves = NULL;
-  double task_comp_size = 100000;
-  double task_comm_size = 10000;
-  double *computation_amount = NULL;
-  double *communication_amount = NULL;
-  msg_task_t ptask = NULL;
   int i, j;
 
-  slaves_dynar = MSG_hosts_as_dynar();
-  slaves_count = xbt_dynar_length(slaves_dynar);
-  slaves = xbt_dynar_to_array(slaves_dynar);
+  /* Retrieve the list of all hosts as an array of hosts */
+  xbt_dynar_t slaves_dynar = MSG_hosts_as_dynar();
+  int slaves_count = xbt_dynar_length(slaves_dynar);
+  msg_host_t *slaves = xbt_dynar_to_array(slaves_dynar);
 
-  computation_amount = xbt_new0(double, slaves_count);
-  communication_amount = xbt_new0(double, slaves_count * slaves_count);
+  XBT_INFO("First, build a classical parallel task, with 1 Gflop to execute on each node, and 10MB to exchange between each pair");
+  double *computation_amounts = xbt_new0(double, slaves_count);
+  double *communication_amounts = xbt_new0(double, slaves_count * slaves_count);
 
   for (i = 0; i < slaves_count; i++)
-    computation_amount[i] = task_comp_size;
+    computation_amounts[i] = 1e9; // 1 Gflop
 
   for (i = 0; i < slaves_count; i++)
     for (j = i + 1; j < slaves_count; j++)
-      communication_amount[i * slaves_count + j] = task_comm_size;
+      communication_amounts[i * slaves_count + j] = 1e7; // 10 MB
 
-  ptask = MSG_parallel_task_create("parallel task",
-                                   slaves_count, slaves,
-                                   computation_amount,
-                                   communication_amount, NULL);
+  msg_task_t ptask = MSG_parallel_task_create("parallel task",
+      slaves_count, slaves, computation_amounts, communication_amounts, NULL /* no specific data to attach */);
   MSG_parallel_task_execute(ptask);
+  MSG_task_destroy(ptask);
+  /* The arrays communication_amounts and computation_amounts are not to be freed manually */
+
+  XBT_INFO("Then, build a parallel task involving only computations and no communication (1 Gflop per node)");
+  computation_amounts = xbt_new0(double, slaves_count);
+  for (i = 0; i < slaves_count; i++)
+    computation_amounts[i] = 1e9; // 1 Gflop
+  ptask = MSG_parallel_task_create("parallel exec", slaves_count, slaves, computation_amounts, NULL/* no comm */, NULL /* no data */);
+  MSG_parallel_task_execute(ptask);
+  MSG_task_destroy(ptask);
 
+  XBT_INFO("Finally, trick the ptask to do a 'remote execution', on host %s", MSG_host_get_name(slaves[1]));
+  computation_amounts = xbt_new0(double, 1);
+  computation_amounts[0] = 1e9; // 1 Gflop
+  msg_host_t *remote = xbt_new(msg_host_t,1);
+  remote[0] = slaves[1];
+  ptask = MSG_parallel_task_create("remote exec", 1, remote, computation_amounts, NULL/* no comm */, NULL /* no data */);
+  MSG_parallel_task_execute(ptask);
   MSG_task_destroy(ptask);
-  /* There is no need to free that! */
-/*   free(communication_amount); */
-/*   free(computation_amount); */
+  free(remote);
 
   XBT_INFO("Goodbye now!");
   free(slaves);
   return 0;
 }
 
-/** Test function */
-msg_error_t test_all(const char *platform_file)
+int main(int argc, char *argv[])
 {
-  msg_error_t res = MSG_OK;
-  xbt_dynar_t all_hosts;
-  msg_host_t first_host;
-
+  MSG_init(&argc, argv);
   MSG_config("host/model", "ptask_L07");
-  MSG_create_environment(platform_file);
-
-  all_hosts = MSG_hosts_as_dynar();
-  first_host = xbt_dynar_getfirst_as(all_hosts,msg_host_t);
-  MSG_process_create("test", test, NULL, first_host);
-  res = MSG_main();
-  xbt_dynar_free(&all_hosts);
 
-  XBT_INFO("Simulation time %g", MSG_get_clock());
-  return res;
-}
+  xbt_assert(argc > 1, "Usage: %s <platform file>", argv[0]);
+  MSG_create_environment(argv[1]);
 
-int main(int argc, char *argv[])
-{
-  msg_error_t res = MSG_OK;
+  /* Pick a process, no matter which, from the platform file */
+  xbt_dynar_t all_hosts = MSG_hosts_as_dynar();
+  msg_host_t first_host = xbt_dynar_getfirst_as(all_hosts,msg_host_t);
+  xbt_dynar_free(&all_hosts);
 
-  MSG_init(&argc, argv);
-  xbt_assert(argc > 1, "Usage: %s platform_file\n"
-       "\tExample: %s msg_platform.xml", argv[0], argv[0]);
-  res = test_all(argv[1]);
+  MSG_process_create("test", runner, NULL, first_host);
+  msg_error_t res = MSG_main();
+  XBT_INFO("Simulation done.");
 
   return res != MSG_OK;
 }
index a5078fa..1f60947 100644 (file)
@@ -1,16 +1,9 @@
 #! ./tesh
 
-p Testing a simple master/slave example application
-
-! output sort 19
 $ $SG_TEST_EXENV parallel_task/parallel_task$EXEEXT ${srcdir:=.}/../platforms/small_platform.xml "--log=root.fmt:[%10.6r]%e(%i:%P@%h)%e%m%n"
 > [  0.000000] (0:maestro@) Switching to the L07 model to handle parallel tasks.
-> [  0.009378] (1:test@Fafard) Goodbye now!
-> [  0.009378] (0:maestro@) Simulation time 0.00937836
-
-! output sort 19
-$ $SG_TEST_EXENV parallel_task/test_ptask$EXEEXT ${srcdir:=.}/../platforms/small_platform.xml ${srcdir:=.}/parallel_task/test_ptask_deployment.xml "--log=root.fmt:[%10.6r]%e(%i:%P@%h)%e%m%n"
-> [  0.000000] (0:maestro@) Switching to the L07 model to handle parallel tasks.
-> [ 20.625396] (1:execute@Ginette) execution_time=20.6254 
-> [ 83.232398] (2:redistribute@Ginette) redistribution_time=83.2324 
-> [ 83.232398] (0:maestro@) Simulation time 83.2324
+> [  0.000000] (1:test@Fafard) First, build a classical parallel task, with 1 Gflop to execute on each node, and 10MB to exchange between each pair
+> [ 20.625396] (1:test@Fafard) Then, build a parallel task involving only computations and no communication (1 Gflop per node)
+> [ 41.247354] (1:test@Fafard) Finally, trick the ptask to do a 'remote execution', on host Tremblay
+> [ 51.441554] (1:test@Fafard) Goodbye now!
+> [ 51.441554] (0:maestro@) Simulation done.
diff --git a/examples/msg/parallel_task/test_ptask.c b/examples/msg/parallel_task/test_ptask.c
deleted file mode 100644 (file)
index 7bf64aa..0000000
+++ /dev/null
@@ -1,168 +0,0 @@
-/* Copyright (c) 2008-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 "simgrid/msg.h"        /* Yeah! If you want to use msg, you need to include simgrid/msg.h */
-#include "xbt/sysdep.h"         /* calloc, printf */
-
-/* Create a log channel to have nice outputs. */
-#include "xbt/log.h"
-#include "xbt/asserts.h"
-XBT_LOG_NEW_DEFAULT_CATEGORY(msg_test,
-                             "Messages specific for this msg example");
-
-int execute(int argc, char *argv[]);
-int redistribute(int argc, char *argv[]);
-msg_error_t test_all(const char *platform_file,
-                     const char *application_file);
-
-
-int execute(int argc, char *argv[])
-{
-  char buffer[32];
-  int i, j;
-  msg_host_t *m_host_list = NULL;
-  msg_task_t task = NULL;
-  int host_list_size;
-  double *computation_duration = NULL;
-  double *communication_table = NULL;
-  double communication_amount = 0;
-  double computation_amount = 0;
-  double execution_time;
-
-
-  host_list_size = argc - 3;
-  XBT_DEBUG("host_list_size=%d", host_list_size);
-  m_host_list = calloc(host_list_size, sizeof(msg_host_t));
-  for (i = 1; i <= host_list_size; i++) {
-    m_host_list[i - 1] = MSG_host_by_name(argv[i]);
-    xbt_assert(m_host_list[i - 1] != NULL,
-                "Unknown host %s. Stopping Now! ", argv[i]);
-  }
-
-  XBT_ATTRIB_UNUSED int read;
-  read = sscanf(argv[argc - 2], "%lg", &computation_amount);
-  xbt_assert(read, "Invalid argument %s\n", argv[argc - 2]);
-  read = sscanf(argv[argc - 1], "%lg", &communication_amount);
-  xbt_assert(read, "Invalid argument %s\n", argv[argc - 1]);
-  computation_duration = (double *) calloc(host_list_size, sizeof(double));
-  communication_table =
-      (double *) calloc(host_list_size * host_list_size, sizeof(double));
-  for (i = 0; i < host_list_size; i++) {
-    computation_duration[i] = computation_amount / host_list_size;
-    for (j = 0; j < host_list_size; j++)
-      communication_table[i * host_list_size + j] =
-          communication_amount / (host_list_size * host_list_size);
-  }
-
-  sprintf(buffer, "redist#0\n");
-  task = MSG_parallel_task_create(buffer,
-                                  host_list_size,
-                                  m_host_list,
-                                  computation_duration,
-                                  communication_table, NULL);
-
-  execution_time = MSG_get_clock();
-  MSG_parallel_task_execute(task);
-  MSG_task_destroy(task);
-  xbt_free(m_host_list);
-  execution_time = MSG_get_clock() - execution_time;
-
-  XBT_INFO("execution_time=%g ", execution_time);
-
-  return 0;
-}
-
-
-int redistribute(int argc, char *argv[])
-{
-  char buffer[32];
-  int i, j;
-  msg_host_t *m_host_list = NULL;
-  msg_task_t task = NULL;
-  int host_list_size;
-  double *computation_duration = NULL;
-  double *communication_table = NULL;
-  double communication_amount = 0;
-  double redistribution_time;
-
-
-  host_list_size = argc - 2;
-  XBT_DEBUG("host_list_size=%d", host_list_size);
-  m_host_list = calloc(host_list_size, sizeof(msg_host_t));
-  for (i = 1; i <= host_list_size; i++) {
-    m_host_list[i - 1] = MSG_host_by_name(argv[i]);
-    xbt_assert(m_host_list[i - 1] != NULL,
-                "Unknown host %s. Stopping Now! ", argv[i]);
-  }
-
-  XBT_ATTRIB_UNUSED int read;
-  read = sscanf(argv[argc - 1], "%lg", &communication_amount);
-  xbt_assert(read, "Invalid argument %s\n", argv[argc - 1]);
-  computation_duration = (double *) calloc(host_list_size, sizeof(double));
-  communication_table =
-      (double *) calloc(host_list_size * host_list_size, sizeof(double));
-  for (i = 0; i < host_list_size; i++) {
-    for (j = 0; j < host_list_size; j++)
-      communication_table[i * host_list_size + j] =
-          communication_amount / (host_list_size * host_list_size);
-  }
-
-  sprintf(buffer, "redist#0\n");
-  task = MSG_parallel_task_create(buffer,
-                                  host_list_size,
-                                  m_host_list,
-                                  computation_duration,
-                                  communication_table, NULL);
-
-  redistribution_time = MSG_get_clock();
-  MSG_parallel_task_execute(task);
-  MSG_task_destroy(task);
-  xbt_free(m_host_list);
-  redistribution_time = MSG_get_clock() - redistribution_time;
-
-  XBT_INFO("redistribution_time=%g ", redistribution_time);
-
-  return 0;
-}
-
-
-msg_error_t test_all(const char *platform_file,
-                     const char *application_file)
-{
-  msg_error_t res = MSG_OK;
-
-
-  MSG_config("host/model", "ptask_L07");
-
-  /*  Simulation setting */
-  MSG_create_environment(platform_file);
-
-  /*   Application deployment */
-  MSG_function_register("execute", execute);
-  MSG_function_register("redistribute", redistribute);
-  MSG_launch_application(application_file);
-
-  res = MSG_main();
-
-  XBT_INFO("Simulation time %g", MSG_get_clock());
-  return res;
-}
-
-
-int main(int argc, char *argv[])
-{
-  msg_error_t res = MSG_OK;
-
-  MSG_init(&argc, argv);
-  xbt_assert(argc > 2, "Usage: %s platform_file deployment_file\n"
-       "\tExample: %s msg_platform.xml msg_deployment.xml\n", 
-       argv[0], argv[0]);
-
-  res = test_all(argv[1], argv[2]);
-
-  return res != MSG_OK;
-}
diff --git a/examples/msg/parallel_task/test_ptask_deployment.xml b/examples/msg/parallel_task/test_ptask_deployment.xml
deleted file mode 100644 (file)
index a8f964a..0000000
+++ /dev/null
@@ -1,20 +0,0 @@
-<?xml version='1.0'?>
-<!DOCTYPE platform SYSTEM "http://simgrid.gforge.inria.fr/simgrid/simgrid.dtd">
-<platform version="4">
-  <!-- The master process (with some arguments) -->
-  <process host="Ginette" function="execute">
-     <argument value="Tremblay"/>
-     <argument value="Jupiter"/>
-     <argument value="Fafard"/> 
-     <argument value="Ginette"/>
-     <argument value="4e9"/>
-     <argument value="16e4"/>
-  </process>
-  <process host="Ginette" function="redistribute">
-    <argument value="Tremblay"/>
-    <argument value="Jupiter"/>
-    <argument value="Fafard"/> 
-    <argument value="Ginette"/>
-    <argument value="16e8"/>
-  </process>
-</platform>