Branch data Line data Source code
1 : : //
2 : : // Class Communicator
3 : : // Defines a class to do MPI communication.
4 : : //
5 : : #ifndef IPPL_MPI_COMMUNICATOR_H
6 : : #define IPPL_MPI_COMMUNICATOR_H
7 : :
8 : : #include <memory>
9 : : #include <mpi.h>
10 : :
11 : : #include "Communicate/BufferHandler.h"
12 : : #include "Communicate/LoggingBufferHandler.h"
13 : : #include "Communicate/Request.h"
14 : : #include "Communicate/Status.h"
15 : :
16 : : ////////////////////////////////////////////////
17 : : // For message size check; see below
18 : : #include <climits>
19 : : #include <cstdlib>
20 : :
21 : : #include "Utility/TypeUtils.h"
22 : :
23 : : #include "Communicate/Archive.h"
24 : : #include "Communicate/TagMaker.h"
25 : : #include "Communicate/Tags.h"
26 : : ////////////////////////////////////////////////////
27 : :
28 : : namespace ippl {
29 : : namespace mpi {
30 : :
31 : : class Communicator : public TagMaker {
32 : : public:
33 : : Communicator();
34 : :
35 : : Communicator(MPI_Comm comm);
36 : :
37 : : Communicator& operator=(MPI_Comm comm);
38 : :
39 : 3602 : ~Communicator() = default;
40 : :
41 : : Communicator split(int color, int key) const;
42 : :
43 : 60 : operator const MPI_Comm&() const noexcept { return *comm_m; }
44 : :
45 : 1750 : int size() const noexcept { return size_m; }
46 : :
47 : 4398 : int rank() const noexcept { return rank_m; }
48 : :
49 : 382 : void barrier() { MPI_Barrier(*comm_m); }
50 : :
51 : 0 : void abort(int errorcode = -1) { MPI_Abort(*comm_m, errorcode); }
52 : :
53 : : /*
54 : : * Blocking point-to-point communication
55 : : *
56 : : */
57 : :
58 : : template <typename T>
59 : : void send(const T& buffer, int count, int dest, int tag);
60 : :
61 : : template <typename T>
62 : : void send(const T* buffer, int count, int dest, int tag);
63 : :
64 : : template <typename T>
65 : : void recv(T& output, int count, int source, int tag, Status& status);
66 : :
67 : : template <typename T>
68 : : void recv(T* output, int count, int source, int tag, Status& status);
69 : :
70 : : void probe(int source, int tag, Status& status);
71 : :
72 : : /*
73 : : * Non-blocking point-to-point communication
74 : : *
75 : : */
76 : :
77 : : template <typename T>
78 : : void isend(const T& buffer, int count, int dest, int tag, Request& request);
79 : :
80 : : template <typename T>
81 : : void isend(const T* buffer, int count, int dest, int tag, Request& request);
82 : :
83 : : template <typename T>
84 : : void irecv(T& buffer, int count, int source, int tag, Request& request);
85 : :
86 : : template <typename T>
87 : : void irecv(T* buffer, int count, int source, int tag, Request& request);
88 : :
89 : : bool iprobe(int source, int tag, Status& status);
90 : :
91 : : /*
92 : : * Collective communication
93 : : */
94 : :
95 : : /* Gather the data in the given source container from all other nodes to a
96 : : * specific node (default: 0).
97 : : */
98 : : template <typename T>
99 : : void gather(const T* input, T* output, int count, int root = 0);
100 : :
101 : : /* Scatter the data from all other nodes to a
102 : : * specific node (default: 0).
103 : : */
104 : : template <typename T>
105 : : void scatter(const T* input, T* output, int count, int root = 0);
106 : :
107 : : /* Reduce data coming from all nodes to a specific node
108 : : * (default: 0). Apply certain operation
109 : : *
110 : : */
111 : : template <typename T, class Op>
112 : : void reduce(const T* input, T* output, int count, Op op, int root = 0);
113 : :
114 : : template <typename T, class Op>
115 : : void reduce(const T& input, T& output, int count, Op op, int root = 0);
116 : :
117 : : template <typename T, class Op>
118 : : void allreduce(const T* input, T* output, int count, Op op);
119 : :
120 : : template <typename T, class Op>
121 : : void allreduce(const T& input, T& output, int count, Op op);
122 : :
123 : : template <typename T, class Op>
124 : : void allreduce(T* inout, int count, Op op);
125 : :
126 : : template <typename T, class Op>
127 : : void allreduce(T& inout, int count, Op op);
128 : :
129 : : /////////////////////////////////////////////////////////////////////////////////////
130 : : template <typename MemorySpace = Kokkos::DefaultExecutionSpace::memory_space>
131 : : using archive_type = detail::Archive<MemorySpace>;
132 : :
133 : : template <typename MemorySpace = Kokkos::DefaultExecutionSpace::memory_space>
134 : : using buffer_type = std::shared_ptr<archive_type<MemorySpace>>;
135 : :
136 : : private:
137 : : template <typename MemorySpace>
138 : : using buffer_container_type = LoggingBufferHandler<MemorySpace>;
139 : :
140 : : using buffer_handler_type =
141 : : typename detail::ContainerForAllSpaces<buffer_container_type>::type;
142 : :
143 : : public:
144 : : using size_type = detail::size_type;
145 : 264 : double getDefaultOverallocation() const { return defaultOveralloc_m; }
146 : : void setDefaultOverallocation(double factor);
147 : :
148 : : template <typename MemorySpace = Kokkos::DefaultExecutionSpace::memory_space,
149 : : typename T = char>
150 : : buffer_type<MemorySpace> getBuffer(size_type size, double overallocation = 1.0);
151 : :
152 : : void deleteAllBuffers();
153 : : void freeAllBuffers();
154 : :
155 : : template <typename MemorySpace = Kokkos::DefaultExecutionSpace::memory_space>
156 : : void freeBuffer(buffer_type<MemorySpace> buffer);
157 : :
158 : 8 : const MPI_Comm& getCommunicator() const noexcept { return *comm_m; }
159 : :
160 : : template <class Buffer, typename Archive>
161 : 0 : void recv(int src, int tag, Buffer& buffer, Archive& ar, size_type msize,
162 : : size_type nrecvs) {
163 : : // Temporary fix. MPI communication seems to have problems when the
164 : : // count argument exceeds the range of int, so large messages should
165 : : // be split into smaller messages
166 [ # # ]: 0 : if (msize > INT_MAX) {
167 [ # # # # ]: 0 : std::cerr << "Message size exceeds range of int" << std::endl;
168 [ # # ]: 0 : this->abort();
169 : : }
170 : : MPI_Status status;
171 [ # # # # ]: 0 : MPI_Recv(ar.getBuffer(), msize, MPI_BYTE, src, tag, *comm_m, &status);
172 : :
173 [ # # ]: 0 : buffer.deserialize(ar, nrecvs);
174 : 0 : }
175 : :
176 : : template <class Buffer, typename Archive>
177 : 0 : void isend(int dest, int tag, Buffer& buffer, Archive& ar, MPI_Request& request,
178 : : size_type nsends) {
179 [ # # ]: 0 : if (ar.getSize() > INT_MAX) {
180 : 0 : std::cerr << "Message size exceeds range of int" << std::endl;
181 : 0 : this->abort();
182 : : }
183 : 0 : buffer.serialize(ar, nsends);
184 : 0 : MPI_Isend(ar.getBuffer(), ar.getSize(), MPI_BYTE, dest, tag, *comm_m, &request);
185 : 0 : }
186 : :
187 : : template <typename Archive>
188 : : void irecv(int src, int tag, Archive& ar, MPI_Request& request, size_type msize) {
189 : : if (msize > INT_MAX) {
190 : : std::cerr << "Message size exceeds range of int" << std::endl;
191 : : this->abort();
192 : : }
193 : : MPI_Irecv(ar.getBuffer(), msize, MPI_BYTE, src, tag, *comm_m, &request);
194 : : }
195 : :
196 : : void printLogs(const std::string& filename);
197 : :
198 : : private:
199 : : std::vector<LogEntry> gatherLocalLogs();
200 : : void sendLogsToRank0(const std::vector<LogEntry>& localLogs);
201 : : std::vector<LogEntry> gatherLogsFromAllRanks(const std::vector<LogEntry>& localLogs);
202 : : void writeLogsToFile(const std::vector<LogEntry>& allLogs, const std::string& filename);
203 : :
204 : : buffer_handler_type buffer_handlers_m;
205 : :
206 : : double defaultOveralloc_m = 1.0;
207 : :
208 : : /////////////////////////////////////////////////////////////////////////////////////
209 : :
210 : : protected:
211 : : std::shared_ptr<MPI_Comm> comm_m;
212 : : int size_m;
213 : : int rank_m;
214 : : };
215 : : } // namespace mpi
216 : : } // namespace ippl
217 : :
218 : : #include "Communicate/Collectives.hpp"
219 : : #include "Communicate/PointToPoint.hpp"
220 : :
221 : : ////////////////////////////////////
222 : :
223 : : #include "Communicate/Buffers.hpp"
224 : :
225 : : ////////////////////////////////////
226 : :
227 : : #endif
|
