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

   1 /*---------------------------------------------------------------------------*\
   2   =========                 |
   3   \\      /  F ield         | OpenFOAM: The Open Source CFD Toolbox
   4    \\    /   O peration     |
   5     \\  /    A nd           | Copyright (C) 2011 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 Description
  25     Exchange data.
  26
  27 \*---------------------------------------------------------------------------*/
  28
  29 #include "Pstream.H"
  30 #include "contiguous.H"
  31 #include "PstreamCombineReduceOps.H"
  32 #include "UPstream.H"
  33
  34 // * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
  35
  36 namespace Foam
  37 {
  38
  39 // * * * * * * * * * * * * * * * Member Functions  * * * * * * * * * * * * * //
  40
  41 //template <template<class> class ListType, class T>
  42 template <class Container, class T>
  43 void Pstream::exchange
  44 (
  45     const List<Container >& sendBufs,
  46     List<Container >& recvBufs,
  47     labelListList& sizes,
  48     const int tag,
  49     const bool block
  50 )
  51 {
  52     if (!contiguous<T>())
  53     {
  54         FatalErrorIn
  55         (
  56             "Pstream::exchange(..)"
  57         )   << "Continuous data only." << Foam::abort(FatalError);
  58     }
  59
  60     if (sendBufs.size() != UPstream::nProcs())
  61     {
  62         FatalErrorIn
  63         (
  64             "Pstream::exchange(..)"
  65         )   << "Size of list:" << sendBufs.size()
  66             << " does not equal the number of processors:"
  67             << UPstream::nProcs()
  68             << Foam::abort(FatalError);
  69     }
  70
  71     sizes.setSize(UPstream::nProcs());
  72     labelList& nsTransPs = sizes[UPstream::myProcNo()];
  73     nsTransPs.setSize(UPstream::nProcs());
  74
  75     forAll(sendBufs, procI)
  76     {
  77         nsTransPs[procI] = sendBufs[procI].size();
  78     }
  79
  80     // Send sizes across.
  81     int oldTag = UPstream::msgType();
  82     UPstream::msgType() = tag;
  83     combineReduce(sizes, UPstream::listEq());
  84     UPstream::msgType() = oldTag;
  85
  86     if (Pstream::parRun())
  87     {
  88         // Set up receives
  89         // ~~~~~~~~~~~~~~~
  90
  91         recvBufs.setSize(sendBufs.size());
  92         forAll(sizes, procI)
  93         {
  94             label nRecv = sizes[procI][UPstream::myProcNo()];
  95
  96             if (procI != Pstream::myProcNo() && nRecv > 0)
  97             {
  98                 recvBufs[procI].setSize(nRecv);
  99                 UIPstream::read
 100                 (
 101                     UPstream::nonBlocking,
 102                     procI,
 103                     reinterpret_cast<char*>(recvBufs[procI].begin()),
 104                     nRecv*sizeof(T),
 105                     tag
 106                 );
 107             }
 108         }
 109
 110
 111         // Set up sends
 112         // ~~~~~~~~~~~~
 113
 114         forAll(sendBufs, procI)
 115         {
 116             if (procI != Pstream::myProcNo() && sendBufs[procI].size() > 0)
 117             {
 118                 if
 119                 (
 120                    !UOPstream::write
 121                     (
 122                         UPstream::nonBlocking,
 123                         procI,
 124                         reinterpret_cast<const char*>(sendBufs[procI].begin()),
 125                         sendBufs[procI].size()*sizeof(T),
 126                         tag
 127                     )
 128                 )
 129                 {
 130                     FatalErrorIn("Pstream::exchange(..)")
 131                         << "Cannot send outgoing message. "
 132                         << "to:" << procI << " nBytes:"
 133                         << label(sendBufs[procI].size()*sizeof(T))
 134                         << Foam::abort(FatalError);
 135                 }
 136             }
 137         }
 138
 139
 140         // Wait for all to finish
 141         // ~~~~~~~~~~~~~~~~~~~~~~
 142
 143         if (block)
 144         {
 145             Pstream::waitRequests();
 146         }
 147     }
 148
 149     // Do myself
 150     recvBufs[Pstream::myProcNo()] = sendBufs[Pstream::myProcNo()];
 151 }
 152
 153
 154 // * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
 155
 156 } // End namespace Foam
 157
 158 // ************************************************************************* //