Logo AND Algorithmique Numérique Distribuée

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