00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024
00025
00026
00027
00028
00029
00030
00031
00032
00033
00034
00035
00036
00037 #include <mpi.h>
00038 #include <vector>
00039
00040 #include "mess.h"
00041 #include "../../core/peo_debug.h"
00042 #include "node.h"
00043
00044 #define MPI_BUF_SIZE 1024*64
00045
00046 static char mpi_buf [MPI_BUF_SIZE];
00047
00048 static int pos_buf;
00049
00050 static std :: vector <char *> act_buf;
00051
00052 static std :: vector <MPI_Request *> act_req;
00053
00054 void initBuffers ()
00055 {
00056
00057 pos_buf = 0;
00058 act_buf.clear ();
00059 act_req.clear ();
00060 }
00061
00062 void cleanBuffers ()
00063 {
00064
00065 for (unsigned i = 0; i < act_req.size ();)
00066 {
00067
00068 MPI_Status stat ;
00069 int flag ;
00070
00071 MPI_Test (act_req [i], & flag, & stat) ;
00072 if (flag)
00073 {
00074
00075 delete[] act_buf [i] ;
00076 delete act_req [i] ;
00077
00078 act_buf [i] = act_buf.back () ;
00079 act_buf.pop_back () ;
00080
00081 act_req [i] = act_req.back () ;
00082 act_req.pop_back () ;
00083 }
00084 else
00085 i ++;
00086 }
00087 }
00088
00089 void waitBuffers ()
00090 {
00091
00092 printDebugMessage ("waiting the termination of the asynchronous operations to complete");
00093
00094 for (unsigned i = 0; i < act_req.size (); i ++)
00095 {
00096
00097 MPI_Status stat ;
00098
00099 MPI_Wait (act_req [i], & stat) ;
00100
00101 delete[] act_buf [i] ;
00102 delete act_req [i] ;
00103 }
00104 }
00105
00106 bool probeMessage (int & __src, int & __tag)
00107 {
00108
00109 int flag;
00110
00111 MPI_Status stat;
00112
00113 MPI_Iprobe (MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, & flag, & stat);
00114
00115 __src = stat.MPI_SOURCE;
00116 __tag = stat.MPI_TAG;
00117
00118 return flag;
00119 }
00120
00121 void waitMessage ()
00122 {
00123
00124 MPI_Status stat;
00125
00126 MPI_Probe (MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, & stat);
00127 }
00128
00129 void initMessage ()
00130 {
00131
00132 pos_buf = 0;
00133 }
00134
00135 void sendMessage (int __to, int __tag)
00136 {
00137
00138 cleanBuffers ();
00139 act_buf.push_back (new char [pos_buf]);
00140 act_req.push_back (new MPI_Request);
00141 memcpy (act_buf.back (), mpi_buf, pos_buf);
00142 MPI_Isend (act_buf.back (), pos_buf, MPI_PACKED, __to, __tag, MPI_COMM_WORLD, act_req.back ());
00143 }
00144
00145 void sendMessageToAll (int __tag)
00146 {
00147
00148 for (int i = 0; i < getNumberOfNodes (); i ++)
00149 sendMessage (i, __tag);
00150 }
00151
00152 void receiveMessage (int __from, int __tag)
00153 {
00154
00155 MPI_Status stat;
00156 MPI_Request req;
00157
00158 MPI_Irecv (mpi_buf, MPI_BUF_SIZE, MPI_PACKED, __from, __tag, MPI_COMM_WORLD, & req);
00159 MPI_Wait (& req, & stat);
00160 }
00161
00162 void synchronizeNodes ()
00163 {
00164
00165 MPI_Barrier ( MPI_COMM_WORLD );
00166 }
00167
00168
00169 void pack (const char & __c)
00170 {
00171
00172 MPI_Pack ((void *) & __c, 1, MPI_CHAR, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00173 }
00174
00175
00176 void pack (const bool & __b, int __nitem)
00177 {
00178
00179 MPI_Pack ((void *) & __b, __nitem, MPI_INT, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00180 }
00181
00182
00183 void pack (const float & __f, int __nitem)
00184 {
00185
00186 MPI_Pack ((void *) & __f, __nitem, MPI_FLOAT, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00187 }
00188
00189
00190 void pack (const double & __d, int __nitem)
00191 {
00192
00193 MPI_Pack ((void *) & __d, __nitem, MPI_DOUBLE, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00194 }
00195
00196
00197 void pack (const int & __i, int __nitem)
00198 {
00199
00200 MPI_Pack ((void *) & __i, __nitem, MPI_INT, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00201 }
00202
00203
00204 void pack (const unsigned int & __ui, int __nitem)
00205 {
00206
00207 MPI_Pack ((void *) & __ui, __nitem, MPI_UNSIGNED, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00208 }
00209
00210
00211 void pack (const short & __sh, int __nitem)
00212 {
00213
00214 MPI_Pack ((void *) & __sh, __nitem, MPI_SHORT, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00215 }
00216
00217
00218 void pack (const unsigned short & __ush, int __nitem)
00219 {
00220
00221 MPI_Pack ((void *) & __ush, __nitem, MPI_UNSIGNED_SHORT, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00222 }
00223
00224
00225 void pack (const long & __l, int __nitem)
00226 {
00227
00228 MPI_Pack ((void *) & __l, __nitem, MPI_LONG, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00229 }
00230
00231
00232 void pack (const unsigned long & __ul, int __nitem)
00233 {
00234
00235 MPI_Pack ((void *) & __ul, __nitem, MPI_UNSIGNED_LONG, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00236 }
00237
00238
00239 void pack (const char * __str)
00240 {
00241
00242 int len = strlen (__str) + 1;
00243 MPI_Pack (& len, 1, MPI_INT, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00244 MPI_Pack ((void *) __str, len, MPI_CHAR, mpi_buf, MPI_BUF_SIZE, & pos_buf, MPI_COMM_WORLD);
00245 }
00246
00247 void pack (const std::string & __str)
00248 {
00249
00250 size_t size = __str.size() + 1;
00251 char * buffer = new char[ size ];
00252 strncpy( buffer, __str.c_str(), size );
00253 pack (buffer);
00254 delete [] buffer;
00255 }
00256
00257
00258 void unpack (char & __c)
00259 {
00260
00261 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __c, 1, MPI_CHAR, MPI_COMM_WORLD);
00262 }
00263
00264
00265 extern void unpack (bool & __b, int __nitem )
00266 {
00267
00268 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __b, __nitem, MPI_INT, MPI_COMM_WORLD);
00269 }
00270
00271
00272 void unpack (float & __f, int __nitem)
00273 {
00274
00275 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __f, __nitem, MPI_FLOAT, MPI_COMM_WORLD);
00276 }
00277
00278
00279 void unpack (double & __d, int __nitem)
00280 {
00281
00282 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __d, __nitem, MPI_DOUBLE, MPI_COMM_WORLD);
00283 }
00284
00285
00286 void unpack (int & __i, int __nitem)
00287 {
00288
00289 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __i, __nitem, MPI_INT, MPI_COMM_WORLD);
00290 }
00291
00292
00293 void unpack (unsigned int & __ui, int __nitem)
00294 {
00295
00296 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __ui, __nitem, MPI_UNSIGNED, MPI_COMM_WORLD);
00297 }
00298
00299
00300 void unpack (short & __sh, int __nitem)
00301 {
00302
00303 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __sh, __nitem, MPI_SHORT, MPI_COMM_WORLD);
00304 }
00305
00306
00307 void unpack (unsigned short & __ush, int __nitem)
00308 {
00309
00310 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __ush, __nitem, MPI_UNSIGNED_SHORT, MPI_COMM_WORLD);
00311 }
00312
00313
00314 void unpack (long & __l, int __nitem)
00315 {
00316
00317 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __l, __nitem, MPI_LONG, MPI_COMM_WORLD);
00318 }
00319
00320
00321 void unpack (unsigned long & __ul, int __nitem)
00322 {
00323
00324 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & __ul, __nitem, MPI_UNSIGNED_LONG, MPI_COMM_WORLD);
00325 }
00326
00327
00328 void unpack (char * __str)
00329 {
00330
00331 int len;
00332 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & len, 1, MPI_INT, MPI_COMM_WORLD);
00333 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, __str, len, MPI_CHAR, MPI_COMM_WORLD);
00334 }
00335 void unpack (std::string & __str)
00336 {
00337
00338 char * buffer;
00339 int len;
00340 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, & len, 1, MPI_INT, MPI_COMM_WORLD);
00341 MPI_Unpack (mpi_buf, MPI_BUF_SIZE, & pos_buf, buffer, len, MPI_CHAR, MPI_COMM_WORLD);
00342 __str.assign( buffer );
00343
00344 }
00345