PstreamBuffers.C
Go to the documentation of this file.
1 /*---------------------------------------------------------------------------*\
2  ========= |
3  \\ / F ield | OpenFOAM: The Open Source CFD Toolbox
4  \\ / O peration | Website: https://openfoam.org
5  \\ / A nd | Copyright (C) 2011-2018 OpenFOAM Foundation
6  \\/ M anipulation |
7 -------------------------------------------------------------------------------
8 License
9  This file is part of OpenFOAM.
10 
11  OpenFOAM is free software: you can redistribute it and/or modify it
12  under the terms of the GNU General Public License as published by
13  the Free Software Foundation, either version 3 of the License, or
14  (at your option) any later version.
15 
16  OpenFOAM is distributed in the hope that it will be useful, but WITHOUT
17  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
18  FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
19  for more details.
20 
21  You should have received a copy of the GNU General Public License
22  along with OpenFOAM. If not, see <http://www.gnu.org/licenses/>.
23 
24 \*---------------------------------------------------------------------------*/
25 
26 #include "PstreamBuffers.H"
27 
28 /* * * * * * * * * * * * * * * Static Member Data * * * * * * * * * * * * * */
29 
30 namespace Foam
31 {
32 
33  DynamicList<char> PstreamBuffers::nullBuf(0);
34 }
35 
36 
37 // * * * * * * * * * * * * * * * * Constructor * * * * * * * * * * * * * * * //
38 
40 (
41  const UPstream::commsTypes commsType,
42  const int tag,
43  const label comm,
46 )
47 :
48  commsType_(commsType),
49  tag_(tag),
50  comm_(comm),
51  format_(format),
52  version_(version),
53  sendBuf_(UPstream::nProcs(comm)),
54  recvBuf_(UPstream::nProcs(comm)),
55  recvBufPos_(UPstream::nProcs(comm), 0),
56  finishedSendsCalled_(false)
57 {}
58 
59 
60 // * * * * * * * * * * * * * * * * Destructor * * * * * * * * * * * * * * * //
61 
63 {
64  // Check that all data has been consumed.
65  forAll(recvBufPos_, proci)
66  {
67  if (recvBufPos_[proci] < recvBuf_[proci].size())
68  {
70  << "Message from processor " << proci
71  << " not fully consumed. messageSize:" << recvBuf_[proci].size()
72  << " bytes of which only " << recvBufPos_[proci]
73  << " consumed."
75  }
76  }
77 }
78 
79 
80 // * * * * * * * * * * * * * * Member Functions * * * * * * * * * * * * * * //
81 
83 {
84  finishedSendsCalled_ = true;
85 
86  if (commsType_ == UPstream::commsTypes::nonBlocking)
87  {
88  Pstream::exchange<DynamicList<char>, char>
89  (
90  sendBuf_,
91  recvBuf_,
92  tag_,
93  comm_,
94  block
95  );
96  }
97 }
98 
99 
101 {
102  finishedSendsCalled_ = true;
103 
104  if (commsType_ == UPstream::commsTypes::nonBlocking)
105  {
106  Pstream::exchangeSizes(sendBuf_, recvSizes, comm_);
107 
108  Pstream::exchange<DynamicList<char>, char>
109  (
110  sendBuf_,
111  recvSizes,
112  recvBuf_,
113  tag_,
114  comm_,
115  block
116  );
117  }
118  else
119  {
121  << "Obtaining sizes not supported in "
122  << UPstream::commsTypeNames[commsType_] << endl
123  << " since transfers already in progress. Use non-blocking instead."
124  << exit(FatalError);
125 
126  // Note: maybe possible only if using different tag from write started
127  // by ~UOPstream. Needs some work.
128  }
129 }
130 
131 
133 {
134  forAll(sendBuf_, i)
135  {
136  sendBuf_[i].clear();
137  }
138  forAll(recvBuf_, i)
139  {
140  recvBuf_[i].clear();
141  }
142  recvBufPos_ = 0;
143  finishedSendsCalled_ = false;
144 }
145 
146 
147 // ************************************************************************* //
#define forAll(list, i)
Loop across all elements in list.
Definition: UList.H:434
void finishedSends(const bool block=true)
Mark all sends as having been done. This will start receives.
errorManipArg< error, int > exit(error &err, const int errNo=1)
Definition: errorManip.H:124
error FatalError
PstreamBuffers(const UPstream::commsTypes commsType, const int tag=UPstream::msgType(), const label comm=UPstream::worldComm, IOstream::streamFormat format=IOstream::BINARY, IOstream::versionNumber version=IOstream::currentVersion)
Construct given comms type,.
#define FatalErrorInFunction
Report an error message using Foam::FatalError.
Definition: error.H:306
commsTypes
Types of communications.
Definition: UPstream.H:64
Ostream & endl(Ostream &os)
Add newline and flush stream.
Definition: Ostream.H:251
static void exchangeSizes(const Container &sendData, labelList &sizes, const label comm=UPstream::worldComm)
Helper: exchange sizes of sendData. sendData is the data per.
Definition: exchange.C:137
static DynamicList< char > nullBuf
streamFormat
Enumeration for the format of data in the stream.
Definition: IOstream.H:86
errorManip< error > abort(error &err)
Definition: errorManip.H:131
Creates a single block of cells from point coordinates, numbers of cells in each direction and an exp...
Definition: block.H:63
~PstreamBuffers()
Destructor.
static const NamedEnum< commsTypes, 3 > commsTypeNames
Definition: UPstream.H:71
void clear()
Clear storage and reset.
static label nProcs(const label communicator=0)
Number of processes in parallel run.
Definition: UPstream.H:411
Version number type.
Definition: IOstream.H:96
Namespace for OpenFOAM.