Logo AND Algorithmique Numérique Distribuée

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