src/Pstream/gamma/PstreamGlobals.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 \*---------------------------------------------------------------------------*/
  25
  26 #include "PstreamGlobals.H"
  27 #include "IOstreams.H"
  28 #include "Pstream.H"
  29
  30 extern "C" {
  31
  32 #include <linux/gamma/libgamma.h>
  33
  34 }
  35
  36 // * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
  37
  38 namespace Foam
  39 {
  40
  41 // * * * * * * * * * * * * * * * Member Functions  * * * * * * * * * * * * * //
  42
  43 // Receive buffers
  44 FixedList<List<List<char> >, 4> PstreamGlobals::recvBuf;
  45
  46 // Length of receive buffers
  47 FixedList<labelList, 4> PstreamGlobals::recvBufLen;
  48
  49 labelList PstreamGlobals::recvIndex;
  50 labelList PstreamGlobals::consumeIndex;
  51
  52 // These are all signalling nans and probably different from the ones that
  53 // the fpu might ever generate.
  54 uint64_t PstreamGlobals::resizeMessage[PstreamGlobals::resizeMessageLen] =
  55 {
  56     0x7ff7ffffffffffABllu,
  57     0x7ff7ffffffffffCDllu,
  58     0x7ff7ffffffffff12llu,
  59     0x7ff7ffffffffff30llu,
  60     0x7ff7ffffffffff19llu,
  61     0x0000000000000000llu       // this word gets overwritten with the length.
  62 };
  63
  64
  65 // Wrapper around gamma_wait
  66 void PstreamGlobals::gammaWait(const label procNo)
  67 {
  68     // Last request. Block.
  69     gamma_wait(procNo, 1);
  70
  71     // Currently unconsumed received message
  72     label ready = PstreamGlobals::consumeIndex[procNo];
  73
  74     // Check received length
  75     if (PstreamGlobals::recvBufLen[ready][procNo] == -2)
  76     {
  77         // Was resize message. Consume and rewait (is always followed by
  78         // real message)
  79
  80         if (Pstream::debug)
  81         {
  82             Pout<< "PstreamGlobals::gammaWait : "
  83                 << "Resize event. consumeIndex:" << ready
  84                 << " Restarting receive from " << procNo << endl;
  85         }
  86         // Consume resize message
  87         PstreamGlobals::recvBufLen[ready][procNo] = -1;
  88         PstreamGlobals::consumeIndex[procNo] =
  89             PstreamGlobals::recvBuf.fcIndex(ready);
  90         // And rewait
  91         gamma_wait(procNo, 1);
  92     }
  93 }
  94
  95
  96 // Copies data from global receive buffer into buf.
  97 label PstreamGlobals::copyReceive
  98 (
  99     const label procNo,
 100     char* buf,
 101     const label bufSize
 102 )
 103 {
 104     // Get the ready buffer
 105     label ready = consumeIndex[procNo];
 106
 107     // Actually received
 108     label receivedLen = recvBufLen[ready][procNo];
 109
 110     if (Pstream::debug)
 111     {
 112         Pout<< "copyReceive : for proc " << procNo
 113             << " copying " << receivedLen << " bytes out of buffer " << ready
 114             << endl;
 115     }
 116
 117     if (receivedLen < 0)
 118     {
 119         FatalErrorIn
 120         (
 121             "Pstream::copyReceive(const label, char*, const label)"
 122         )   << "Illegal message length "
 123             << receivedLen
 124             << " received from proc " << procNo << " into buffer " << ready
 125             << endl
 126             << "This is probably caused by receiving more than is actually"
 127             << " sent (e.g. gather without scatter)." << endl
 128             << abort(FatalError);
 129     }
 130
 131     if (receivedLen > bufSize)
 132     {
 133         FatalErrorIn
 134         (
 135             "Pstream::copyReceive(const label, char*, const label)"
 136         )   << "buffer ("
 137             << bufSize
 138             << ") not large enough for incomming message ("
 139             << receivedLen << ')'
 140             << " received from proc " << procNo << " into buffer " << ready
 141             << abort(FatalError);
 142     }
 143
 144     // Copy out of receive buffer
 145     memcpy
 146     (
 147         buf,
 148         recvBuf[ready][procNo].begin(),
 149         receivedLen
 150     );
 151     // Release receive buffer
 152     recvBufLen[ready][procNo] = -1;
 153     // Go to next buffer to consume
 154     consumeIndex[procNo] = recvBuf.fcIndex(ready);
 155
 156     return receivedLen;
 157 }
 158
 159
 160 // Checks whether an incoming message is a resize message. If not returns -1,
 161 // otherwise returns size read from header.
 162 label PstreamGlobals::getSizeFromHeader(const char* buf, const label len)
 163 {
 164     if (len != resizeMessageLen*sizeof(uint64_t))
 165     {
 166         return -1;
 167     }
 168
 169     const uint64_t* dPtr = reinterpret_cast<const uint64_t*>(buf);
 170
 171     // Check all but the last word
 172     for (label i = 0; i < resizeMessageLen-1; i++)
 173     {
 174         if (*dPtr++ != resizeMessage[i])
 175         {
 176             return -1;
 177         }
 178     }
 179
 180     return *reinterpret_cast<const label*>(dPtr);
 181 }
 182
 183
 184 void PstreamGlobals::setResizeMessage(const label len)
 185 {
 186     reinterpret_cast<label&>(resizeMessage[resizeMessageLen-1]) = len;
 187 }
 188
 189
 190 label PstreamGlobals::getMaxBufSize(const int procNo)
 191 {
 192     label maxSz = 0;
 193
 194     forAll(recvBuf, i)
 195     {
 196         maxSz = max(maxSz, recvBuf[i][procNo].size());
 197     }
 198     return maxSz;
 199 }
 200
 201
 202 // * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
 203
 204 } // End namespace Foam
 205
 206 // ************************************************************************* //