Many changes done.

msg_test.c works.
Not all functions implemented yet.

git-svn-id: svn+ssh://scm.gforge.inria.fr/svn/simgrid/simgrid/trunk@3365 48e7efb5-ca39-0410-a469-dd3cf9ba447f

diff --git a/src/msg_simix/msg_simix_global.c b/src/msg_simix/msg_simix_global.c
index b7a7664..c549968 100644 (file)
--- a/src/msg_simix/msg_simix_global.c
+++ b/src/msg_simix/msg_simix_global.c
@@ -145,6 +145,7 @@ MSG_error_t MSG_main(void)
                                /* remove conditional from action */
                                xbt_fifo_remove(smx_action->cond_list,cond);
                        }
                                /* remove conditional from action */
                                xbt_fifo_remove(smx_action->cond_list,cond);
                        }

+                       SIMIX_action_destroy(smx_action);

                }
 
                while ( (smx_action = xbt_fifo_pop(actions_done)) ) {
                }
 
                while ( (smx_action = xbt_fifo_pop(actions_done)) ) {


@@ -156,6 +157,7 @@ MSG_error_t MSG_main(void)
                                /* remove conditional from action */
                                xbt_fifo_remove(smx_action->cond_list,cond);
                        }
                                /* remove conditional from action */
                                xbt_fifo_remove(smx_action->cond_list,cond);
                        }

+                       SIMIX_action_destroy(smx_action);

                }
        }
   return MSG_OK;
                }
        }
   return MSG_OK;

diff --git a/src/msg_simix/msg_simix_gos.c b/src/msg_simix/msg_simix_gos.c
index 583266b..439609a 100644 (file)
--- a/src/msg_simix/msg_simix_gos.c
+++ b/src/msg_simix/msg_simix_gos.c
@@ -2,6 +2,8 @@
 #include "xbt/sysdep.h"
 #include "xbt/log.h"
 
 #include "xbt/sysdep.h"
 #include "xbt/log.h"
 

+XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_gos, msg, "Logging specific to MSG (gos)");
+

 /** \defgroup msg_gos_functions MSG Operating System Functions
  *  \brief This section describes the functions that can be used
  *  by an agent for handling some task.
 /** \defgroup msg_gos_functions MSG Operating System Functions
  *  \brief This section describes the functions that can be used
  *  by an agent for handling some task.


@@ -13,7 +15,106 @@ static MSG_error_t __MSG_task_get_with_time_out_from_host(m_task_t * task,
                                                        m_host_t host)
 {
 
                                                        m_host_t host)
 {
 

-       xbt_die("not implemented yet");
+  m_process_t process = MSG_process_self();
+  m_task_t t = NULL;
+  m_host_t h = NULL;
+  simdata_task_t t_simdata = NULL;
+  simdata_host_t h_simdata = NULL;
+  int first_time = 1;
+  xbt_fifo_item_t item = NULL;
+
+       smx_cond_t cond = NULL;                 //conditional wait if the task isn't on the channel yet
+
+  CHECK_HOST();
+  xbt_assert1((channel>=0) && (channel < msg_global->max_channel),"Invalid channel %d",channel);
+  /* Sanity check */
+  xbt_assert0(task,"Null pointer for the task\n");
+
+  if (*task) 
+    CRITICAL0("MSG_task_get() was asked to write in a non empty task struct.");
+
+  /* Get the task */
+  h = MSG_host_self();
+  h_simdata = h->simdata;
+
+  DEBUG2("Waiting for a task on channel %d (%s)", channel,h->name);
+
+       SIMIX_mutex_lock(h->simdata->mutex);
+  while (1) {
+               if(xbt_fifo_size(h_simdata->mbox[channel])>0) {
+                       if(!host) {
+                               t = xbt_fifo_shift(h_simdata->mbox[channel]);
+                               break;
+                       } else {
+                               xbt_fifo_foreach(h->simdata->mbox[channel],item,t,m_task_t) {
+                                       if(t->simdata->source==host) break;
+                               }
+                               if(item) {
+                                       xbt_fifo_remove_item(h->simdata->mbox[channel],item);
+                                       break;
+                               } 
+                       }
+               }
+               
+               if(max_duration>0) {
+                       if(!first_time) {
+                               MSG_RETURN(MSG_TRANSFER_FAILURE);
+                       }
+               }
+           xbt_assert1(!(h_simdata->sleeping[channel]),
+    "A process is already blocked on channel %d",
+    channel);
+       
+               cond = SIMIX_cond_init();
+               h_simdata->sleeping[channel] = cond;
+               if (max_duration > 0) {
+                       SIMIX_cond_wait_timeout(cond, h->simdata->mutex, max_duration);
+               }
+               else SIMIX_cond_wait(h_simdata->sleeping[channel],h->simdata->mutex);
+
+               first_time = 0;
+       }
+       SIMIX_mutex_unlock(h->simdata->mutex);
+
+  DEBUG1("OK, got a task (%s)", t->name);
+       /* clean conditional */
+       if (cond) {
+               SIMIX_cond_destroy(cond);
+               h_simdata->sleeping[channel] = NULL;
+       }
+
+  t_simdata = t->simdata;
+  /*   *task = __MSG_task_copy(t); */
+  *task=t;
+
+       SIMIX_mutex_lock(t_simdata->mutex);
+ // DEBUG1("OK, Mutex task locked (%s)", t->name);
+
+  /* Transfer */
+  t_simdata->using++;
+       /* create SIMIX action to the communication */
+ // DEBUG3("Action (%s), Size (%lf), Rate (%lf)", t->name,t_simdata->message_size, t_simdata->rate);
+       t_simdata->comm = SIMIX_action_communicate(t_simdata->sender->simdata->host->simdata->host,
+                                                                                                                                                                               process->simdata->host->simdata->host,t->name, t_simdata->message_size, t_simdata->rate); 
+                                                                                                                                                                               /*
+       if (MSG_process_is_suspended(t_simdata->sender)) {
+               SIMIX_set_priority(t_simdata->comm,0);
+               t_simdata->comm = SIMIX_action_communicate(t_simdata->sender->simdata->host->simdata->host,
+                                                                                                                                                                                       process->simdata->host->simdata->host,t->name, t_simdata->message_size, t_simdata->rate); 
+       }
+                                                                                                                                                                                       */
+               /* if the process is suspend, create the action but stop its execution, it will be restart when the sender process resume */
+       SIMIX_register_action_to_condition(t_simdata->comm, t_simdata->cond);
+       SIMIX_register_condition_to_action(t_simdata->comm, t_simdata->cond);
+       SIMIX_cond_wait(t_simdata->cond,t_simdata->mutex);
+       DEBUG1("TASSK %s",t->name );
+       /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
+       t->simdata->comm = NULL;
+       t->simdata->compute = NULL;
+       SIMIX_mutex_unlock(t_simdata->mutex);
+
+       MSG_task_destroy(t);
+

        MSG_RETURN(MSG_OK);
 }
 
        MSG_RETURN(MSG_OK);
 }
 


