src/smpi/smpi_mpi_dt.c

   1 /* smpi_mpi_dt.c -- MPI primitives to handle datatypes                        */
   2 /* FIXME: a very incomplete implementation                                    */
   3
   4 /* Copyright (c) 2009, 2010. The SimGrid Team.
   5  * All rights reserved.                                                     */
   6
   7 /* This program is free software; you can redistribute it and/or modify it
   8  * under the terms of the license (GNU LGPL) which comes with this package. */
   9
  10 #include <stdio.h>
  11 #include <stdlib.h>
  12 #include <string.h>
  13
  14 #include "private.h"
  15 #include "smpi_mpi_dt_private.h"
  16
  17 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_mpi_dt, smpi,
  18                                 "Logging specific to SMPI (datatype)");
  19
  20 #define CREATE_MPI_DATATYPE(name, type)       \
  21   static s_smpi_mpi_datatype_t mpi_##name = { \
  22     sizeof(type),  /* size */                 \
  23     0,             /*was 1 has_subtype*/             \
  24     0,             /* lb */                   \
  25     sizeof(type),  /* ub = lb + size */       \
  26     DT_FLAG_BASIC,  /* flags */              \
  27     NULL           /* pointer on extended struct*/ \
  28   };                                          \
  29 MPI_Datatype name = &mpi_##name;
  30
  31 #define CREATE_MPI_DATATYPE_NULL(name)       \
  32   static s_smpi_mpi_datatype_t mpi_##name = { \
  33     0,  /* size */                 \
  34     0,             /*was 1 has_subtype*/             \
  35     0,             /* lb */                   \
  36     0,  /* ub = lb + size */       \
  37     DT_FLAG_BASIC,  /* flags */              \
  38     NULL           /* pointer on extended struct*/ \
  39   };                                          \
  40 MPI_Datatype name = &mpi_##name;
  41
  42 //The following are datatypes for the MPI functions MPI_MAXLOC and MPI_MINLOC.
  43 typedef struct {
  44   float value;
  45   int index;
  46 } float_int;
  47 typedef struct {
  48   long value;
  49   int index;
  50 } long_int;
  51 typedef struct {
  52   double value;
  53   int index;
  54 } double_int;
  55 typedef struct {
  56   short value;
  57   int index;
  58 } short_int;
  59 typedef struct {
  60   int value;
  61   int index;
  62 } int_int;
  63 typedef struct {
  64   long double value;
  65   int index;
  66 } long_double_int;
  67
  68 // Predefined data types
  69 CREATE_MPI_DATATYPE(MPI_CHAR, char);
  70 CREATE_MPI_DATATYPE(MPI_SHORT, short);
  71 CREATE_MPI_DATATYPE(MPI_INT, int);
  72 CREATE_MPI_DATATYPE(MPI_LONG, long);
  73 CREATE_MPI_DATATYPE(MPI_LONG_LONG, long long);
  74 CREATE_MPI_DATATYPE(MPI_SIGNED_CHAR, signed char);
  75 CREATE_MPI_DATATYPE(MPI_UNSIGNED_CHAR, unsigned char);
  76 CREATE_MPI_DATATYPE(MPI_UNSIGNED_SHORT, unsigned short);
  77 CREATE_MPI_DATATYPE(MPI_UNSIGNED, unsigned int);
  78 CREATE_MPI_DATATYPE(MPI_UNSIGNED_LONG, unsigned long);
  79 CREATE_MPI_DATATYPE(MPI_UNSIGNED_LONG_LONG, unsigned long long);
  80 CREATE_MPI_DATATYPE(MPI_FLOAT, float);
  81 CREATE_MPI_DATATYPE(MPI_DOUBLE, double);
  82 CREATE_MPI_DATATYPE(MPI_LONG_DOUBLE, long double);
  83 CREATE_MPI_DATATYPE(MPI_WCHAR, wchar_t);
  84 CREATE_MPI_DATATYPE(MPI_C_BOOL, _Bool);
  85 CREATE_MPI_DATATYPE(MPI_INT8_T, int8_t);
  86 CREATE_MPI_DATATYPE(MPI_INT16_T, int16_t);
  87 CREATE_MPI_DATATYPE(MPI_INT32_T, int32_t);
  88 CREATE_MPI_DATATYPE(MPI_INT64_T, int64_t);
  89 CREATE_MPI_DATATYPE(MPI_UINT8_T, uint8_t);
  90 CREATE_MPI_DATATYPE(MPI_UINT16_T, uint16_t);
  91 CREATE_MPI_DATATYPE(MPI_UINT32_T, uint32_t);
  92 CREATE_MPI_DATATYPE(MPI_UINT64_T, uint64_t);
  93 CREATE_MPI_DATATYPE(MPI_C_FLOAT_COMPLEX, float _Complex);
  94 CREATE_MPI_DATATYPE(MPI_C_DOUBLE_COMPLEX, double _Complex);
  95 CREATE_MPI_DATATYPE(MPI_C_LONG_DOUBLE_COMPLEX, long double _Complex);
  96 CREATE_MPI_DATATYPE(MPI_AINT, MPI_Aint);
  97 CREATE_MPI_DATATYPE(MPI_OFFSET, MPI_Offset);
  98
  99 CREATE_MPI_DATATYPE(MPI_FLOAT_INT, float_int);
 100 CREATE_MPI_DATATYPE(MPI_LONG_INT, long_int);
 101 CREATE_MPI_DATATYPE(MPI_DOUBLE_INT, double_int);
 102 CREATE_MPI_DATATYPE(MPI_SHORT_INT, short_int);
 103 CREATE_MPI_DATATYPE(MPI_2INT, int_int);
 104 CREATE_MPI_DATATYPE(MPI_LONG_DOUBLE_INT, long_double_int);
 105
 106 CREATE_MPI_DATATYPE_NULL(MPI_UB);
 107 CREATE_MPI_DATATYPE_NULL(MPI_LB);
 108 CREATE_MPI_DATATYPE_NULL(MPI_PACKED);
 109 // Internal use only
 110 CREATE_MPI_DATATYPE(MPI_PTR, void*);
 111
 112
 113 size_t smpi_datatype_size(MPI_Datatype datatype)
 114 {
 115   return datatype->size;
 116 }
 117
 118
 119
 120 MPI_Aint smpi_datatype_lb(MPI_Datatype datatype)
 121 {
 122   return datatype->lb;
 123 }
 124
 125 MPI_Aint smpi_datatype_ub(MPI_Datatype datatype)
 126 {
 127   return datatype->ub;
 128 }
 129
 130 int smpi_datatype_extent(MPI_Datatype datatype, MPI_Aint * lb,
 131                          MPI_Aint * extent)
 132 {
 133   int retval;
 134
 135   if ((datatype->flags & DT_FLAG_COMMITED) != DT_FLAG_COMMITED) {
 136     retval = MPI_ERR_TYPE;
 137   } else {
 138     *lb = datatype->lb;
 139     *extent = datatype->ub - datatype->lb;
 140     retval = MPI_SUCCESS;
 141   }
 142   return retval;
 143 }
 144
 145 int smpi_datatype_copy(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 146                        void *recvbuf, int recvcount, MPI_Datatype recvtype)
 147 {
 148   int retval, count;
 149
 150   /* First check if we really have something to do */
 151   if (recvcount == 0) {
 152     retval = sendcount == 0 ? MPI_SUCCESS : MPI_ERR_TRUNCATE;
 153   } else {
 154     /* FIXME: treat packed cases */
 155     sendcount *= smpi_datatype_size(sendtype);
 156     recvcount *= smpi_datatype_size(recvtype);
 157     count = sendcount < recvcount ? sendcount : recvcount;
 158
 159     if(sendtype->has_subtype == 0 && recvtype->has_subtype == 0) {
 160       memcpy(recvbuf, sendbuf, count);
 161     }
 162     else if (sendtype->has_subtype == 0)
 163     {
 164       s_smpi_subtype_t *subtype =  recvtype->substruct;
 165       subtype->unserialize( sendbuf, recvbuf,1, subtype);
 166     }
 167     else if (recvtype->has_subtype == 0)
 168     {
 169       s_smpi_subtype_t *subtype =  sendtype->substruct;
 170       subtype->serialize(sendbuf, recvbuf,1, subtype);
 171     }else{
 172       s_smpi_subtype_t *subtype =  sendtype->substruct;
 173
 174       s_smpi_mpi_vector_t* type_c = (s_smpi_mpi_vector_t*)sendtype;
 175
 176       void * buf_tmp = malloc(count * type_c->size_oldtype);
 177
 178       subtype->serialize( sendbuf, buf_tmp,1, subtype);
 179       subtype =  recvtype->substruct;
 180       subtype->unserialize(recvbuf, buf_tmp,1, subtype);
 181
 182       free(buf_tmp);
 183     }
 184     retval = sendcount > recvcount ? MPI_ERR_TRUNCATE : MPI_SUCCESS;
 185   }
 186
 187   return retval;
 188 }
 189
 190 /*
 191  *  Copies noncontiguous data into contiguous memory.
 192  *  @param contiguous_vector - output vector
 193  *  @param noncontiguous_vector - input vector
 194  *  @param type - pointer contening :
 195  *      - stride - stride of between noncontiguous data
 196  *      - block_length - the width or height of blocked matrix
 197  *      - count - the number of rows of matrix
 198  */
 199 void serialize_vector( const void *noncontiguous_vector,
 200                        void *contiguous_vector,
 201                        size_t count,
 202                        void *type)
 203 {
 204   s_smpi_mpi_vector_t* type_c = (s_smpi_mpi_vector_t*)type;
 205   int i;
 206   char* contiguous_vector_char = (char*)contiguous_vector;
 207   char* noncontiguous_vector_char = (char*)noncontiguous_vector;
 208
 209   for (i = 0; i < type_c->block_count * count; i++) {
 210     memcpy(contiguous_vector_char,
 211            noncontiguous_vector_char, type_c->block_length * type_c->size_oldtype);
 212
 213     contiguous_vector_char += type_c->block_length*type_c->size_oldtype;
 214     noncontiguous_vector_char += type_c->block_stride*type_c->size_oldtype;
 215   }
 216 }
 217
 218 /*
 219  *  Copies contiguous data into noncontiguous memory.
 220  *  @param noncontiguous_vector - output vector
 221  *  @param contiguous_vector - input vector
 222  *  @param type - pointer contening :
 223  *      - stride - stride of between noncontiguous data
 224  *      - block_length - the width or height of blocked matrix
 225  *      - count - the number of rows of matrix
 226  */
 227 void unserialize_vector( const void *contiguous_vector,
 228                          void *noncontiguous_vector,
 229                          size_t count,
 230                          void *type)
 231 {
 232   s_smpi_mpi_vector_t* type_c = (s_smpi_mpi_vector_t*)type;
 233   int i;
 234
 235   char* contiguous_vector_char = (char*)contiguous_vector;
 236   char* noncontiguous_vector_char = (char*)noncontiguous_vector;
 237
 238   for (i = 0; i < type_c->block_count * count; i++) {
 239     memcpy(noncontiguous_vector_char,
 240            contiguous_vector_char, type_c->block_length * type_c->size_oldtype);
 241
 242     contiguous_vector_char += type_c->block_length*type_c->size_oldtype;
 243     noncontiguous_vector_char += type_c->block_stride*type_c->size_oldtype;
 244   }
 245 }
 246
 247 /*
 248  * Create a Sub type vector to be able to serialize and unserialize it
 249  * the structure s_smpi_mpi_vector_t is derived from s_smpi_subtype which
 250  * required the functions unserialize and serialize
 251  *
 252  */
 253 s_smpi_mpi_vector_t* smpi_datatype_vector_create( int block_stride,
 254                                                   int block_length,
 255                                                   int block_count,
 256                                                   MPI_Datatype old_type,
 257                                                   int size_oldtype){
 258   s_smpi_mpi_vector_t *new_t= xbt_new(s_smpi_mpi_vector_t,1);
 259   new_t->base.serialize = &serialize_vector;
 260   new_t->base.unserialize = &unserialize_vector;
 261   new_t->base.subtype_free = &free_vector;
 262   new_t->block_stride = block_stride;
 263   new_t->block_length = block_length;
 264   new_t->block_count = block_count;
 265   new_t->old_type = old_type;
 266   new_t->size_oldtype = size_oldtype;
 267   return new_t;
 268 }
 269
 270 void smpi_datatype_create(MPI_Datatype* new_type, int size,int extent, int has_subtype,
 271                           void *struct_type, int flags){
 272   MPI_Datatype new_t= xbt_new(s_smpi_mpi_datatype_t,1);
 273   new_t->size = size;
 274   new_t->has_subtype = has_subtype;
 275   new_t->lb = 0;
 276   new_t->ub = extent;
 277   new_t->flags = flags;
 278   new_t->substruct = struct_type;
 279   *new_type = new_t;
 280 }
 281
 282 void smpi_datatype_free(MPI_Datatype* type){
 283   if ((*type)->has_subtype == 1){
 284     ((s_smpi_subtype_t *)(*type)->substruct)->subtype_free(type);
 285   }
 286   xbt_free(*type);
 287 }
 288
 289 int smpi_datatype_contiguous(int count, MPI_Datatype old_type, MPI_Datatype* new_type)
 290 {
 291   int retval;
 292   if ((old_type->flags & DT_FLAG_COMMITED) != DT_FLAG_COMMITED) {
 293     retval = MPI_ERR_TYPE;
 294   } else {
 295     smpi_datatype_create(new_type, count *
 296                          smpi_datatype_size(old_type),count *
 297                          smpi_datatype_size(old_type),0,NULL, DT_FLAG_CONTIGUOUS);
 298     retval=MPI_SUCCESS;
 299   }
 300   return retval;
 301 }
 302
 303 int smpi_datatype_vector(int count, int blocklen, int stride, MPI_Datatype old_type, MPI_Datatype* new_type)
 304 {
 305   int retval;
 306   if (blocklen<=0) return MPI_ERR_ARG;
 307   if ((old_type->flags & DT_FLAG_COMMITED) != DT_FLAG_COMMITED) {
 308     retval = MPI_ERR_TYPE;
 309   } else {
 310     if(stride != blocklen){
 311 if (old_type->has_subtype == 1)
 312       XBT_WARN("vector contains a complex type - not yet handled");
 313       s_smpi_mpi_vector_t* subtype = smpi_datatype_vector_create( stride,
 314                                                                   blocklen,
 315                                                                   count,
 316                                                                   old_type,
 317                                                                   smpi_datatype_size(old_type));
 318
 319       smpi_datatype_create(new_type, count * (blocklen) *
 320                            smpi_datatype_size(old_type),
 321                           ((count -1) * stride + blocklen) * smpi_datatype_size(old_type),
 322                            1,
 323                            subtype,
 324                            DT_FLAG_VECTOR);
 325       retval=MPI_SUCCESS;
 326     }else{
 327       /* in this situation the data are contignous thus it's not
 328        * required to serialize and unserialize it*/
 329       smpi_datatype_create(new_type, count * blocklen *
 330                            smpi_datatype_size(old_type), ((count -1) * stride + blocklen)*
 331                            smpi_datatype_size(old_type),
 332                            0,
 333                            NULL,
 334                            DT_FLAG_VECTOR|DT_FLAG_CONTIGUOUS);
 335       retval=MPI_SUCCESS;
 336     }
 337   }
 338   return retval;
 339 }
 340
 341 void free_vector(MPI_Datatype* d){
 342 }
 343
 344 /*
 345 Hvector Implementation - Vector with stride in bytes
 346 */
 347
 348
 349 /*
 350  *  Copies noncontiguous data into contiguous memory.
 351  *  @param contiguous_hvector - output hvector
 352  *  @param noncontiguous_hvector - input hvector
 353  *  @param type - pointer contening :
 354  *      - stride - stride of between noncontiguous data, in bytes
 355  *      - block_length - the width or height of blocked matrix
 356  *      - count - the number of rows of matrix
 357  */
 358 void serialize_hvector( const void *noncontiguous_hvector,
 359                        void *contiguous_hvector,
 360                        size_t count,
 361                        void *type)
 362 {
 363   s_smpi_mpi_hvector_t* type_c = (s_smpi_mpi_hvector_t*)type;
 364   int i;
 365   char* contiguous_vector_char = (char*)contiguous_hvector;
 366   char* noncontiguous_vector_char = (char*)noncontiguous_hvector;
 367
 368   for (i = 0; i < type_c->block_count * count; i++) {
 369     memcpy(contiguous_vector_char,
 370            noncontiguous_vector_char, type_c->block_length * type_c->size_oldtype);
 371
 372     contiguous_vector_char += type_c->block_length*type_c->size_oldtype;
 373     noncontiguous_vector_char += type_c->block_stride;
 374   }
 375 }
 376 /*
 377  *  Copies contiguous data into noncontiguous memory.
 378  *  @param noncontiguous_vector - output hvector
 379  *  @param contiguous_vector - input hvector
 380  *  @param type - pointer contening :
 381  *      - stride - stride of between noncontiguous data, in bytes
 382  *      - block_length - the width or height of blocked matrix
 383  *      - count - the number of rows of matrix
 384  */
 385 void unserialize_hvector( const void *contiguous_vector,
 386                          void *noncontiguous_vector,
 387                          size_t count,
 388                          void *type)
 389 {
 390   s_smpi_mpi_hvector_t* type_c = (s_smpi_mpi_hvector_t*)type;
 391   int i;
 392
 393   char* contiguous_vector_char = (char*)contiguous_vector;
 394   char* noncontiguous_vector_char = (char*)noncontiguous_vector;
 395
 396   for (i = 0; i < type_c->block_count * count; i++) {
 397     memcpy(noncontiguous_vector_char,
 398            contiguous_vector_char, type_c->block_length * type_c->size_oldtype);
 399
 400     contiguous_vector_char += type_c->block_length*type_c->size_oldtype;
 401     noncontiguous_vector_char += type_c->block_stride;
 402   }
 403 }
 404
 405 /*
 406  * Create a Sub type vector to be able to serialize and unserialize it
 407  * the structure s_smpi_mpi_vector_t is derived from s_smpi_subtype which
 408  * required the functions unserialize and serialize
 409  *
 410  */
 411 s_smpi_mpi_hvector_t* smpi_datatype_hvector_create( MPI_Aint block_stride,
 412                                                   int block_length,
 413                                                   int block_count,
 414                                                   MPI_Datatype old_type,
 415                                                   int size_oldtype){
 416   s_smpi_mpi_hvector_t *new_t= xbt_new(s_smpi_mpi_hvector_t,1);
 417   new_t->base.serialize = &serialize_hvector;
 418   new_t->base.unserialize = &unserialize_hvector;
 419   new_t->base.subtype_free = &free_hvector;
 420   new_t->block_stride = block_stride;
 421   new_t->block_length = block_length;
 422   new_t->block_count = block_count;
 423   new_t->old_type = old_type;
 424   new_t->size_oldtype = size_oldtype;
 425   return new_t;
 426 }
 427
 428 //do nothing for vector types
 429 void free_hvector(MPI_Datatype* d){
 430 }
 431
 432 int smpi_datatype_hvector(int count, int blocklen, MPI_Aint stride, MPI_Datatype old_type, MPI_Datatype* new_type)
 433 {
 434   int retval;
 435   if (blocklen<=0) return MPI_ERR_ARG;
 436   if ((old_type->flags & DT_FLAG_COMMITED) != DT_FLAG_COMMITED) {
 437     retval = MPI_ERR_TYPE;
 438   } else {
 439 if (old_type->has_subtype == 1)
 440       XBT_WARN("hvector contains a complex type - not yet handled");
 441     if(stride != blocklen*smpi_datatype_size(old_type)){
 442       s_smpi_mpi_hvector_t* subtype = smpi_datatype_hvector_create( stride,
 443                                                                     blocklen,
 444                                                                     count,
 445                                                                     old_type,
 446                                                                     smpi_datatype_size(old_type));
 447
 448       smpi_datatype_create(new_type, count * blocklen *
 449                            smpi_datatype_size(old_type), (count-1) * stride + blocklen *
 450                            smpi_datatype_size(old_type),
 451                            1,
 452                            subtype,
 453                            DT_FLAG_VECTOR);
 454       retval=MPI_SUCCESS;
 455     }else{
 456       smpi_datatype_create(new_type, count * blocklen *
 457                                                smpi_datatype_size(old_type),count * blocklen *
 458                                                smpi_datatype_size(old_type),
 459                                               0,
 460                                               NULL,
 461                                               DT_FLAG_VECTOR|DT_FLAG_CONTIGUOUS);
 462       retval=MPI_SUCCESS;
 463     }
 464   }
 465   return retval;
 466 }
 467
 468
 469 /*
 470 Indexed Implementation
 471 */
 472
 473 /*
 474  *  Copies noncontiguous data into contiguous memory.
 475  *  @param contiguous_indexed - output indexed
 476  *  @param noncontiguous_indexed - input indexed
 477  *  @param type - pointer contening :
 478  *      - block_lengths - the width or height of blocked matrix
 479  *      - block_indices - indices of each data, in element
 480  *      - count - the number of rows of matrix
 481  */
 482 void serialize_indexed( const void *noncontiguous_indexed,
 483                        void *contiguous_indexed,
 484                        size_t count,
 485                        void *type)
 486 {
 487   s_smpi_mpi_indexed_t* type_c = (s_smpi_mpi_indexed_t*)type;
 488   int i,j;
 489   char* contiguous_indexed_char = (char*)contiguous_indexed;
 490   char* noncontiguous_indexed_char = (char*)noncontiguous_indexed;
 491   for(j=0; j<count;j++){
 492     for (i = 0; i < type_c->block_count; i++) {
 493       memcpy(contiguous_indexed_char,
 494              noncontiguous_indexed_char, type_c->block_lengths[i] * type_c->size_oldtype);
 495
 496       contiguous_indexed_char += type_c->block_lengths[i]*type_c->size_oldtype;
 497       if (i<type_c->block_count-1)noncontiguous_indexed_char = (char*)noncontiguous_indexed + type_c->block_indices[i+1]*type_c->size_oldtype;
 498       else noncontiguous_indexed_char += type_c->block_lengths[i]*type_c->size_oldtype;
 499     }
 500     noncontiguous_indexed=(void*)noncontiguous_indexed_char;
 501   }
 502 }
 503 /*
 504  *  Copies contiguous data into noncontiguous memory.
 505  *  @param noncontiguous_indexed - output indexed
 506  *  @param contiguous_indexed - input indexed
 507  *  @param type - pointer contening :
 508  *      - block_lengths - the width or height of blocked matrix
 509  *      - block_indices - indices of each data, in element
 510  *      - count - the number of rows of matrix
 511  */
 512 void unserialize_indexed( const void *contiguous_indexed,
 513                          void *noncontiguous_indexed,
 514                          size_t count,
 515                          void *type)
 516 {
 517   s_smpi_mpi_indexed_t* type_c = (s_smpi_mpi_indexed_t*)type;
 518   int i,j;
 519
 520   char* contiguous_indexed_char = (char*)contiguous_indexed;
 521   char* noncontiguous_indexed_char = (char*)noncontiguous_indexed;
 522   for(j=0; j<count;j++){
 523     for (i = 0; i < type_c->block_count; i++) {
 524       memcpy(noncontiguous_indexed_char,
 525              contiguous_indexed_char, type_c->block_lengths[i] * type_c->size_oldtype);
 526
 527       contiguous_indexed_char += type_c->block_lengths[i]*type_c->size_oldtype;
 528       if (i<type_c->block_count-1)noncontiguous_indexed_char = (char*)noncontiguous_indexed + type_c->block_indices[i+1]*type_c->size_oldtype;
 529       else noncontiguous_indexed_char += type_c->block_lengths[i]*type_c->size_oldtype;
 530     }
 531     noncontiguous_indexed=(void*)noncontiguous_indexed_char;
 532   }
 533 }
 534
 535 void free_indexed(MPI_Datatype* type){
 536   xbt_free(((s_smpi_mpi_indexed_t *)(*type)->substruct)->block_lengths);
 537   xbt_free(((s_smpi_mpi_indexed_t *)(*type)->substruct)->block_indices);
 538 }
 539
 540 /*
 541  * Create a Sub type indexed to be able to serialize and unserialize it
 542  * the structure s_smpi_mpi_indexed_t is derived from s_smpi_subtype which
 543  * required the functions unserialize and serialize
 544  */
 545 s_smpi_mpi_indexed_t* smpi_datatype_indexed_create( int* block_lengths,
 546                                                   int* block_indices,
 547                                                   int block_count,
 548                                                   MPI_Datatype old_type,
 549                                                   int size_oldtype){
 550   s_smpi_mpi_indexed_t *new_t= xbt_new(s_smpi_mpi_indexed_t,1);
 551   new_t->base.serialize = &serialize_indexed;
 552   new_t->base.unserialize = &unserialize_indexed;
 553   new_t->base.subtype_free = &free_indexed;
 554  //TODO : add a custom function for each time to clean these
 555   new_t->block_lengths= xbt_new(int, block_count);
 556   new_t->block_indices= xbt_new(int, block_count);
 557   int i;
 558   for(i=0;i<block_count;i++){
 559     new_t->block_lengths[i]=block_lengths[i];
 560     new_t->block_indices[i]=block_indices[i];
 561   }
 562   new_t->block_count = block_count;
 563   new_t->old_type = old_type;
 564   new_t->size_oldtype = size_oldtype;
 565   return new_t;
 566 }
 567
 568
 569 int smpi_datatype_indexed(int count, int* blocklens, int* indices, MPI_Datatype old_type, MPI_Datatype* new_type)
 570 {
 571   int i;
 572   int retval;
 573   int size = 0;
 574   int contiguous=1;
 575   for(i=0; i< count; i++){
 576     if   (blocklens[i]<=0)
 577       return MPI_ERR_ARG;
 578     size += blocklens[i];
 579
 580     if ( (i< count -1) && (indices[i]+blocklens[i] != indices[i+1]) )contiguous=0;
 581   }
 582   if ((old_type->flags & DT_FLAG_COMMITED) != DT_FLAG_COMMITED) {
 583     retval = MPI_ERR_TYPE;
 584   } else {
 585
 586     if (old_type->has_subtype == 1)
 587       XBT_WARN("indexed contains a complex type - not yet handled");
 588
 589     if(!contiguous){
 590       s_smpi_mpi_indexed_t* subtype = smpi_datatype_indexed_create( blocklens,
 591                                                                     indices,
 592                                                                     count,
 593                                                                     old_type,
 594                                                                     smpi_datatype_size(old_type));
 595
 596       smpi_datatype_create(new_type,  size *
 597                            smpi_datatype_size(old_type),(indices[count-1]+blocklens[count-1])*smpi_datatype_size(old_type),1, subtype, DT_FLAG_DATA);
 598 }else{
 599       smpi_datatype_create(new_type,  size *
 600                            smpi_datatype_size(old_type),size *
 601                            smpi_datatype_size(old_type),0, NULL, DT_FLAG_DATA|DT_FLAG_CONTIGUOUS);
 602 }
 603     retval=MPI_SUCCESS;
 604   }
 605   return retval;
 606 }
 607
 608
 609 /*
 610 Hindexed Implementation - Indexed with indices in bytes
 611 */
 612
 613 /*
 614  *  Copies noncontiguous data into contiguous memory.
 615  *  @param contiguous_hindexed - output hindexed
 616  *  @param noncontiguous_hindexed - input hindexed
 617  *  @param type - pointer contening :
 618  *      - block_lengths - the width or height of blocked matrix
 619  *      - block_indices - indices of each data, in bytes
 620  *      - count - the number of rows of matrix
 621  */
 622 void serialize_hindexed( const void *noncontiguous_hindexed,
 623                        void *contiguous_hindexed,
 624                        size_t count,
 625                        void *type)
 626 {
 627   s_smpi_mpi_hindexed_t* type_c = (s_smpi_mpi_hindexed_t*)type;
 628   int i,j;
 629   char* contiguous_hindexed_char = (char*)contiguous_hindexed;
 630   char* noncontiguous_hindexed_char = (char*)noncontiguous_hindexed;
 631   for(j=0; j<count;j++){
 632     for (i = 0; i < type_c->block_count; i++) {
 633       memcpy(contiguous_hindexed_char,
 634              noncontiguous_hindexed_char, type_c->block_lengths[i] * type_c->size_oldtype);
 635
 636       contiguous_hindexed_char += type_c->block_lengths[i]*type_c->size_oldtype;
 637       if (i<type_c->block_count-1)noncontiguous_hindexed_char = (char*)noncontiguous_hindexed + type_c->block_indices[i+1];
 638       else noncontiguous_hindexed_char += type_c->block_lengths[i]*type_c->size_oldtype;
 639     }
 640     noncontiguous_hindexed=(void*)noncontiguous_hindexed_char;
 641   }
 642 }
 643 /*
 644  *  Copies contiguous data into noncontiguous memory.
 645  *  @param noncontiguous_hindexed - output hindexed
 646  *  @param contiguous_hindexed - input hindexed
 647  *  @param type - pointer contening :
 648  *      - block_lengths - the width or height of blocked matrix
 649  *      - block_indices - indices of each data, in bytes
 650  *      - count - the number of rows of matrix
 651  */
 652 void unserialize_hindexed( const void *contiguous_hindexed,
 653                          void *noncontiguous_hindexed,
 654                          size_t count,
 655                          void *type)
 656 {
 657   s_smpi_mpi_hindexed_t* type_c = (s_smpi_mpi_hindexed_t*)type;
 658   int i,j;
 659
 660   char* contiguous_hindexed_char = (char*)contiguous_hindexed;
 661   char* noncontiguous_hindexed_char = (char*)noncontiguous_hindexed;
 662   for(j=0; j<count;j++){
 663     for (i = 0; i < type_c->block_count; i++) {
 664       memcpy(noncontiguous_hindexed_char,
 665              contiguous_hindexed_char, type_c->block_lengths[i] * type_c->size_oldtype);
 666
 667       contiguous_hindexed_char += type_c->block_lengths[i]*type_c->size_oldtype;
 668       if (i<type_c->block_count-1)noncontiguous_hindexed_char = (char*)noncontiguous_hindexed + type_c->block_indices[i+1];
 669       else noncontiguous_hindexed_char += type_c->block_lengths[i]*type_c->size_oldtype;
 670     }
 671     noncontiguous_hindexed=(void*)noncontiguous_hindexed_char;
 672   }
 673 }
 674
 675 void free_hindexed(MPI_Datatype* type){
 676   xbt_free(((s_smpi_mpi_hindexed_t *)(*type)->substruct)->block_lengths);
 677   xbt_free(((s_smpi_mpi_hindexed_t *)(*type)->substruct)->block_indices);
 678 }
 679
 680 /*
 681  * Create a Sub type hindexed to be able to serialize and unserialize it
 682  * the structure s_smpi_mpi_hindexed_t is derived from s_smpi_subtype which
 683  * required the functions unserialize and serialize
 684  */
 685 s_smpi_mpi_hindexed_t* smpi_datatype_hindexed_create( int* block_lengths,
 686                                                   MPI_Aint* block_indices,
 687                                                   int block_count,
 688                                                   MPI_Datatype old_type,
 689                                                   int size_oldtype){
 690   s_smpi_mpi_hindexed_t *new_t= xbt_new(s_smpi_mpi_hindexed_t,1);
 691   new_t->base.serialize = &serialize_hindexed;
 692   new_t->base.unserialize = &unserialize_hindexed;
 693   new_t->base.subtype_free = &free_hindexed;
 694  //TODO : add a custom function for each time to clean these
 695   new_t->block_lengths= xbt_new(int, block_count);
 696   new_t->block_indices= xbt_new(MPI_Aint, block_count);
 697   int i;
 698   for(i=0;i<block_count;i++){
 699     new_t->block_lengths[i]=block_lengths[i];
 700     new_t->block_indices[i]=block_indices[i];
 701   }
 702   new_t->block_count = block_count;
 703   new_t->old_type = old_type;
 704   new_t->size_oldtype = size_oldtype;
 705   return new_t;
 706 }
 707
 708
 709 int smpi_datatype_hindexed(int count, int* blocklens, MPI_Aint* indices, MPI_Datatype old_type, MPI_Datatype* new_type)
 710 {
 711   int i;
 712   int retval;
 713   int size = 0;
 714   int contiguous=1;
 715   for(i=0; i< count; i++){
 716     if   (blocklens[i]<=0)
 717       return MPI_ERR_ARG;
 718     size += blocklens[i];
 719
 720
 721     if ( (i< count -1) && (indices[i]+blocklens[i]*smpi_datatype_size(old_type) != indices[i+1]) )contiguous=0;
 722   }
 723   if ((old_type->flags & DT_FLAG_COMMITED) != DT_FLAG_COMMITED) {
 724     retval = MPI_ERR_TYPE;
 725   } else {
 726     if (old_type->has_subtype == 1)
 727       XBT_WARN("hindexed contains a complex type - not yet handled");
 728
 729     if(!contiguous){
 730       s_smpi_mpi_hindexed_t* subtype = smpi_datatype_hindexed_create( blocklens,
 731                                                                     indices,
 732                                                                     count,
 733                                                                     old_type,
 734                                                                     smpi_datatype_size(old_type));
 735
 736       smpi_datatype_create(new_type,  size *
 737                            smpi_datatype_size(old_type),indices[count-1]+blocklens[count-1]*smpi_datatype_size(old_type)
 738                            ,1, subtype, DT_FLAG_DATA);
 739     }else{
 740       smpi_datatype_create(new_type,  size *
 741                            smpi_datatype_size(old_type),size *
 742                            smpi_datatype_size(old_type),0, NULL, DT_FLAG_DATA|DT_FLAG_CONTIGUOUS);
 743     }
 744     retval=MPI_SUCCESS;
 745   }
 746   return retval;
 747 }
 748
 749
 750 /*
 751 struct Implementation - Indexed with indices in bytes
 752 */
 753
 754 /*
 755  *  Copies noncontiguous data into contiguous memory.
 756  *  @param contiguous_struct - output struct
 757  *  @param noncontiguous_struct - input struct
 758  *  @param type - pointer contening :
 759  *      - stride - stride of between noncontiguous data
 760  *      - block_length - the width or height of blocked matrix
 761  *      - count - the number of rows of matrix
 762  */
 763 void serialize_struct( const void *noncontiguous_struct,
 764                        void *contiguous_struct,
 765                        size_t count,
 766                        void *type)
 767 {
 768   s_smpi_mpi_struct_t* type_c = (s_smpi_mpi_struct_t*)type;
 769   int i,j;
 770   char* contiguous_struct_char = (char*)contiguous_struct;
 771   char* noncontiguous_struct_char = (char*)noncontiguous_struct;
 772   for(j=0; j<count;j++){
 773     for (i = 0; i < type_c->block_count; i++) {
 774       memcpy(contiguous_struct_char,
 775              noncontiguous_struct_char, type_c->block_lengths[i] * smpi_datatype_size(type_c->old_types[i]));
 776       contiguous_struct_char += type_c->block_lengths[i]*smpi_datatype_size(type_c->old_types[i]);
 777       if (i<type_c->block_count-1)noncontiguous_struct_char = (char*)noncontiguous_struct + type_c->block_indices[i+1];
 778       else noncontiguous_struct_char += type_c->block_lengths[i]*smpi_datatype_size(type_c->old_types[i]);//let's hope this is MPI_UB ?
 779     }
 780     noncontiguous_struct=(void*)noncontiguous_struct_char;
 781   }
 782 }
 783 /*
 784  *  Copies contiguous data into noncontiguous memory.
 785  *  @param noncontiguous_struct - output struct
 786  *  @param contiguous_struct - input struct
 787  *  @param type - pointer contening :
 788  *      - stride - stride of between noncontiguous data
 789  *      - block_length - the width or height of blocked matrix
 790  *      - count - the number of rows of matrix
 791  */
 792 void unserialize_struct( const void *contiguous_struct,
 793                          void *noncontiguous_struct,
 794                          size_t count,
 795                          void *type)
 796 {
 797   s_smpi_mpi_struct_t* type_c = (s_smpi_mpi_struct_t*)type;
 798   int i,j;
 799
 800   char* contiguous_struct_char = (char*)contiguous_struct;
 801   char* noncontiguous_struct_char = (char*)noncontiguous_struct;
 802   for(j=0; j<count;j++){
 803     for (i = 0; i < type_c->block_count; i++) {
 804       memcpy(noncontiguous_struct_char,
 805              contiguous_struct_char, type_c->block_lengths[i] * smpi_datatype_size(type_c->old_types[i]));
 806       contiguous_struct_char += type_c->block_lengths[i]*smpi_datatype_size(type_c->old_types[i]);
 807       if (i<type_c->block_count-1)noncontiguous_struct_char =  (char*)noncontiguous_struct + type_c->block_indices[i+1];
 808       else noncontiguous_struct_char += type_c->block_lengths[i]*smpi_datatype_size(type_c->old_types[i]);
 809     }
 810     noncontiguous_struct=(void*)noncontiguous_struct_char;
 811
 812   }
 813 }
 814
 815 void free_struct(MPI_Datatype* type){
 816   xbt_free(((s_smpi_mpi_struct_t *)(*type)->substruct)->block_lengths);
 817   xbt_free(((s_smpi_mpi_struct_t *)(*type)->substruct)->block_indices);
 818   xbt_free(((s_smpi_mpi_struct_t *)(*type)->substruct)->old_types);
 819 }
 820
 821 /*
 822  * Create a Sub type struct to be able to serialize and unserialize it
 823  * the structure s_smpi_mpi_struct_t is derived from s_smpi_subtype which
 824  * required the functions unserialize and serialize
 825  */
 826 s_smpi_mpi_struct_t* smpi_datatype_struct_create( int* block_lengths,
 827                                                   MPI_Aint* block_indices,
 828                                                   int block_count,
 829                                                   MPI_Datatype* old_types){
 830   s_smpi_mpi_struct_t *new_t= xbt_new(s_smpi_mpi_struct_t,1);
 831   new_t->base.serialize = &serialize_struct;
 832   new_t->base.unserialize = &unserialize_struct;
 833   new_t->base.subtype_free = &free_struct;
 834  //TODO : add a custom function for each time to clean these
 835   new_t->block_lengths= xbt_new(int, block_count);
 836   new_t->block_indices= xbt_new(MPI_Aint, block_count);
 837   new_t->old_types=  xbt_new(MPI_Datatype, block_count);
 838   int i;
 839   for(i=0;i<block_count;i++){
 840     new_t->block_lengths[i]=block_lengths[i];
 841     new_t->block_indices[i]=block_indices[i];
 842     new_t->old_types[i]=old_types[i];
 843   }
 844   //new_t->block_lengths = block_lengths;
 845   //new_t->block_indices = block_indices;
 846   new_t->block_count = block_count;
 847   //new_t->old_types = old_types;
 848   return new_t;
 849 }
 850
 851
 852 int smpi_datatype_struct(int count, int* blocklens, MPI_Aint* indices, MPI_Datatype* old_types, MPI_Datatype* new_type)
 853 {
 854   int i;
 855   size_t size = 0;
 856   int contiguous=1;
 857   size = 0;
 858   for(i=0; i< count; i++){
 859     if (blocklens[i]<=0)
 860       return MPI_ERR_ARG;
 861     if ((old_types[i]->flags & DT_FLAG_COMMITED) != DT_FLAG_COMMITED)
 862       return MPI_ERR_TYPE;
 863     if (old_types[i]->has_subtype == 1)
 864       XBT_WARN("Struct contains a complex type - not yet handled");
 865     size += blocklens[i]*smpi_datatype_size(old_types[i]);
 866
 867     if ( (i< count -1) && (indices[i]+blocklens[i]*smpi_datatype_size(old_types[i]) != indices[i+1]) )contiguous=0;
 868   }
 869
 870   if(!contiguous){
 871     s_smpi_mpi_struct_t* subtype = smpi_datatype_struct_create( blocklens,
 872                                                               indices,
 873                                                               count,
 874                                                               old_types);
 875
 876     smpi_datatype_create(new_type,  size, indices[count-1] + blocklens[count-1]*smpi_datatype_size(old_types[count-1]),1, subtype, DT_FLAG_DATA);
 877   }else{
 878     smpi_datatype_create(new_type,  size, indices[count-1] + blocklens[count-1]*smpi_datatype_size(old_types[count-1]),0, NULL, DT_FLAG_DATA|DT_FLAG_CONTIGUOUS);
 879   }
 880   return MPI_SUCCESS;
 881 }
 882
 883 void smpi_datatype_commit(MPI_Datatype *datatype)
 884 {
 885   (*datatype)->flags=  ((*datatype)->flags | DT_FLAG_COMMITED);
 886 }
 887
 888 typedef struct s_smpi_mpi_op {
 889   MPI_User_function *func;
 890 } s_smpi_mpi_op_t;
 891
 892 #define MAX_OP(a, b)  (b) = (a) < (b) ? (b) : (a)
 893 #define MIN_OP(a, b)  (b) = (a) < (b) ? (a) : (b)
 894 #define SUM_OP(a, b)  (b) += (a)
 895 #define PROD_OP(a, b) (b) *= (a)
 896 #define LAND_OP(a, b) (b) = (a) && (b)
 897 #define LOR_OP(a, b)  (b) = (a) || (b)
 898 #define LXOR_OP(a, b) (b) = (!(a) && (b)) || ((a) && !(b))
 899 #define BAND_OP(a, b) (b) &= (a)
 900 #define BOR_OP(a, b)  (b) |= (a)
 901 #define BXOR_OP(a, b) (b) ^= (a)
 902 #define MAXLOC_OP(a, b)  (b) = (a.value) < (b.value) ? (b) : (a)
 903 #define MINLOC_OP(a, b)  (b) = (a.value) < (b.value) ? (a) : (b)
 904 //TODO : MINLOC & MAXLOC
 905
 906 #define APPLY_FUNC(a, b, length, type, func) \
 907 {                                          \
 908   int i;                                   \
 909   type* x = (type*)(a);                    \
 910   type* y = (type*)(b);                    \
 911   for(i = 0; i < *(length); i++) {         \
 912     func(x[i], y[i]);                      \
 913   }                                        \
 914 }
 915
 916 static void max_func(void *a, void *b, int *length,
 917                      MPI_Datatype * datatype)
 918 {
 919   if (*datatype == MPI_CHAR) {
 920     APPLY_FUNC(a, b, length, char, MAX_OP);
 921   } else if (*datatype == MPI_SHORT) {
 922     APPLY_FUNC(a, b, length, short, MAX_OP);
 923   } else if (*datatype == MPI_INT) {
 924     APPLY_FUNC(a, b, length, int, MAX_OP);
 925   } else if (*datatype == MPI_LONG) {
 926     APPLY_FUNC(a, b, length, long, MAX_OP);
 927   } else if (*datatype == MPI_UNSIGNED_SHORT) {
 928     APPLY_FUNC(a, b, length, unsigned short, MAX_OP);
 929   } else if (*datatype == MPI_UNSIGNED) {
 930     APPLY_FUNC(a, b, length, unsigned int, MAX_OP);
 931   } else if (*datatype == MPI_UNSIGNED_LONG) {
 932     APPLY_FUNC(a, b, length, unsigned long, MAX_OP);
 933   } else if (*datatype == MPI_FLOAT) {
 934     APPLY_FUNC(a, b, length, float, MAX_OP);
 935   } else if (*datatype == MPI_DOUBLE) {
 936     APPLY_FUNC(a, b, length, double, MAX_OP);
 937   } else if (*datatype == MPI_LONG_DOUBLE) {
 938     APPLY_FUNC(a, b, length, long double, MAX_OP);
 939   }
 940 }
 941
 942 static void min_func(void *a, void *b, int *length,
 943                      MPI_Datatype * datatype)
 944 {
 945   if (*datatype == MPI_CHAR) {
 946     APPLY_FUNC(a, b, length, char, MIN_OP);
 947   } else if (*datatype == MPI_SHORT) {
 948     APPLY_FUNC(a, b, length, short, MIN_OP);
 949   } else if (*datatype == MPI_INT) {
 950     APPLY_FUNC(a, b, length, int, MIN_OP);
 951   } else if (*datatype == MPI_LONG) {
 952     APPLY_FUNC(a, b, length, long, MIN_OP);
 953   } else if (*datatype == MPI_UNSIGNED_SHORT) {
 954     APPLY_FUNC(a, b, length, unsigned short, MIN_OP);
 955   } else if (*datatype == MPI_UNSIGNED) {
 956     APPLY_FUNC(a, b, length, unsigned int, MIN_OP);
 957   } else if (*datatype == MPI_UNSIGNED_LONG) {
 958     APPLY_FUNC(a, b, length, unsigned long, MIN_OP);
 959   } else if (*datatype == MPI_FLOAT) {
 960     APPLY_FUNC(a, b, length, float, MIN_OP);
 961   } else if (*datatype == MPI_DOUBLE) {
 962     APPLY_FUNC(a, b, length, double, MIN_OP);
 963   } else if (*datatype == MPI_LONG_DOUBLE) {
 964     APPLY_FUNC(a, b, length, long double, MIN_OP);
 965   }
 966 }
 967
 968 static void sum_func(void *a, void *b, int *length,
 969                      MPI_Datatype * datatype)
 970 {
 971   if (*datatype == MPI_CHAR) {
 972     APPLY_FUNC(a, b, length, char, SUM_OP);
 973   } else if (*datatype == MPI_SHORT) {
 974     APPLY_FUNC(a, b, length, short, SUM_OP);
 975   } else if (*datatype == MPI_INT) {
 976     APPLY_FUNC(a, b, length, int, SUM_OP);
 977   } else if (*datatype == MPI_LONG) {
 978     APPLY_FUNC(a, b, length, long, SUM_OP);
 979   } else if (*datatype == MPI_UNSIGNED_SHORT) {
 980     APPLY_FUNC(a, b, length, unsigned short, SUM_OP);
 981   } else if (*datatype == MPI_UNSIGNED) {
 982     APPLY_FUNC(a, b, length, unsigned int, SUM_OP);
 983   } else if (*datatype == MPI_UNSIGNED_LONG) {
 984     APPLY_FUNC(a, b, length, unsigned long, SUM_OP);
 985   } else if (*datatype == MPI_FLOAT) {
 986     APPLY_FUNC(a, b, length, float, SUM_OP);
 987   } else if (*datatype == MPI_DOUBLE) {
 988     APPLY_FUNC(a, b, length, double, SUM_OP);
 989   } else if (*datatype == MPI_LONG_DOUBLE) {
 990     APPLY_FUNC(a, b, length, long double, SUM_OP);
 991   } else if (*datatype == MPI_C_FLOAT_COMPLEX) {
 992     APPLY_FUNC(a, b, length, float _Complex, SUM_OP);
 993   } else if (*datatype == MPI_C_DOUBLE_COMPLEX) {
 994     APPLY_FUNC(a, b, length, double _Complex, SUM_OP);
 995   } else if (*datatype == MPI_C_LONG_DOUBLE_COMPLEX) {
 996     APPLY_FUNC(a, b, length, long double _Complex, SUM_OP);
 997   }
 998 }
 999
