src/OpenFOAM/db/IOstreams/Pstreams/PstreamBuffers.C

   1 /*---------------------------------------------------------------------------*\
   2   =========                 |
   3   \\      /  F ield         | OpenFOAM: The Open Source CFD Toolbox
   4    \\    /   O peration     |
   5     \\  /    A nd           | Copyright (C) 2004-2010 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
  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
  39 Foam::PstreamBuffers::PstreamBuffers
  40 (
  41     const UPstream::commsTypes commsType,
  42     const int tag,
  43     IOstream::streamFormat format,
  44     IOstream::versionNumber version
  45 )
  46 :
  47     commsType_(commsType),
  48     tag_(tag),
  49     format_(format),
  50     version_(version),
  51     sendBuf_(UPstream::nProcs()),
  52     recvBuf_(UPstream::nProcs()),
  53     recvBufPos_(UPstream::nProcs(),  0),
  54     finishedSendsCalled_(false)
  55 {}
  56
  57
  58 // * * * * * * * * * * * * * * * * Destructor  * * * * * * * * * * * * * * * //
  59
  60 Foam::PstreamBuffers::~PstreamBuffers()
  61 {
  62     // Check that all data has been consumed.
  63     forAll(recvBufPos_, procI)
  64     {
  65         if (recvBufPos_[procI] < recvBuf_[procI].size())
  66         {
  67             FatalErrorIn("PstreamBuffers::~PstreamBuffers()")
  68                 << "Message from processor " << procI
  69                 << " not fully consumed. messageSize:" << recvBuf_[procI].size()
  70                 << " bytes of which only " << recvBufPos_[procI]
  71                 << " consumed."
  72                 << Foam::abort(FatalError);
  73         }
  74     }
  75 }
  76
  77
  78 // * * * * * * * * * * * * * * Member Functions  * * * * * * * * * * * * * * //
  79
  80 void Foam::PstreamBuffers::finishedSends(const bool block)
  81 {
  82     finishedSendsCalled_ = true;
  83
  84     if (commsType_ == UPstream::nonBlocking)
  85     {
  86         labelListList sizes;
  87         Pstream::exchange<DynamicList<char>, char>
  88         (
  89             sendBuf_,
  90             recvBuf_,
  91             sizes,
  92             tag_,
  93             block
  94         );
  95     }
  96 }
  97
  98
  99 void Foam::PstreamBuffers::finishedSends(labelListList& sizes, const bool block)
 100 {
 101     finishedSendsCalled_ = true;
 102
 103     if (commsType_ == UPstream::nonBlocking)
 104     {
 105         Pstream::exchange<DynamicList<char>, char>
 106         (
 107             sendBuf_,
 108             recvBuf_,
 109             sizes,
 110             tag_,
 111             block
 112         );
 113     }
 114     else
 115     {
 116         FatalErrorIn
 117         (
 118             "PstreamBuffers::finishedSends(labelListList&, const bool)"
 119         )   << "Obtaining sizes not supported in "
 120             << UPstream::commsTypeNames[commsType_] << endl
 121             << " since transfers already in progress. Use non-blocking instead."
 122             << exit(FatalError);
 123
 124         // Note: possible only if using different tag from write started
 125         // by ~UOPstream. Needs some work.
 126         //sizes.setSize(UPstream::nProcs());
 127         //labelList& nsTransPs = sizes[UPstream::myProcNo()];
 128         //nsTransPs.setSize(UPstream::nProcs());
 129         //
 130         //forAll(sendBuf_, procI)
 131         //{
 132         //    nsTransPs[procI] = sendBuf_[procI].size();
 133         //}
 134         //
 135         //// Send sizes across.
 136         //int oldTag = UPstream::msgType();
 137         //UPstream::msgType() = tag_;
 138         //combineReduce(sizes, UPstream::listEq());
 139         //UPstream::msgType() = oldTag;
 140     }
 141 }
 142
 143
 144 // ************************************************************************* //