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