processorLduInterfaceTemplates.C
Go to the documentation of this file.
1 /*---------------------------------------------------------------------------*\
2  ========= |
3  \\ / F ield | OpenFOAM: The Open Source CFD Toolbox
4  \\ / O peration |
5  \\ / A nd | www.openfoam.com
6  \\/ M anipulation |
7 -------------------------------------------------------------------------------
8  Copyright (C) 2011-2017 OpenFOAM Foundation
9  Copyright (C) 2019-2025 OpenCFD Ltd.
10 -------------------------------------------------------------------------------
11 License
12  This file is part of OpenFOAM.
13 
14  OpenFOAM is free software: you can redistribute it and/or modify it
15  under the terms of the GNU General Public License as published by
16  the Free Software Foundation, either version 3 of the License, or
17  (at your option) any later version.
18 
19  OpenFOAM is distributed in the hope that it will be useful, but WITHOUT
20  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
21  FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
22  for more details.
23 
24  You should have received a copy of the GNU General Public License
25  along with OpenFOAM. If not, see <http://www.gnu.org/licenses/>.
26 
27 \*---------------------------------------------------------------------------*/
28 
29 #include "processorLduInterface.H"
30 #include "IPstream.H"
31 #include "OPstream.H"
32 
33 // * * * * * * * * * * * * * * * Member Functions * * * * * * * * * * * * * //
34 
35 template<class Type>
37 (
38  const UPstream::commsTypes commsType,
39  const UList<Type>& fld
40 ) const
41 {
42  if constexpr (!is_contiguous_v<Type>)
43  {
45  << "Invalid for non-contiguous data types"
47  }
48 
49  if
50  (
52  || commsType == UPstream::commsTypes::scheduled
53  )
54  {
56  (
57  commsType,
58  neighbProcNo(),
59  fld,
60  tag(),
61  comm()
62  );
63  }
64  else if (commsType == UPstream::commsTypes::nonBlocking)
65  {
66  if (fld.empty())
67  {
68  // Can skip empty messages
69  return;
70  }
71 
72  const label nBytes = fld.size_bytes();
73 
74  resizeBuf(byteSendBuf_, nBytes);
75  resizeBuf(byteRecvBuf_, nBytes);
76 
77  std::memcpy
78  (
79  static_cast<void*>(byteSendBuf_.data()), fld.cdata(), nBytes
80  );
81 
83  (
84  commsType,
85  neighbProcNo(),
86  byteRecvBuf_.data(),
87  nBytes,
88  tag(),
89  comm()
90  );
91 
93  (
94  commsType,
95  neighbProcNo(),
96  byteSendBuf_.cdata(),
97  nBytes,
98  tag(),
99  comm()
100  );
101  }
102  else
103  {
105  << "Unsupported communications type " << int(commsType)
107  }
108 }
109 
110 
111 template<class Type>
113 (
114  const UPstream::commsTypes commsType,
115  UList<Type>& fld
116 ) const
117 {
118  if constexpr (!is_contiguous_v<Type>)
119  {
121  << "Invalid for non-contiguous data types"
123  }
124 
125  if
126  (
127  commsType == UPstream::commsTypes::buffered
128  || commsType == UPstream::commsTypes::scheduled
129  )
130  {
132  (
133  commsType,
134  neighbProcNo(),
135  fld,
136  tag(),
137  comm()
138  );
139  }
140  else if (commsType == UPstream::commsTypes::nonBlocking)
141  {
142  const label nBytes = fld.size_bytes();
143 
144  std::memcpy
145  (
146  static_cast<void*>(fld.data()), byteRecvBuf_.cdata(), nBytes
147  );
148  }
149  else
150  {
152  << "Unsupported communications type " << int(commsType)
154  }
155 }
156 
157 
158 template<class Type>
160 (
161  const UPstream::commsTypes commsType,
162  const label size
163 ) const
164 {
165  auto tfld = tmp<Field<Type>>::New(size);
166  receive(commsType, tfld.ref());
167  return tfld;
168 }
169 
170 
171 template<class Type>
173 (
174  const UPstream::commsTypes commsType,
175  const UList<Type>& f
176 ) const
177 {
178  if constexpr
179  (
180  std::is_integral_v<Type>
181  || (sizeof(float) == sizeof(scalar))
182  )
183  {
184  // No compression if integral or scalar is float
185  this->send(commsType, f);
186  }
187  else if (f.empty() || !UPstream::floatTransfer)
188  {
189  // No compression
190  this->send(commsType, f);
191  }
192  else // (!f.empty() && UPstream::floatTransfer)
193  {
194  static const label nCmpts = (sizeof(Type)/sizeof(scalar));
195  const label nm1 = (f.size() - 1)*nCmpts;
196  const label nBytes = f.size()*nCmpts*sizeof(float);
197 
198  const scalar *sArray = reinterpret_cast<const scalar*>(f.cdata());
199  const scalar *slast = &sArray[nm1];
200  resizeBuf(byteSendBuf_, nBytes);
201  float *fArray = reinterpret_cast<float*>(byteSendBuf_.data());
202 
203  for (label i=0; i<nm1; i++)
204  {
205  fArray[i] = sArray[i] - slast[i%nCmpts];
206  }
207 
208  reinterpret_cast<Type&>(fArray[nm1]) = f.last();
209 
210  if
211  (
212  commsType == UPstream::commsTypes::buffered
213  || commsType == UPstream::commsTypes::scheduled
214  )
215  {
217  (
218  commsType,
219  neighbProcNo(),
220  byteSendBuf_.cdata(),
221  nBytes,
222  tag(),
223  comm()
224  );
225  }
226  else if (commsType == UPstream::commsTypes::nonBlocking)
227  {
228  resizeBuf(byteRecvBuf_, nBytes);
229 
231  (
232  commsType,
233  neighbProcNo(),
234  byteRecvBuf_.data(),
235  nBytes,
236  tag(),
237  comm()
238  );
239 
241  (
242  commsType,
243  neighbProcNo(),
244  byteSendBuf_.cdata(),
245  nBytes,
246  tag(),
247  comm()
248  );
249  }
250  else
251  {
253  << "Unsupported communications type " << int(commsType)
254  << exit(FatalError);
255  }
256  }
257 }
258 
259 
260 template<class Type>
262 (
263  const UPstream::commsTypes commsType,
264  UList<Type>& f
265 ) const
266 {
267  if constexpr
268  (
269  std::is_integral_v<Type>
270  || (sizeof(float) == sizeof(scalar))
271  )
272  {
273  // No compression if integral or scalar is float
274  this->receive<Type>(commsType, f);
275  }
276  else if (f.empty() || !UPstream::floatTransfer)
277  {
278  // Nothing to compress
279  this->receive<Type>(commsType, f);
280  }
281  else // (!f.empty() && UPstream::floatTransfer)
282  {
283  static const label nCmpts = (sizeof(Type)/sizeof(scalar));
284  const label nm1 = (f.size() - 1)*nCmpts;
285  const label nBytes = f.size()*nCmpts*sizeof(float);
286 
287  if
288  (
289  commsType == UPstream::commsTypes::buffered
290  || commsType == UPstream::commsTypes::scheduled
291  )
292  {
293  resizeBuf(byteRecvBuf_, nBytes);
294 
296  (
297  commsType,
298  neighbProcNo(),
299  byteRecvBuf_.data(),
300  nBytes,
301  tag(),
302  comm()
303  );
304  }
305  else if (commsType != UPstream::commsTypes::nonBlocking)
306  {
308  << "Unsupported communications type " << int(commsType)
309  << exit(FatalError);
310  }
311 
312  const float *fArray =
313  reinterpret_cast<const float*>(byteRecvBuf_.cdata());
314  f.last() = reinterpret_cast<const Type&>(fArray[nm1]);
315  scalar *sArray = reinterpret_cast<scalar*>(f.data());
316  const scalar *slast = &sArray[nm1];
317 
318  for (label i=0; i<nm1; i++)
319  {
320  sArray[i] = fArray[i] + slast[i%nCmpts];
321  }
322  }
323 }
324 
325 
326 template<class Type>
328 (
329  const UPstream::commsTypes commsType,
330  const label size
331 ) const
332 {
333  auto tfld = tmp<Field<Type>>::New(size);
334  compressedReceive(commsType, tfld.ref());
335  return tfld;
336 }
337 
338 
339 // ************************************************************************* //
static bool floatTransfer
Should compact transfer be used in which floats replace doubles reducing the bandwidth requirement at...
Definition: UPstream.H:942
void size(const label n)
Older name for setAddressableSize.
Definition: UList.H:114
void send(const UPstream::commsTypes commsType, const UList< Type > &f) const
Raw send function.
errorManipArg< error, int > exit(error &err, const int errNo=1)
Definition: errorManip.H:125
commsTypes
Communications types.
Definition: UPstream.H:77
error FatalError
Error stream (stdout output on all processes), with additional &#39;FOAM FATAL ERROR&#39; header text and sta...
#define FatalErrorInFunction
Report an error message using Foam::FatalError.
Definition: error.H:600
bool empty() const noexcept
True if List is empty (ie, size() is zero)
Definition: UList.H:697
T * data() noexcept
Return pointer to the underlying array serving as data storage.
Definition: UListI.H:265
tmp< DimensionedField< TypeR, GeoMesh > > New(const tmp< DimensionedField< TypeR, GeoMesh >> &tf1, const word &name, const dimensionSet &dimensions, const bool initCopy=false)
Global function forwards to reuseTmpDimensionedField::New.
static std::streamsize read(const UPstream::commsTypes commsType, const int fromProcNo, Type *buffer, std::streamsize count, const int tag=UPstream::msgType(), const int communicator=UPstream::worldComm, UPstream::Request *req=nullptr)
Receive buffer contents (contiguous types) from given processor.
"scheduled" (MPI standard) : (MPI_Send, MPI_Recv)
static bool write(const UPstream::commsTypes commsType, const int toProcNo, const Type *buffer, std::streamsize count, const int tag=UPstream::msgType(), const int communicator=UPstream::worldComm, UPstream::Request *req=nullptr, const UPstream::sendModes sendMode=UPstream::sendModes::normal)
Write buffer contents (contiguous types only) to given processor.
virtual label comm() const =0
Return communicator used for parallel communication.
virtual int tag() const =0
Return message tag used for sending.
errorManip< error > abort(error &err)
Definition: errorManip.H:139
labelList f(nPoints)
T & last()
Access last element of the list, position [size()-1].
Definition: UList.H:900
gmvFile<< "tracers "<< particles.size()<< nl;for(const passiveParticle &p :particles){ gmvFile<< p.position().x()<< ' ';}gmvFile<< nl;for(const passiveParticle &p :particles){ gmvFile<< p.position().y()<< ' ';}gmvFile<< nl;for(const passiveParticle &p :particles){ gmvFile<< p.position().z()<< ' ';}gmvFile<< nl;for(const word &name :lagrangianScalarNames){ IOField< scalar > fld(IOobject(name, runTime.timeName(), cloud::prefix, mesh, IOobject::MUST_READ, IOobject::NO_WRITE))
void compressedReceive(const UPstream::commsTypes commsType, UList< Type > &f) const
Raw receive function with data compression.
virtual int neighbProcNo() const =0
Return neighbour processor number (rank in communicator)
"nonBlocking" (immediate) : (MPI_Isend, MPI_Irecv)
const T * cdata() const noexcept
Return pointer to the underlying array serving as data storage.
Definition: UListI.H:258
A class for managing temporary objects.
Definition: HashPtrTable.H:50
"buffered" : (MPI_Bsend, MPI_Recv)
void compressedSend(const UPstream::commsTypes commsType, const UList< Type > &f) const
Raw send function with data compression.
void receive(const UPstream::commsTypes commsType, UList< Type > &f) const
Raw receive function.