PstreamBuffers.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 | Copyright (C) 2011-2013 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  {
69  FatalErrorIn("PstreamBuffers::~PstreamBuffers()")
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::nonBlocking)
87  {
88  labelListList sizes;
89  Pstream::exchange<DynamicList<char>, char>
90  (
91  sendBuf_,
92  recvBuf_,
93  sizes,
94  tag_,
95  comm_,
96  block
97  );
98  }
99 }
100 
101 
103 {
104  finishedSendsCalled_ = true;
105 
106  if (commsType_ == UPstream::nonBlocking)
107  {
108  Pstream::exchange<DynamicList<char>, char>
109  (
110  sendBuf_,
111  recvBuf_,
112  sizes,
113  tag_,
114  comm_,
115  block
116  );
117  }
118  else
119  {
121  (
122  "PstreamBuffers::finishedSends(labelListList&, const bool)"
123  ) << "Obtaining sizes not supported in "
124  << UPstream::commsTypeNames[commsType_] << endl
125  << " since transfers already in progress. Use non-blocking instead."
126  << exit(FatalError);
127 
128  // Note: possible only if using different tag from write started
129  // by ~UOPstream. Needs some work.
130  //sizes.setSize(UPstream::nProcs(comm));
131  //labelList& nsTransPs = sizes[UPstream::myProcNo(comm)];
132  //nsTransPs.setSize(UPstream::nProcs(comm));
133  //
134  //forAll(sendBuf_, procI)
135  //{
136  // nsTransPs[procI] = sendBuf_[procI].size();
137  //}
138  //
140  //int oldTag = UPstream::msgType();
141  //UPstream::msgType() = tag_;
142  //combineReduce(sizes, UPstream::listEq());
143  //UPstream::msgType() = oldTag;
144  }
145 }
146 
147 
149 {
150  forAll(sendBuf_, i)
151  {
152  sendBuf_[i].clear();
153  }
154  forAll(recvBuf_, i)
155  {
156  recvBuf_[i].clear();
157  }
158  recvBufPos_ = 0;
159  finishedSendsCalled_ = false;
160 }
161 
162 
163 // ************************************************************************* //
Creates a single block of cells from point coordinates, numbers of cells in each direction and an exp...
Definition: block.H:63
intWM_LABEL_SIZE_t label
A label is an int32_t or int64_t as specified by the pre-processor macro WM_LABEL_SIZE.
Definition: label.H:59
errorManipArg< error, int > exit(error &err, const int errNo=1)
Definition: errorManip.H:124
Namespace for OpenFOAM.
static label nProcs(const label communicator=0)
Number of processes in parallel run.
Definition: UPstream.H:386
Ostream & endl(Ostream &os)
Add newline and flush stream.
Definition: Ostream.H:251
streamFormat
Enumeration for the format of data in the stream.
Definition: IOstream.H:86
commsTypes
Types of communications.
Definition: UPstream.H:64
#define forAll(list, i)
Definition: UList.H:421
static DynamicList< char > nullBuf
errorManip< error > abort(error &err)
Definition: errorManip.H:131
#define FatalErrorIn(functionName)
Report an error message using Foam::FatalError.
Definition: error.H:314
~PstreamBuffers()
Destructor.
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,.
static const NamedEnum< commsTypes, 3 > commsTypeNames
Definition: UPstream.H:71
void clear()
Clear storage and reset.
void finishedSends(const bool block=true)
Mark all sends as having been done. This will start receives.
Version number type.
Definition: IOstream.H:96