Changeset 676 for XIOS/trunk/src/node/domain.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/domain.cpp (modified) (21 diffs)

XIOS/trunk/src/node/domain.cpp

@@ -13 +13 @@
 #include "context.hpp"
 #include "context_client.hpp"
+#include "context_server.hpp"
 #include "array_new.hpp"
+#include "distribution_client.hpp"
 #include "server_distribution_description.hpp"
 #include "client_server_mapping_distributed.hpp"
@@ -19 +21 @@
 #include "interpolate_from_file_domain.hpp"
 
+#include <algorithm>
+
 namespace xios {
 
@@ -26 +30 @@
       : CObjectTemplate<CDomain>(), CDomainAttributes()
       , isChecked(false), relFiles(), isClientChecked(false), nbConnectedClients_(), indSrv_(), connectedServerRank_()
-      , hasBounds(false), hasArea(false), isDistributed_(false), nGlobDomain_(), isUnstructed_(false)
+      , hasBounds(false), hasArea(false), isDistributed_(false), nGlobDomain_(), isCompressible_(false), isUnstructed_(false)
       , global_zoom_ni(0), global_zoom_ibegin(0), global_zoom_nj(0), global_zoom_jbegin(0)
       , isClientAfterTransformationChecked(false), hasLonLat(false)
@@ -35 +39 @@
       : CObjectTemplate<CDomain>(id), CDomainAttributes()
       , isChecked(false), relFiles(), isClientChecked(false), nbConnectedClients_(), indSrv_(), connectedServerRank_()
-      , hasBounds(false), hasArea(false), isDistributed_(false), nGlobDomain_(), isUnstructed_(false)
+      , hasBounds(false), hasArea(false), isDistributed_(false), nGlobDomain_(), isCompressible_(false), isUnstructed_(false)
       , global_zoom_ni(0), global_zoom_ibegin(0), global_zoom_nj(0), global_zoom_jbegin(0)
       , isClientAfterTransformationChecked(false), hasLonLat(false)
@@ -64 +68 @@
 
    //----------------------------------------------------------------
+
+   const std::vector<int>& CDomain::getIndexesToWrite(void) const
+   {
+     return indexesToWrite;
+   }
+
+   /*!
+     Returns the number of indexes written by each server.
+     \return the number of indexes written by each server
+   */
+   int CDomain::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 CDomain::getTotalNumberWrittenIndexes() const
+   {
+     return totalNumberWrittenIndexes_;
+   }
+
+   /*!
+     Returns the offset of indexes written by each server.
+     \return the offset of indexes written by each server
+   */
+   int CDomain::getOffsetWrittenIndexes() const
+   {
+     return offsetWrittenIndexes_;
+   }
+
+   //----------------------------------------------------------------
+
    bool CDomain::isEmpty(void) const
    {
@@ -71 +110 @@
 
    //----------------------------------------------------------------
+
    bool CDomain::IsWritten(const StdString & filename) const
    {
@@ -76 +116 @@
    }
 
+   bool CDomain::isWrittenCompressed(const StdString& filename) const
+   {
+      return (this->relFilesCompressed.find(filename) != this->relFilesCompressed.end());
+   }
+
    //----------------------------------------------------------------
+
    bool CDomain::isDistributed(void) const
    {
@@ -83 +129 @@
 
    //----------------------------------------------------------------
+
+   /*!
+    * Test whether the data defined on the domain can be outputted in a compressed way.
+    * 
+    * \return true if and only if a mask was defined for this domain
+    */
+   bool CDomain::isCompressible(void) const
+   {
+      return isCompressible_;
+   }
+
+   //----------------------------------------------------------------
+
    void CDomain::addRelFile(const StdString & filename)
    {
       this->relFiles.insert(filename);
+   }
+
+   void CDomain::addRelFileCompressed(const StdString& filename)
+   {
+      this->relFilesCompressed.insert(filename);
    }
 
@@ -246 +310 @@
    }
 
-
    //----------------------------------------------------------------
+
    void CDomain::checkDomainData(void)
    {
@@ -357 +421 @@
 
    //----------------------------------------------------------------
+
+   void CDomain::checkEligibilityForCompressedOutput(void)
+   {
+     // We don't check if the mask or the indexes are valid here, just if they have been defined at this point.
+     isCompressible_ = !mask_1d.isEmpty() || !mask_2d.isEmpty() || !data_i_index.isEmpty();
+   }
+
+   //----------------------------------------------------------------
+
    void CDomain::completeLonLatClient(void)
    {
@@ -746 +819 @@
         msg << ni_srv << ibegin_srv << iend_srv << nj_srv << jbegin_srv << jend_srv;
         msg << global_zoom_ni << global_zoom_ibegin << global_zoom_nj << global_zoom_jbegin;
+        msg << isCompressible_;
 
         event.push(*itRank,1,msg);
@@ -783 +857 @@
       {
         ++globalIndexCountZoom;
+      }
+    }
+
+    int globalIndexWrittenCount = 0;
+    if (isCompressible_)
+    {
+      for (i = 0; i < data_i_index.numElements(); ++i)
+      {
+        i_ind = CDistributionClient::getDomainIndex(data_i_index(i), data_j_index(i),
+                                                    data_ibegin, data_jbegin, data_dim, ni,
+                                                    j_ind);
+        if (i_ind >= 0 && i_ind < ni && j_ind >= 0 && j_ind < nj && mask_1d(i_ind + j_ind * ni))
+        {
+          i_ind += ibegin;
+          j_ind += jbegin;
+          if (i_ind >= global_zoom_ibegin && i_ind <= global_zoom_iend && j_ind >= global_zoom_jbegin && j_ind <= global_zoom_jend)
+            ++globalIndexWrittenCount;
+        }
       }
     }
@@ -809 +901 @@
     }
 
-     size_t globalSizeIndex = 1, indexBegin, indexEnd;
-     int range, clientSize = client->clientSize;
-     for (int i = 0; i < nGlobDomain_.size(); ++i) globalSizeIndex *= nGlobDomain_[i];
-     indexBegin = 0;
-     for (int i = 0; i < clientSize; ++i)
-     {
-       range = globalSizeIndex / clientSize;
-       if (i < (globalSizeIndex%clientSize)) ++range;
-       if (i == client->clientRank) break;
-       indexBegin += range;
-     }
-     indexEnd = indexBegin + range - 1;
+    CArray<int,1> globalIndexWrittenDomain(globalIndexWrittenCount);
+    if (isCompressible_)
+    {
+      globalIndexWrittenCount = 0;
+      for (i = 0; i < data_i_index.numElements(); ++i)
+      {
+        i_ind = CDistributionClient::getDomainIndex(data_i_index(i), data_j_index(i),
+                                                    data_ibegin, data_jbegin, data_dim, ni,
+                                                    j_ind);
+        if (i_ind >= 0 && i_ind < ni && j_ind >= 0 && j_ind < nj && mask_1d(i_ind + j_ind * ni))
+        {
+          i_ind += ibegin;
+          j_ind += jbegin;
+          if (i_ind >= global_zoom_ibegin && i_ind <= global_zoom_iend && j_ind >= global_zoom_jbegin && j_ind <= global_zoom_jend)
+          {
+            globalIndexWrittenDomain(globalIndexWrittenCount) = i_ind + j_ind * ni_glo;
+            ++globalIndexWrittenCount;
+          }
+        }
+      }
+    }
+
+    size_t globalSizeIndex = 1, indexBegin, indexEnd;
+    int range, clientSize = client->clientSize;
+    for (int i = 0; i < nGlobDomain_.size(); ++i) globalSizeIndex *= nGlobDomain_[i];
+    indexBegin = 0;
+    for (int i = 0; i < clientSize; ++i)
+    {
+      range = globalSizeIndex / clientSize;
+      if (i < (globalSizeIndex%clientSize)) ++range;
+      if (i == client->clientRank) break;
+      indexBegin += range;
+    }
+    indexEnd = indexBegin + range - 1;
 
     CServerDistributionDescription serverDescription(nGlobDomain_);
@@ -837 +951 @@
 
     indSrv_.clear();
+    indWrittenSrv_.clear();
     for (; it != ite; ++it)
     {
@@ -853 +968 @@
         }
       }
+      for (int i = 0; i < globalIndexWrittenDomain.numElements(); ++i)
+      {
+        if (binSearch.search(permutIndex.begin(), permutIndex.end(), globalIndexWrittenDomain(i), itVec))
+        {
+          indWrittenSrv_[rank].push_back(globalIndexWrittenDomain(i));
+        }
+      }
     }
 