@@ -188,7 +289,57 @@ int MSG_task_probe_from_host(int channel, m_host_t host)
 MSG_error_t MSG_task_put_with_timeout(m_task_t task, m_host_t dest, 
                                      m_channel_t channel, double max_duration)
 {
 MSG_error_t MSG_task_put_with_timeout(m_task_t task, m_host_t dest, 
                                      m_channel_t channel, double max_duration)
 {

-       xbt_die("not implemented yet");
+
+
+  m_process_t process = MSG_process_self();
+  simdata_task_t task_simdata = NULL;
+  m_host_t local_host = NULL;
+  m_host_t remote_host = NULL;
+  CHECK_HOST();
+
+  xbt_assert1((channel>=0) && (channel < msg_global->max_channel),"Invalid channel %d",channel);
+
+  task_simdata = task->simdata;
+  task_simdata->sender = process;
+  task_simdata->source = MSG_process_get_host(process);
+  xbt_assert0(task_simdata->using==1,
+             "This taks is still being used somewhere else. You cannot send it now. Go fix your code!");
+  task_simdata->comm = NULL;
+       task_simdata->cond = SIMIX_cond_init();
+  
+  local_host = ((simdata_process_t) process->simdata)->host;
+  remote_host = dest;
+
+  DEBUG4("Trying to send a task (%g kB) from %s to %s on channel %d", 
+        task->simdata->message_size/1000,local_host->name, remote_host->name, channel);
+
+       SIMIX_mutex_lock(remote_host->simdata->mutex);
+  xbt_fifo_push(((simdata_host_t) remote_host->simdata)->
+               mbox[channel], task);
+
+  
+  if(remote_host->simdata->sleeping[channel]) {
+    DEBUG0("Somebody is listening. Let's wake him up!");
+    //__MSG_process_unblock(remote_host->simdata->sleeping[channel]);
+               SIMIX_cond_signal(remote_host->simdata->sleeping[channel]);
+  }
+       SIMIX_mutex_unlock(remote_host->simdata->mutex);
+
+  process->simdata->put_host = dest;
+  process->simdata->put_channel = channel;
+       SIMIX_mutex_lock(task->simdata->mutex);
+ // DEBUG4("Task sent (%g kB) from %s to %s on channel %d, waiting...", task->simdata->message_size/1000,local_host->name, remote_host->name, channel);
+       DEBUG0("Waiting action finish!");
+       if (max_duration >0) {
+               SIMIX_cond_wait_timeout(task->simdata->cond,task->simdata->mutex,max_duration);
+       }
+       else {
+               SIMIX_cond_wait(task->simdata->cond,task->simdata->mutex);
+       }
+       DEBUG1("Action terminated %s",task->name);    
+       SIMIX_mutex_unlock(task->simdata->mutex);
+
+

        MSG_RETURN(MSG_OK);
 }
 /** \ingroup msg_gos_functions
        MSG_RETURN(MSG_OK);
 }
 /** \ingroup msg_gos_functions


@@ -258,6 +409,8 @@ MSG_error_t MSG_task_execute(m_task_t task)
   simdata = task->simdata;
   xbt_assert0((!simdata->compute)&&(task->simdata->using==1),
              "This taks is executed somewhere else. Go fix your code!");
   simdata = task->simdata;
   xbt_assert0((!simdata->compute)&&(task->simdata->using==1),
              "This taks is executed somewhere else. Go fix your code!");

+       
+       DEBUG1("Computing on %s", MSG_process_self()->simdata->host->name);

   simdata->using++;
        SIMIX_mutex_lock(simdata->mutex);
   simdata->compute = SIMIX_action_execute(SIMIX_host_self(), task->name, simdata->computation_amount);
   simdata->using++;
        SIMIX_mutex_lock(simdata->mutex);
   simdata->compute = SIMIX_action_execute(SIMIX_host_self(), task->name, simdata->computation_amount);


@@ -268,6 +421,10 @@ MSG_error_t MSG_task_execute(m_task_t task)
        
        SIMIX_cond_wait(simdata->cond, simdata->mutex);
 
        
        SIMIX_cond_wait(simdata->cond, simdata->mutex);
 

+       /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
+       simdata->comm = NULL;
+       simdata->compute = NULL;
+

        SIMIX_mutex_unlock(simdata->mutex);
   simdata->using--;
 
        SIMIX_mutex_unlock(simdata->mutex);
   simdata->using--;
 

diff --git a/src/msg_simix/msg_simix_process.c b/src/msg_simix/msg_simix_process.c
index 789862b..b2af5ac 100644 (file)
--- a/src/msg_simix/msg_simix_process.c
+++ b/src/msg_simix/msg_simix_process.c
@@ -34,7 +34,6 @@ static void MSG_process_cleanup(void *arg)
 {
        /* arg is a pointer to a simix process, we can get the msg process with the field data */
        m_process_t proc = ((smx_process_t)arg)->data;
 {
        /* arg is a pointer to a simix process, we can get the msg process with the field data */
        m_process_t proc = ((smx_process_t)arg)->data;

-       printf("remove MSG process %s\n", proc->name);

   xbt_fifo_remove(msg_global->process_list, proc);
        SIMIX_process_cleanup(arg);
   free(proc->name);
   xbt_fifo_remove(msg_global->process_list, proc);
        SIMIX_process_cleanup(arg);
   free(proc->name);


@@ -94,8 +93,7 @@ m_process_t MSG_process_create_with_arguments(const char *name,
   simdata->host = host;
   simdata->argc = argc;
   simdata->argv = argv;
   simdata->host = host;
   simdata->argc = argc;
   simdata->argv = argv;

-       simdata->smx_process = SIMIX_process_create_with_arguments(name, (smx_process_code_t)code,
-                                                                                                                                                                                                                                                       (void*)process, host->name, argc, argv, MSG_process_cleanup );
+       simdata->smx_process = SIMIX_process_create_with_arguments(name, (smx_process_code_t)code, (void*)process, host->name, argc, argv, MSG_process_cleanup );

 
        if (SIMIX_process_self()) {
                simdata->PPID = MSG_process_get_PID(SIMIX_process_self()->data);
 
        if (SIMIX_process_self()) {
                simdata->PPID = MSG_process_get_PID(SIMIX_process_self()->data);

diff --git a/src/msg_simix/msg_simix_task.c b/src/msg_simix/msg_simix_task.c
index fdbcd25..e5c7e36 100644 (file)
--- a/src/msg_simix/msg_simix_task.c
+++ b/src/msg_simix/msg_simix_task.c
@@ -57,6 +57,10 @@ m_task_t MSG_task_create(const char *name, double compute_duration,
        simdata->compute = NULL;
        simdata->comm = NULL;
 
        simdata->compute = NULL;
        simdata->comm = NULL;
 

+       simdata->host_list = NULL;
+       simdata->comp_amount = NULL;
+       simdata->comm_amount = NULL;
+

   return task;
 }
 
   return task;
 }
 


@@ -118,7 +122,7 @@ MSG_error_t MSG_task_destroy(m_task_t task)
   smx_action_t action = NULL;
   xbt_assert0((task != NULL), "Invalid parameter");
 
   smx_action_t action = NULL;
   xbt_assert0((task != NULL), "Invalid parameter");
 

-       /* why? if somebody is using, then you can't free! ok... but will return MSG_OK? when this task will be destroyed, isn't the code wrong? */
+       /* why? if somebody is using, then you can't free! ok... but will return MSG_OK? when this task will be destroyed? isn't the user code wrong? */

   task->simdata->using--;
   if(task->simdata->using>0) return MSG_OK;
 
   task->simdata->using--;
   if(task->simdata->using>0) return MSG_OK;
 

diff --git a/src/simix/smx_deployment.c b/src/simix/smx_deployment.c
index 6df3376..03181ce 100644 (file)
--- a/src/simix/smx_deployment.c
+++ b/src/simix/smx_deployment.c
@@ -66,19 +66,19 @@ static void parse_process_finalize(void)
 
   }
   if((start_time<0) || (start_time==SIMIX_get_clock())) {
 
   }
   if((start_time<0) || (start_time==SIMIX_get_clock())) {

-    DEBUG2("Starting Process %s(%s) right now", parse_argv[0],
-          parse_host);
-                if (simix_global->create_process_function)
-                        process = simix_global->create_process_function(parse_argv[0], parse_code, NULL, parse_host,   parse_argc,parse_argv);
-                else
-                        process = SIMIX_process_create_with_arguments(parse_argv[0], parse_code, NULL, parse_host,     parse_argc,parse_argv, NULL);
-
-    if(kill_time > SIMIX_get_clock()) {
-                if (simix_global->kill_process_function)
-                        surf_timer_resource->extension_public->set(start_time, (void*) simix_global->kill_process_function, arg);
-                else
-      surf_timer_resource->extension_public->set(kill_time, (void*) &SIMIX_process_kill, (void*) process);
-    }
+    DEBUG2("Starting Process %s(%s) right now", parse_argv[0], parse_host);
+
+               if (simix_global->create_process_function)
+                       process = simix_global->create_process_function(parse_argv[0], parse_code, NULL, parse_host,    parse_argc,parse_argv);
+               else
+                       process = SIMIX_process_create_with_arguments(parse_argv[0], parse_code, NULL, parse_host,      parse_argc,parse_argv, NULL);
+
+               if(kill_time > SIMIX_get_clock()) {
+                       if (simix_global->kill_process_function)
+                               surf_timer_resource->extension_public->set(start_time, (void*) simix_global->kill_process_function, arg);
+                       else
+                               surf_timer_resource->extension_public->set(kill_time, (void*) &SIMIX_process_kill, (void*) process);
+               }

   }
 }
 
   }
 }
 

