#include <errno.h>
#include "simix/smx_private.h"
#include "surf/surf.h"
+#include "simgrid/sg_config.h"
XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi, "Logging specific to SMPI (base)");
} else return 0;
}
+
+typedef struct s_smpi_factor *smpi_factor_t;
+typedef struct s_smpi_factor {
+ long factor;
+ int nb_values;
+ double values[4];//arbitrary set to 4
+} s_smpi_factor_t;
+xbt_dynar_t smpi_os_values = NULL;
+xbt_dynar_t smpi_or_values = NULL;
+
+
+// Methods used to parse and store the values for timing injections in smpi
+// These are taken from surf/network.c and generalized to have more factors
+// These methods should be merged with those in surf/network.c (moved somewhere in xbt ?)
+
+static int factor_cmp(const void *pa, const void *pb)
+{
+ return (((s_smpi_factor_t*)pa)->factor > ((s_smpi_factor_t*)pb)->factor);
+}
+
+
+static xbt_dynar_t parse_factor(const char *smpi_coef_string)
+{
+ char *value = NULL;
+ unsigned int iter = 0;
+ s_smpi_factor_t fact;
+ int i=0;
+ xbt_dynar_t smpi_factor, radical_elements, radical_elements2 = NULL;
+
+ smpi_factor = xbt_dynar_new(sizeof(s_smpi_factor_t), NULL);
+ radical_elements = xbt_str_split(smpi_coef_string, ";");
+ xbt_dynar_foreach(radical_elements, iter, value) {
+ fact.nb_values=0;
+ radical_elements2 = xbt_str_split(value, ":");
+ if (xbt_dynar_length(radical_elements2) <2 || xbt_dynar_length(radical_elements2) > 5)
+ xbt_die("Malformed radical for smpi factor!");
+ for(i =0; i<xbt_dynar_length(radical_elements2);i++ ){
+ if (i==0){
+ fact.factor = atol(xbt_dynar_get_as(radical_elements2, i, char *));
+ }else{
+ fact.values[fact.nb_values] = atof(xbt_dynar_get_as(radical_elements2, i, char *));
+ fact.nb_values++;
+ }
+ }
+
+ xbt_dynar_push_as(smpi_factor, s_smpi_factor_t, fact);
+ XBT_DEBUG("smpi_factor:\t%ld : %d values, first: %f", fact.factor, fact.nb_values ,fact.values[0]);
+ xbt_dynar_free(&radical_elements2);
+ }
+ xbt_dynar_free(&radical_elements);
+ iter=0;
+ xbt_dynar_sort(smpi_factor, &factor_cmp);
+ xbt_dynar_foreach(smpi_factor, iter, fact) {
+ XBT_DEBUG("smpi_factor:\t%ld : %d values, first: %f", fact.factor, fact.nb_values ,fact.values[0]);
+ }
+ return smpi_factor;
+}
+
+static double smpi_os(double size)
+{
+ if (!smpi_os_values)
+ smpi_os_values =
+ parse_factor(sg_cfg_get_string("smpi/os"));
+
+ unsigned int iter = 0;
+ s_smpi_factor_t fact;
+ double current=0.0;
+ xbt_dynar_foreach(smpi_os_values, iter, fact) {
+ if (size <= fact.factor) {
+ XBT_DEBUG("os : %lf <= %ld return %f", size, fact.factor, current);
+ return current;
+ }else{
+ current=fact.values[0]+fact.values[1]*size;
+ }
+ }
+ XBT_DEBUG("os : %lf > %ld return %f", size, fact.factor, current);
+
+ return current;
+}
+
+static double smpi_or(double size)
+{
+ if (!smpi_or_values)
+ smpi_or_values =
+ parse_factor(sg_cfg_get_string("smpi/or"));
+
+ unsigned int iter = 0;
+ s_smpi_factor_t fact;
+ double current=0.0;
+ xbt_dynar_foreach(smpi_or_values, iter, fact) {
+ if (size <= fact.factor) {
+ XBT_DEBUG("or : %lf <= %ld return %f", size, fact.factor, current);
+ return current;
+ }else
+ current=fact.values[0]+fact.values[1]*size;
+ }
+ XBT_DEBUG("or : %lf > %ld return %f", size, fact.factor, current);
+
+ return current;
+}
+
static MPI_Request build_request(void *buf, int count,
MPI_Datatype datatype, int src, int dst,
int tag, MPI_Comm comm, unsigned flags)
if(datatype->has_subtype == 1){
// This part handles the problem of non-contiguous memory
old_buf = buf;
- buf = malloc(count*smpi_datatype_size(datatype));
+ buf = xbt_malloc(count*smpi_datatype_size(datatype));
if (flags & SEND) {
subtype->serialize(old_buf, buf, count, datatype->substruct);
}
"Cannot (re)start a non-finished communication");
if(request->flags & RECV) {
print_request("New recv", request);
- if (request->size < surf_cfg_get_int("smpi/async_small_thres"))
+ if (request->size < sg_cfg_get_int("smpi/async_small_thres"))
mailbox = smpi_process_mailbox_small();
else
mailbox = smpi_process_mailbox();
request->real_size=request->size;
smpi_datatype_use(request->old_type);
request->action = simcall_comm_irecv(mailbox, request->buf, &request->real_size, &match_recv, request);
+
+ double sleeptime = smpi_or(request->size);
+ if(sleeptime!=0.0){
+ simcall_process_sleep(sleeptime);
+ XBT_DEBUG("receiving size of %ld : sleep %lf ", request->size, smpi_or(request->size));
+ }
+
} else {
int receiver = smpi_group_index(smpi_comm_group(request->comm), request->dst);
/* return;*/
/* }*/
print_request("New send", request);
- if (request->size < surf_cfg_get_int("smpi/async_small_thres")) { // eager mode
+ if (request->size < sg_cfg_get_int("smpi/async_small_thres")) { // eager mode
mailbox = smpi_process_remote_mailbox_small(receiver);
}else{
XBT_DEBUG("Send request %p is not in the permanent receive mailbox (buf: %p)",request,request->buf);
if(request->old_type->has_subtype == 0){
oldbuf = request->buf;
if (oldbuf){
- request->buf = malloc(request->size);
+ request->buf = xbt_malloc(request->size);
memcpy(request->buf,oldbuf,request->size);
}
}
// we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later
request->real_size=request->size;
smpi_datatype_use(request->old_type);
-
+ double sleeptime = smpi_os(request->size);
+ if(sleeptime!=0.0){
+ simcall_process_sleep(sleeptime);
+ XBT_DEBUG("sending size of %ld : sleep %lf ", request->size, smpi_os(request->size));
+ }
request->action =
simcall_comm_isend(mailbox, request->size, -1.0,
request->buf, request->real_size,
print_request("New iprobe", request);
// We have to test both mailboxes as we don't know if we will receive one one or another
- if (surf_cfg_get_int("smpi/async_small_thres")>0){
+ if (sg_cfg_get_int("smpi/async_small_thres")>0){
mailbox = smpi_process_mailbox_small();
XBT_DEBUG("trying to probe the perm recv mailbox");
request->action = simcall_comm_iprobe(mailbox, request->src, request->tag, &match_recv, (void*)request);