@@ -888 +1010 @@
 
     list<CMessage> list_msgsIndex;
-    list<CArray<int,1> > list_indi, list_indj;
+    list<CArray<int,1> > list_indi, list_indj, list_writtenInd;
 
     std::map<int, std::vector<size_t> >::const_iterator it, iteMap;
@@ -918 +1040 @@
       list_msgsIndex.back() << isCurvilinear;
       list_msgsIndex.back() << list_indi.back() << list_indj.back();
+
+      if (isCompressible_)
+      {
+        std::vector<int>& writtenIndSrc = indWrittenSrv_[rank];
+        list_writtenInd.push_back(CArray<int,1>(writtenIndSrc.size()));
+        CArray<int,1>& writtenInd = list_writtenInd.back();
+
+        for (n = 0; n < writtenInd.numElements(); ++n)
+          writtenInd(n) = writtenIndSrc[n];
+
+        list_msgsIndex.back() << writtenInd;
+      }
 
       eventIndex.push(rank, nbConnectedClients_[rank], list_msgsIndex.back());
@@ -1118 +1252 @@
   {
     buffer >> ni_srv >> ibegin_srv >> iend_srv >> nj_srv >> jbegin_srv >> jend_srv
-           >> global_zoom_ni >> global_zoom_ibegin >> global_zoom_nj >> global_zoom_jbegin;
+           >> global_zoom_ni >> global_zoom_ibegin >> global_zoom_nj >> global_zoom_jbegin
+           >> isCompressible_;
 
     int zoom_iend = global_zoom_ibegin + global_zoom_ni - 1;
@@ -1158 +1293 @@
   void CDomain::recvIndex(CEventServer& event)
   {
+    CDomain* domain;
+
     list<CEventServer::SSubEvent>::iterator it;
     for (it = event.subEvents.begin(); it != event.subEvents.end(); ++it)
@@ -1164 +1301 @@
       string domainId;
       *buffer >> domainId;
-      get(domainId)->recvIndex(it->rank, *buffer);
+      domain = get(domainId);
+      domain->recvIndex(it->rank, *buffer);
+    }
+
+    if (domain->isCompressible_)
+    {
+      std::sort(domain->indexesToWrite.begin(), domain->indexesToWrite.end());
+
+      CContextServer* server = CContext::getCurrent()->server;
+      domain->numberWrittenIndexes_ = domain->indexesToWrite.size();
+      MPI_Allreduce(&domain->numberWrittenIndexes_, &domain->totalNumberWrittenIndexes_, 1, MPI_INT, MPI_SUM, server->intraComm);
+      MPI_Scan(&domain->numberWrittenIndexes_, &domain->offsetWrittenIndexes_, 1, MPI_INT, MPI_SUM, server->intraComm);
+      domain->offsetWrittenIndexes_ -= domain->numberWrittenIndexes_;
     }
   }
@@ -1178 +1327 @@
     buffer >> type_int >> isCurvilinear >> indiSrv[rank] >> indjSrv[rank];
     type.setValue((type_attr::t_enum)type_int); // probleme des type enum avec les buffers : ToFix
+
+    if (isCompressible_)
+    {
+      CArray<int, 1> writtenIndexes;
+      buffer >> writtenIndexes;
+      indexesToWrite.reserve(indexesToWrite.size() + writtenIndexes.numElements());
+      for (int i = 0; i < writtenIndexes.numElements(); ++i)
+        indexesToWrite.push_back(writtenIndexes(i));
+    }
   }