Changeset 676 for XIOS/trunk/src/node/grid.cpp

Timestamp:

08/25/15 16:52:45 (9 years ago)

Author:

rlacroix

Message:

Add support for indexed output.

If the new field attribute "indexed_output" is set to true and a mask is defined (either at grid, domain or axis level), the indexed data will be outputed instead of the full data with missing values.

See ​http://cfconventions.org/Data/cf-conventions/cf-conventions-1.5/build/cf-conventions.html#compression-by-gathering for more information.

File:

• XIOS/trunk/src/node/grid.cpp (modified) (13 diffs)

XIOS/trunk/src/node/grid.cpp

@@ -25 +25 @@
       : CObjectTemplate<CGrid>(), CGridAttributes()
       , isChecked(false), isDomainAxisChecked(false)
-      , vDomainGroup_(), vAxisGroup_(), axisList_(), isAxisListSet(false), isDomListSet(false), clientDistribution_(0), isIndexSent(false)
-      , serverDistribution_(0), clientServerMap_(0), writtenDataSize_(0), globalDim_()
-      , connectedDataSize_(), connectedServerRank_(), isDataDistributed_(true), transformations_(0), isTransformed_(false)
+      , vDomainGroup_(), vAxisGroup_(), axisList_(), isAxisListSet(false), isDomListSet(false)
+      , clientDistribution_(0), isIndexSent(false) , serverDistribution_(0), clientServerMap_(0)
+      , writtenDataSize_(0), numberWrittenIndexes_(0), totalNumberWrittenIndexes_(0), offsetWrittenIndexes_(0)
+      , globalDim_(), connectedDataSize_(), connectedServerRank_(), isDataDistributed_(true), isCompressible_(false)
+      , transformations_(0), isTransformed_(false)
       , axisPositionInGrid_(), positionDimensionDistributed_(1)
    {
@@ -37 +39 @@
       : CObjectTemplate<CGrid>(id), CGridAttributes()
       , isChecked(false), isDomainAxisChecked(false)
-      , vDomainGroup_(), vAxisGroup_(), axisList_(), isAxisListSet(false), isDomListSet(false), clientDistribution_(0), isIndexSent(false)
-      , serverDistribution_(0), clientServerMap_(0), writtenDataSize_(0), globalDim_()
-      , connectedDataSize_(), connectedServerRank_(), isDataDistributed_(true), transformations_(0), isTransformed_(false)
+      , vDomainGroup_(), vAxisGroup_(), axisList_(), isAxisListSet(false), isDomListSet(false)
+      , clientDistribution_(0), isIndexSent(false) , serverDistribution_(0), clientServerMap_(0)
+      , writtenDataSize_(0), numberWrittenIndexes_(0), totalNumberWrittenIndexes_(0), offsetWrittenIndexes_(0)
+      , globalDim_(), connectedDataSize_(), connectedServerRank_(), isDataDistributed_(true), isCompressible_(false)
+      , transformations_(0), isTransformed_(false)
       , axisPositionInGrid_(), positionDimensionDistributed_(1)
    {
@@ -149 +153 @@
    }
 
+   //---------------------------------------------------------------
+
+   /*!
+    * Test whether the data defined on the grid can be outputted in a compressed way.
+    * 
+    * \return true if and only if a mask was defined for this grid
+    */
+   bool CGrid::isCompressible(void) const
+   {
+      return isCompressible_;
+   }
+
+   //---------------------------------------------------------------
+
+   void CGrid::addRelFileCompressed(const StdString& filename)
+   {
+      this->relFilesCompressed.insert(filename);
+   }
+
+   bool CGrid::isWrittenCompressed(const StdString& filename) const
+   {
+      return (this->relFilesCompressed.find(filename) != this->relFilesCompressed.end());
+   }
+
+   //---------------------------------------------------------------
+
    void CGrid::solveDomainAxisRef(bool areAttributesChecked)
    {
@@ -157 +187 @@
      computeGridGlobalDimension(getDomains(), getAxis(), axis_domain_order);
      this->isDomainAxisChecked = areAttributesChecked;
+   }
+
+   void CGrid::checkEligibilityForCompressedOutput()
+   {
+     // We don't check if the mask is valid here, just if a mask has been defined at this point.
+     isCompressible_ = !mask1.isEmpty() || !mask2.isEmpty() || !mask3.isEmpty();
    }
 
@@ -488 +524 @@
    }
 
+   void CGrid::outputCompressedField(int rank, const CArray<double,1>& stored, double* field)
+   {
+     const CArray<size_t,1>& out_i = compressedOutIndexFromClient[rank];
+     StdSize numElements = stored.numElements();
+     for (StdSize n = 0; n < numElements; ++n)
+     {
+       field[out_i(n)] = stored(n);
+     }
+   }
+
    //----------------------------------------------------------------
 
@@ -518 +564 @@
     }
     isDataDistributed_ = false;