diff --git a/src/simix/smx_environment.c b/src/simix/smx_environment.c
index f346e64..0de1fbf 100644 (file)
--- a/src/simix/smx_environment.c
+++ b/src/simix/smx_environment.c
@@ -55,8 +55,7 @@ void SIMIX_create_environment(const char *file)
   DEBUG1("Model : %s", workstation_model_name);
   if (!strcmp(workstation_model_name,"KCCFLN05")) {
     surf_workstation_resource_init_KCCFLN05(file);
   DEBUG1("Model : %s", workstation_model_name);
   if (!strcmp(workstation_model_name,"KCCFLN05")) {
     surf_workstation_resource_init_KCCFLN05(file);

-  }
-  else if (!strcmp(workstation_model_name,"KCCFLN05_proportionnal")) {
+  } else if (!strcmp(workstation_model_name,"KCCFLN05_proportionnal")) {

     surf_workstation_resource_init_KCCFLN05_proportionnal(file);
   } else if (!strcmp(workstation_model_name,"CLM03")) {
     surf_workstation_resource_init_CLM03(file);
     surf_workstation_resource_init_KCCFLN05_proportionnal(file);
   } else if (!strcmp(workstation_model_name,"CLM03")) {
     surf_workstation_resource_init_CLM03(file);

diff --git a/src/simix/smx_global.c b/src/simix/smx_global.c
index 06e8107..55d01e6 100644 (file)
--- a/src/simix/smx_global.c
+++ b/src/simix/smx_global.c
@@ -51,6 +51,9 @@ void SIMIX_global_init(int *argc, char **argv)
                simix_global->process_list = xbt_swag_new(xbt_swag_offset(proc,process_hookup));
                simix_global->current_process = NULL;
                simix_global->registered_functions = xbt_dict_new();
                simix_global->process_list = xbt_swag_new(xbt_swag_offset(proc,process_hookup));
                simix_global->current_process = NULL;
                simix_global->registered_functions = xbt_dict_new();

+
+               simix_global->create_process_function = NULL;
+               simix_global->kill_process_function = NULL;

        }
 }
 
        }
 }
 