1000 static void prod_func(void *a, void *b, int *length,
1001                       MPI_Datatype * datatype)
1002 {
1003   if (*datatype == MPI_CHAR) {
1004     APPLY_FUNC(a, b, length, char, PROD_OP);
1005   } else if (*datatype == MPI_SHORT) {
1006     APPLY_FUNC(a, b, length, short, PROD_OP);
1007   } else if (*datatype == MPI_INT) {
1008     APPLY_FUNC(a, b, length, int, PROD_OP);
1009   } else if (*datatype == MPI_LONG) {
1010     APPLY_FUNC(a, b, length, long, PROD_OP);
1011   } else if (*datatype == MPI_UNSIGNED_SHORT) {
1012     APPLY_FUNC(a, b, length, unsigned short, PROD_OP);
1013   } else if (*datatype == MPI_UNSIGNED) {
1014     APPLY_FUNC(a, b, length, unsigned int, PROD_OP);
1015   } else if (*datatype == MPI_UNSIGNED_LONG) {
1016     APPLY_FUNC(a, b, length, unsigned long, PROD_OP);
1017   } else if (*datatype == MPI_FLOAT) {
1018     APPLY_FUNC(a, b, length, float, PROD_OP);
1019   } else if (*datatype == MPI_DOUBLE) {
1020     APPLY_FUNC(a, b, length, double, PROD_OP);
1021   } else if (*datatype == MPI_LONG_DOUBLE) {
1022     APPLY_FUNC(a, b, length, long double, PROD_OP);
1023   } else if (*datatype == MPI_C_FLOAT_COMPLEX) {
1024     APPLY_FUNC(a, b, length, float _Complex, PROD_OP);
1025   } else if (*datatype == MPI_C_DOUBLE_COMPLEX) {
1026     APPLY_FUNC(a, b, length, double _Complex, PROD_OP);
1027   } else if (*datatype == MPI_C_LONG_DOUBLE_COMPLEX) {
1028     APPLY_FUNC(a, b, length, long double _Complex, PROD_OP);
1029   }
1030 }
1031
1032 static void land_func(void *a, void *b, int *length,
1033                       MPI_Datatype * datatype)
1034 {
1035   if (*datatype == MPI_CHAR) {
1036     APPLY_FUNC(a, b, length, char, LAND_OP);
1037   } else if (*datatype == MPI_SHORT) {
1038     APPLY_FUNC(a, b, length, short, LAND_OP);
1039   } else if (*datatype == MPI_INT) {
1040     APPLY_FUNC(a, b, length, int, LAND_OP);
1041   } else if (*datatype == MPI_LONG) {
1042     APPLY_FUNC(a, b, length, long, LAND_OP);
1043   } else if (*datatype == MPI_UNSIGNED_SHORT) {
1044     APPLY_FUNC(a, b, length, unsigned short, LAND_OP);
1045   } else if (*datatype == MPI_UNSIGNED) {
1046     APPLY_FUNC(a, b, length, unsigned int, LAND_OP);
1047   } else if (*datatype == MPI_UNSIGNED_LONG) {
1048     APPLY_FUNC(a, b, length, unsigned long, LAND_OP);
1049   } else if (*datatype == MPI_C_BOOL) {
1050     APPLY_FUNC(a, b, length, _Bool, LAND_OP);
1051   }
1052 }
1053
1054 static void lor_func(void *a, void *b, int *length,
1055                      MPI_Datatype * datatype)
1056 {
1057   if (*datatype == MPI_CHAR) {
1058     APPLY_FUNC(a, b, length, char, LOR_OP);
1059   } else if (*datatype == MPI_SHORT) {
1060     APPLY_FUNC(a, b, length, short, LOR_OP);
1061   } else if (*datatype == MPI_INT) {
1062     APPLY_FUNC(a, b, length, int, LOR_OP);
1063   } else if (*datatype == MPI_LONG) {
1064     APPLY_FUNC(a, b, length, long, LOR_OP);
1065   } else if (*datatype == MPI_UNSIGNED_SHORT) {
1066     APPLY_FUNC(a, b, length, unsigned short, LOR_OP);
1067   } else if (*datatype == MPI_UNSIGNED) {
1068     APPLY_FUNC(a, b, length, unsigned int, LOR_OP);
1069   } else if (*datatype == MPI_UNSIGNED_LONG) {
1070     APPLY_FUNC(a, b, length, unsigned long, LOR_OP);
1071   } else if (*datatype == MPI_C_BOOL) {
1072     APPLY_FUNC(a, b, length, _Bool, LOR_OP);
1073   }
1074 }
1075
1076 static void lxor_func(void *a, void *b, int *length,
1077                       MPI_Datatype * datatype)
1078 {
1079   if (*datatype == MPI_CHAR) {
1080     APPLY_FUNC(a, b, length, char, LXOR_OP);
1081   } else if (*datatype == MPI_SHORT) {
1082     APPLY_FUNC(a, b, length, short, LXOR_OP);
1083   } else if (*datatype == MPI_INT) {
1084     APPLY_FUNC(a, b, length, int, LXOR_OP);
1085   } else if (*datatype == MPI_LONG) {
1086     APPLY_FUNC(a, b, length, long, LXOR_OP);
1087   } else if (*datatype == MPI_UNSIGNED_SHORT) {
1088     APPLY_FUNC(a, b, length, unsigned short, LXOR_OP);
1089   } else if (*datatype == MPI_UNSIGNED) {
1090     APPLY_FUNC(a, b, length, unsigned int, LXOR_OP);
1091   } else if (*datatype == MPI_UNSIGNED_LONG) {
1092     APPLY_FUNC(a, b, length, unsigned long, LXOR_OP);
1093   } else if (*datatype == MPI_C_BOOL) {
1094     APPLY_FUNC(a, b, length, _Bool, LXOR_OP);
1095   }
1096 }
1097
1098 static void band_func(void *a, void *b, int *length,
1099                       MPI_Datatype * datatype)
1100 {
1101   if (*datatype == MPI_CHAR) {
1102     APPLY_FUNC(a, b, length, char, BAND_OP);
1103   }
1104   if (*datatype == MPI_SHORT) {
1105     APPLY_FUNC(a, b, length, short, BAND_OP);
1106   } else if (*datatype == MPI_INT) {
1107     APPLY_FUNC(a, b, length, int, BAND_OP);
1108   } else if (*datatype == MPI_LONG) {
1109     APPLY_FUNC(a, b, length, long, BAND_OP);
1110   } else if (*datatype == MPI_UNSIGNED_SHORT) {
1111     APPLY_FUNC(a, b, length, unsigned short, BAND_OP);
1112   } else if (*datatype == MPI_UNSIGNED) {
1113     APPLY_FUNC(a, b, length, unsigned int, BAND_OP);
1114   } else if (*datatype == MPI_UNSIGNED_LONG) {
1115     APPLY_FUNC(a, b, length, unsigned long, BAND_OP);
1116   } else if (*datatype == MPI_BYTE) {
1117     APPLY_FUNC(a, b, length, uint8_t, BAND_OP);
1118   }
1119 }
1120
1121 static void bor_func(void *a, void *b, int *length,
1122                      MPI_Datatype * datatype)
1123 {
1124   if (*datatype == MPI_CHAR) {
1125     APPLY_FUNC(a, b, length, char, BOR_OP);
1126   } else if (*datatype == MPI_SHORT) {
1127     APPLY_FUNC(a, b, length, short, BOR_OP);
1128   } else if (*datatype == MPI_INT) {
1129     APPLY_FUNC(a, b, length, int, BOR_OP);
1130   } else if (*datatype == MPI_LONG) {
1131     APPLY_FUNC(a, b, length, long, BOR_OP);
1132   } else if (*datatype == MPI_UNSIGNED_SHORT) {
1133     APPLY_FUNC(a, b, length, unsigned short, BOR_OP);
1134   } else if (*datatype == MPI_UNSIGNED) {
1135     APPLY_FUNC(a, b, length, unsigned int, BOR_OP);
1136   } else if (*datatype == MPI_UNSIGNED_LONG) {
1137     APPLY_FUNC(a, b, length, unsigned long, BOR_OP);
1138   } else if (*datatype == MPI_BYTE) {
1139     APPLY_FUNC(a, b, length, uint8_t, BOR_OP);
1140   }
1141 }
1142
1143 static void bxor_func(void *a, void *b, int *length,
1144                       MPI_Datatype * datatype)
1145 {
1146   if (*datatype == MPI_CHAR) {
1147     APPLY_FUNC(a, b, length, char, BXOR_OP);
1148   } else if (*datatype == MPI_SHORT) {
1149     APPLY_FUNC(a, b, length, short, BXOR_OP);
1150   } else if (*datatype == MPI_INT) {
1151     APPLY_FUNC(a, b, length, int, BXOR_OP);
1152   } else if (*datatype == MPI_LONG) {
1153     APPLY_FUNC(a, b, length, long, BXOR_OP);
1154   } else if (*datatype == MPI_UNSIGNED_SHORT) {
1155     APPLY_FUNC(a, b, length, unsigned short, BXOR_OP);
1156   } else if (*datatype == MPI_UNSIGNED) {
1157     APPLY_FUNC(a, b, length, unsigned int, BXOR_OP);
1158   } else if (*datatype == MPI_UNSIGNED_LONG) {
1159     APPLY_FUNC(a, b, length, unsigned long, BXOR_OP);
1160   } else if (*datatype == MPI_BYTE) {
1161     APPLY_FUNC(a, b, length, uint8_t, BXOR_OP);
1162   }
1163 }
1164
1165 static void minloc_func(void *a, void *b, int *length,
1166                         MPI_Datatype * datatype)
1167 {
1168   if (*datatype == MPI_FLOAT_INT) {
1169     APPLY_FUNC(a, b, length, float_int, MINLOC_OP);
1170   } else if (*datatype == MPI_LONG_INT) {
1171     APPLY_FUNC(a, b, length, long_int, MINLOC_OP);
1172   } else if (*datatype == MPI_DOUBLE_INT) {
1173     APPLY_FUNC(a, b, length, double_int, MINLOC_OP);
1174   } else if (*datatype == MPI_SHORT_INT) {
1175     APPLY_FUNC(a, b, length, short_int, MINLOC_OP);
1176   } else if (*datatype == MPI_2INT) {
1177     APPLY_FUNC(a, b, length, int_int, MINLOC_OP);
1178   } else if (*datatype == MPI_LONG_DOUBLE_INT) {
1179     APPLY_FUNC(a, b, length, long_double_int, MINLOC_OP);
1180   }
1181 }
1182
1183 static void maxloc_func(void *a, void *b, int *length,
1184                         MPI_Datatype * datatype)
1185 {
1186   if (*datatype == MPI_FLOAT_INT) {
1187     APPLY_FUNC(a, b, length, float_int, MAXLOC_OP);
1188   } else if (*datatype == MPI_LONG_INT) {
1189     APPLY_FUNC(a, b, length, long_int, MAXLOC_OP);
1190   } else if (*datatype == MPI_DOUBLE_INT) {
1191     APPLY_FUNC(a, b, length, double_int, MAXLOC_OP);
1192   } else if (*datatype == MPI_SHORT_INT) {
1193     APPLY_FUNC(a, b, length, short_int, MAXLOC_OP);
1194   } else if (*datatype == MPI_2INT) {
1195     APPLY_FUNC(a, b, length, int_int, MAXLOC_OP);
1196   } else if (*datatype == MPI_LONG_DOUBLE_INT) {
1197     APPLY_FUNC(a, b, length, long_double_int, MAXLOC_OP);
1198   }
1199 }
1200
1201
1202 #define CREATE_MPI_OP(name, func)                             \
1203   static s_smpi_mpi_op_t mpi_##name = { &(func) /* func */ }; \
1204 MPI_Op name = &mpi_##name;
1205
1206 CREATE_MPI_OP(MPI_MAX, max_func);
1207 CREATE_MPI_OP(MPI_MIN, min_func);
1208 CREATE_MPI_OP(MPI_SUM, sum_func);
1209 CREATE_MPI_OP(MPI_PROD, prod_func);
1210 CREATE_MPI_OP(MPI_LAND, land_func);
1211 CREATE_MPI_OP(MPI_LOR, lor_func);
1212 CREATE_MPI_OP(MPI_LXOR, lxor_func);
1213 CREATE_MPI_OP(MPI_BAND, band_func);
1214 CREATE_MPI_OP(MPI_BOR, bor_func);
1215 CREATE_MPI_OP(MPI_BXOR, bxor_func);
1216 CREATE_MPI_OP(MPI_MAXLOC, maxloc_func);
1217 CREATE_MPI_OP(MPI_MINLOC, minloc_func);
1218
1219 MPI_Op smpi_op_new(MPI_User_function * function, int commute)
1220 {
1221   MPI_Op op;
1222
1223   //FIXME: add commute param
1224   op = xbt_new(s_smpi_mpi_op_t, 1);
1225   op->func = function;
1226   return op;
1227 }
1228
1229 void smpi_op_destroy(MPI_Op op)
1230 {
1231   xbt_free(op);
1232 }
1233
1234 void smpi_op_apply(MPI_Op op, void *invec, void *inoutvec, int *len,
1235                    MPI_Datatype * datatype)
1236 {
1237   op->func(invec, inoutvec, len, datatype);
1238 }