+  }
+
+  void CGrid::computeCompressedIndex()
+  {
+    compressedOutIndexFromClient = outIndexFromClient;
+
+    std::map<size_t, size_t> indexes;
+
+    {
+      std::map<int, CArray<size_t,1> >::const_iterator it = compressedOutIndexFromClient.begin();
+      std::map<int, CArray<size_t,1> >::const_iterator itEnd = compressedOutIndexFromClient.end();
+      for (; it != itEnd; ++it)
+      {
+        for (int i = 0; i < it->second.numElements(); ++i)
+          indexes.insert(std::make_pair(it->second(i), 0));
+      }
+    }
+
+    {
+      std::map<size_t, size_t>::iterator it = indexes.begin();
+      std::map<size_t, size_t>::iterator itEnd = indexes.end();
+      for (size_t i = 0; it != itEnd; ++it, ++i)
+        it->second = i;
+    }
+
+    {
+      std::map<int, CArray<size_t,1> >::iterator it = compressedOutIndexFromClient.begin();
+      std::map<int, CArray<size_t,1> >::iterator itEnd = compressedOutIndexFromClient.end();
+      for (; it != itEnd; ++it)
+      {
+        for (int i = 0; i < it->second.numElements(); ++i)
+          it->second(i) = indexes[it->second(i)];
+      }
+    }
   }
 
@@ -547 +627 @@
 
         listMsg.push_back(CMessage());
-        listMsg.back() << getId( )<< isDataDistributed_ << listOutIndex.back();
+        listMsg.back() << getId( )<< isDataDistributed_ << isCompressible_ << listOutIndex.back();
 
         event.push(rank, 1, listMsg.back());
@@ -589 +669 @@
 
           listMsg.push_back(CMessage());
-          listMsg.back() << getId() << isDataDistributed_ << listOutIndex.back();
+          listMsg.back() << getId() << isDataDistributed_ << isCompressible_ << listOutIndex.back();
 
           event.push(rank, 1, listMsg.back());
@@ -650 +730 @@
 
         listMsg.push_back(CMessage());
-        listMsg.back() << getId() << isDataDistributed_ << listOutIndex.back();
+        listMsg.back() << getId() << isDataDistributed_ << isCompressible_ << listOutIndex.back();
 
         event.push(rank, nbSenders[rank], listMsg.back());
@@ -741 +821 @@
   }
 
+  /*!
+    Returns the number of indexes written by each server.
+    \return the number of indexes written by each server
+  */
+  int CGrid::getNumberWrittenIndexes() const
+  {
+    return numberWrittenIndexes_;
+  }
+
+  /*!
+    Returns the total number of indexes written by the servers.
+    \return the total number of indexes written by the servers
+  */
+  int CGrid::getTotalNumberWrittenIndexes() const
+  {
+    return totalNumberWrittenIndexes_;
+  }
+
+  /*!
+    Returns the offset of indexes written by each server.
+    \return the offset of indexes written by each server
+  */
+  int CGrid::getOffsetWrittenIndexes() const
+  {
+    return offsetWrittenIndexes_;
+  }
+
   const CDistributionServer* CGrid::getDistributionServer() const
   {
@@ -762 +869 @@
     CContext* context = CContext::getCurrent();
     CContextServer* server = context->server;
+    numberWrittenIndexes_ = totalNumberWrittenIndexes_ = offsetWrittenIndexes_ = 0;
 
     for (int n = 0; n < ranks.size(); n++)
@@ -768 +876 @@
       CBufferIn& buffer = *buffers[n];
 
-      buffer >> isDataDistributed_;
+      buffer >> isDataDistributed_ >> isCompressible_;
       size_t dataSize = 0;
 
       if (isScalarGrid())
       {
-        writtenDataSize_ = 1;
+        writtenDataSize_ = numberWrittenIndexes_ = totalNumberWrittenIndexes_ = 1;
         CArray<size_t,1> outIndex;
         buffer >> outIndex;
@@ -849 +957 @@
       outIndexFromClient.insert(std::make_pair(rank, outIndex));
       connectedDataSize_[rank] = outIndex.numElements();
-    }
+      numberWrittenIndexes_ += outIndex.numElements();
+    }
+
+    if (isDataDistributed_)
+    {
+      MPI_Allreduce(&numberWrittenIndexes_, &totalNumberWrittenIndexes_, 1, MPI_INT, MPI_SUM, server->intraComm);
+      MPI_Scan(&numberWrittenIndexes_, &offsetWrittenIndexes_, 1, MPI_INT, MPI_SUM, server->intraComm);
+      offsetWrittenIndexes_ -= numberWrittenIndexes_;
+    }
+    else
+      totalNumberWrittenIndexes_ = numberWrittenIndexes_;
 
     nbSenders = CClientServerMappingDistributed::computeConnectedClients(context->client->serverSize, context->client->clientSize, context->client->intraComm, ranks);