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