1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
#include <mpi.h>
#include <cstdlib>
#include <cstdio>
#include "send2one.h"
#include "memory.h"
#include "error.h"
/* ---------------------------------------------------------------------- */
Send2One::Send2One(MPI_Comm caller_comm)
{
comm = caller_comm;
MPI_Comm_rank(comm,&me);
MPI_Comm_size(comm,&nprocs);
memory = new Memory(comm);
error = new Error(comm);
buf = NULL;
maxbuf = 0;
}
/* ---------------------------------------------------------------------- */
Send2One::~Send2One()
{
delete memory;
delete error;
memory->sfree(buf);
}
/* ---------------------------------------------------------------------- */
void Send2One::execute()
{
int nme,nmax,nsize,ping;
MPI_Status status;
MPI_Request request;
// pre-processing before ping loop
pre();
// nme = size of data I contribute, in bytes
// nmax = max size of data on any proc, in bytes
// reallocate buf if necessary
nme = size();
MPI_Allreduce(&nme,&nmax,1,MPI_INT,MPI_MAX,comm);
if (nmax > maxbuf) {
maxbuf = nmax;
memory->sfree(buf);
buf = (char *) memory->smalloc(maxbuf,"foo:buf");
}
// pack my data into buf
pack(buf);
// proc 0 pings each proc, receives its data
// all other procs wait for ping, send their data to proc 0
// invoke process() to work with data
if (me == 0) {
for (int iproc = 0; iproc < nprocs; iproc++) {
if (iproc) {
MPI_Irecv(buf,maxbuf,MPI_CHAR,iproc,0,comm,&request);
MPI_Send(&ping,0,MPI_INT,iproc,0,comm);
MPI_Wait(&request,&status);
MPI_Get_count(&status,MPI_CHAR,&nsize);
} else nsize = nme;
process(nsize,buf);
}
} else {
MPI_Recv(&ping,0,MPI_INT,0,0,comm,&status);
MPI_Rsend(buf,nme,MPI_CHAR,0,0,comm);
}
post();
}
|