@@ -226,16 +229,17 @@ double SIMIX_solve(xbt_fifo_t actions_done, xbt_fifo_t actions_failed)
        smx_process_t process = NULL;
        int i;
        double elapsed_time = 0.0;
        smx_process_t process = NULL;
        int i;
        double elapsed_time = 0.0;

-       int state_modifications = 0;
-
-
-       //surf_solve(); /* Takes traces into account. Returns 0.0  I don't know what to do with this*/
+       static int state_modifications = 1;
+       static int first = 1;

 
        xbt_context_empty_trash();
        if(xbt_swag_size(simix_global->process_to_run) && (elapsed_time>0)) {
                DEBUG0("**************************************************");
        }
 
        xbt_context_empty_trash();
        if(xbt_swag_size(simix_global->process_to_run) && (elapsed_time>0)) {
                DEBUG0("**************************************************");
        }

-
+       if (first) {
+               surf_solve();/* Takes traces into account. Returns 0.0 */
+               first=0;
+       }

        while ((process = xbt_swag_extract(simix_global->process_to_run))) {
                DEBUG2("Scheduling %s on %s",        
                                process->name,
        while ((process = xbt_swag_extract(simix_global->process_to_run))) {
                DEBUG2("Scheduling %s on %s",        
                                process->name,


@@ -256,8 +260,12 @@ double SIMIX_solve(xbt_fifo_t actions_done, xbt_fifo_t actions_failed)
 
                xbt_dynar_foreach(resource_list, i, resource) {
                        if(xbt_swag_size(resource->common_public->states.failed_action_set) ||
 
                xbt_dynar_foreach(resource_list, i, resource) {
                        if(xbt_swag_size(resource->common_public->states.failed_action_set) ||

-                                       xbt_swag_size(resource->common_public->states.done_action_set))
+                                       xbt_swag_size(resource->common_public->states.done_action_set)) {
+                                       DEBUG2("SWAG SIZES %d %d\n",xbt_swag_size(resource->common_public->states.failed_action_set),
+                                       xbt_swag_size(resource->common_public->states.done_action_set));
+

                                state_modifications = 1;
                                state_modifications = 1;

+                               }

                }
 
                if(!state_modifications) {
                }
 
                if(!state_modifications) {


@@ -305,6 +313,7 @@ double SIMIX_solve(xbt_fifo_t actions_done, xbt_fifo_t actions_failed)
                        }
                }
        }
                        }
                }
        }

