Logo AND Algorithmique Numérique Distribuée

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