src/Pstream/mpi/IPread.C

   1 /*---------------------------------------------------------------------------*\
   2   =========                 |
   3   \\      /  F ield         | OpenFOAM: The Open Source CFD Toolbox
   4    \\    /   O peration     |
   5     \\  /    A nd           | Copyright (C) 1991-2009 OpenCFD Ltd.
   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 the
  13     Free Software Foundation; either version 2 of the License, or (at your
  14     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, write to the Free Software Foundation,
  23     Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
  24
  25 Description
  26     Read token and binary block from IPstream
  27
  28 \*---------------------------------------------------------------------------*/
  29
  30 #include "mpi.h"
  31
  32 #include "IPstream.H"
  33
  34 // * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
  35
  36 // * * * * * * * * * * * * * * Static Data Members * * * * * * * * * * * * * //
  37
  38 // Outstanding non-blocking operations.
  39 //! @cond fileScope
  40 Foam::DynamicList<MPI_Request> IPstream_outstandingRequests_;
  41 //! @endcond fileScope
  42
  43 // * * * * * * * * * * * * * * * * Constructor * * * * * * * * * * * * * * * //
  44
  45 Foam::IPstream::IPstream
  46 (
  47     const commsTypes commsType,
  48     const int fromProcNo,
  49     const label bufSize,
  50     streamFormat format,
  51     versionNumber version
  52 )
  53 :
  54     Pstream(commsType, bufSize),
  55     Istream(format, version),
  56     fromProcNo_(fromProcNo),
  57     messageSize_(0)
  58 {
  59     setOpened();
  60     setGood();
  61
  62     MPI_Status status;
  63
  64     // If the buffer size is not specified, probe the incomming message
  65     // and set it
  66     if (!bufSize)
  67     {
  68         MPI_Probe(procID(fromProcNo_), msgType(), MPI_COMM_WORLD, &status);
  69         MPI_Get_count(&status, MPI_BYTE, &messageSize_);
  70
  71         buf_.setSize(messageSize_);
  72     }
  73
  74     messageSize_ = read(commsType, fromProcNo_, buf_.begin(), buf_.size());
  75
  76     if (!messageSize_)
  77     {
  78         FatalErrorIn
  79         (
  80             "IPstream::IPstream(const int fromProcNo, "
  81             "const label bufSize, streamFormat format, versionNumber version)"
  82         )   << "read failed"
  83             << Foam::abort(FatalError);
  84     }
  85 }
  86
  87
  88 // * * * * * * * * * * * * * * * Member Functions  * * * * * * * * * * * * * //
  89
  90 Foam::label Foam::IPstream::read
  91 (
  92     const commsTypes commsType,
  93     const int fromProcNo,
  94     char* buf,
  95     const std::streamsize bufSize
  96 )
  97 {
  98     if (commsType == blocking || commsType == scheduled)
  99     {
 100         MPI_Status status;
 101
 102         if
 103         (
 104             MPI_Recv
 105             (
 106                 buf,
 107                 bufSize,
 108                 MPI_PACKED,
 109                 procID(fromProcNo),
 110                 msgType(),
 111                 MPI_COMM_WORLD,
 112                 &status
 113             )
 114         )
 115         {
 116             FatalErrorIn
 117             (
 118                 "IPstream::read"
 119                 "(const int fromProcNo, char* buf, std::streamsize bufSize)"
 120             )   << "MPI_Recv cannot receive incomming message"
 121                 << Foam::abort(FatalError);
 122
 123             return 0;
 124         }
 125
 126
 127         // Check size of message read
 128
 129         label messageSize;
 130         MPI_Get_count(&status, MPI_BYTE, &messageSize);
 131
 132         if (messageSize > bufSize)
 133         {
 134             FatalErrorIn
 135             (
 136                 "IPstream::read"
 137                 "(const int fromProcNo, char* buf, std::streamsize bufSize)"
 138             )   << "buffer (" << label(bufSize)
 139                 << ") not large enough for incomming message ("
 140                 << messageSize << ')'
 141                 << Foam::abort(FatalError);
 142         }
 143
 144         return messageSize;
 145     }
 146     else if (commsType == nonBlocking)
 147     {
 148         MPI_Request request;
 149
 150         if
 151         (
 152             MPI_Irecv
 153             (
 154                 buf,
 155                 bufSize,
 156                 MPI_PACKED,
 157                 procID(fromProcNo),
 158                 msgType(),
 159                 MPI_COMM_WORLD,
 160                 &request
 161             )
 162         )
 163         {
 164             FatalErrorIn
 165             (
 166                 "IPstream::read"
 167                 "(const int fromProcNo, char* buf, std::streamsize bufSize)"
 168             )   << "MPI_Recv cannot start non-blocking receive"
 169                 << Foam::abort(FatalError);
 170
 171             return 0;
 172         }
 173
 174         IPstream_outstandingRequests_.append(request);
 175
 176         return 1;
 177     }
 178     else
 179     {
 180         FatalErrorIn
 181         (
 182             "IPstream::read"
 183             "(const int fromProcNo, char* buf, std::streamsize bufSize)"
 184         )   << "Unsupported communications type " << commsType
 185             << Foam::abort(FatalError);
 186
 187         return 0;
 188     }
 189 }
 190
 191
 192 void Foam::IPstream::waitRequests()
 193 {
 194     if (IPstream_outstandingRequests_.size())
 195     {
 196         if
 197         (
 198             MPI_Waitall
 199             (
 200                 IPstream_outstandingRequests_.size(),
 201                 IPstream_outstandingRequests_.begin(),
 202                 MPI_STATUSES_IGNORE
 203             )
 204         )
 205         {
 206             FatalErrorIn
 207             (
 208                 "IPstream::waitRequests()"
 209             )   << "MPI_Waitall returned with error" << endl;
 210         }
 211
 212         IPstream_outstandingRequests_.clear();
 213     }
 214 }
 215
 216
 217 bool Foam::IPstream::finishedRequest(const label i)
 218 {
 219     if (i >= IPstream_outstandingRequests_.size())
 220     {
 221         FatalErrorIn
 222         (
 223             "IPstream::finishedRequest(const label)"
 224         )   << "There are " << IPstream_outstandingRequests_.size()
 225             << " outstanding send requests and you are asking for i=" << i
 226             << nl
 227             << "Maybe you are mixing blocking/non-blocking comms?"
 228             << Foam::abort(FatalError);
 229     }
 230
 231     int flag;
 232     MPI_Test(&IPstream_outstandingRequests_[i], &flag, MPI_STATUS_IGNORE);
 233
 234     return flag != 0;
 235 }
 236
 237
 238 // * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
 239
 240 // ************************************************************************* //