+       state_modifications = 0;

 
        if (elapsed_time == -1) {
                if (xbt_swag_size(simix_global->process_list) == 0) {
 
        if (elapsed_time == -1) {
                if (xbt_swag_size(simix_global->process_list) == 0) {

diff --git a/src/simix/smx_synchro.c b/src/simix/smx_synchro.c
index b6d418b..dd084b1 100644 (file)
--- a/src/simix/smx_synchro.c
+++ b/src/simix/smx_synchro.c
@@ -124,11 +124,18 @@ void SIMIX_cond_signal(smx_cond_t cond)
 void SIMIX_cond_wait(smx_cond_t cond,smx_mutex_t mutex)
 {
        smx_process_t self = SIMIX_process_self();
 void SIMIX_cond_wait(smx_cond_t cond,smx_mutex_t mutex)
 {
        smx_process_t self = SIMIX_process_self();

+       smx_action_t act_sleep;

        xbt_assert0((mutex != NULL), "Invalid parameters");
        
        cond->mutex = mutex;
 
        SIMIX_mutex_unlock(mutex);
        xbt_assert0((mutex != NULL), "Invalid parameters");
        
        cond->mutex = mutex;
 
        SIMIX_mutex_unlock(mutex);

+       /* create an action null only if there are no actions already on the condition, usefull if the host crashs */
+       if (xbt_fifo_size(cond->actions) ==0 ) {
+               act_sleep = SIMIX_action_sleep(SIMIX_host_self(), -1);
+               SIMIX_register_action_to_condition(act_sleep,cond);
+               SIMIX_register_condition_to_action(act_sleep,cond);
+       }

        __SIMIX_cond_wait(cond);
        /* get the mutex again */
        self->simdata->mutex = cond->mutex;
        __SIMIX_cond_wait(cond);
        /* get the mutex again */
        self->simdata->mutex = cond->mutex;


@@ -146,7 +153,6 @@ void __SIMIX_cond_wait(smx_cond_t cond)
        self->simdata->cond = cond;
 
        xbt_swag_insert(self, cond->sleeping);
        self->simdata->cond = cond;
 
        xbt_swag_insert(self, cond->sleeping);

-       

        xbt_context_yield();
        self->simdata->cond = NULL;
        while (self->simdata->suspended) {
        xbt_context_yield();
        self->simdata->cond = NULL;
        while (self->simdata->suspended) {