Changeset 633 for XIOS/trunk/src/node/axis.cpp

Timestamp:

07/09/15 14:05:43 (9 years ago)

Author:

mhnguyen

Message:

Correcting value written on an distributed axis

+) Seperate writing value on axis in multiple file and one file case

Test
+) On curie
+) test_complet and test_client pass

File:

• XIOS/trunk/src/node/axis.cpp (modified) (9 diffs)

XIOS/trunk/src/node/axis.cpp

@@ -12 +12 @@
 #include "zoom_axis.hpp"
 #include "interpolate_axis.hpp"
+#include "server_distribution_description.hpp"
+#include "client_server_mapping_distributed.hpp"
 
 namespace xios {
@@ -20 +22 @@
       : CObjectTemplate<CAxis>()
       , CAxisAttributes(), isChecked(false), relFiles(), baseRefObject(), areClientAttributesChecked_(false)
-      , isDistributed_(false)
+      , isDistributed_(false), hasBounds_(false)
       , transformationMap_(), global_zoom_begin(0), global_zoom_size(0)
    {
@@ -28 +30 @@
       : CObjectTemplate<CAxis>(id)
       , CAxisAttributes(), isChecked(false), relFiles(), baseRefObject(), areClientAttributesChecked_(false)
-      , isDistributed_(false)
+      , isDistributed_(false), hasBounds_(false)
       , transformationMap_(), global_zoom_begin(0), global_zoom_size(0)
    {
@@ -99 +101 @@
       else this->ni.setValue(size);
 
-//      StdSize true_size = value.numElements();
-//      if (this->ni.getValue() != true_size)
-//         ERROR("CAxis::checkAttributes(void)",
-//               << "The array \'value\' of axis [ id = '" << getId() << "' , context = '" << CObjectFactory::GetCurrentContextId() << "' ] has a different size that the one defined by the \'size\' attribute");
-
+      StdSize true_size = value.numElements();
+      if (this->ni.getValue() != true_size)
+         ERROR("CAxis::checkAttributes(void)",
+               << "The array \'value\' of axis [ id = '" << getId() << "' , context = '" << CObjectFactory::GetCurrentContextId() << "' ] has a different size that the one defined by the \'size\' attribute");
 
       this->checkData();
+      this->checkZoom();
       this->checkMask();
-      this->checkZoom();
-
-      if (!bounds.isEmpty())
-      {
-        if (bounds.extent(0) != size || bounds.extent(1) != 2)
-            ERROR("CAxis::checkAttributes(void)",
-                  << "The bounds array of the axis [ id = '" << getId() << "' , context = '" << CObjectFactory::GetCurrentContextId() << "' ] must be of dimension axis size x 2" << endl
-                  << "Axis size is " << size << endl
-                  << "Bounds size is "<< bounds.extent(0) << " x " << bounds.extent(1));
-      }
+      this->checkBounds();
    }
 
@@ -185 +178 @@
    }
 
+  void CAxis::checkBounds()
+  {
+    if (!bounds.isEmpty())
+    {
+      if (bounds.extent(0) != ni || bounds.extent(1) != 2)
+          ERROR("CAxis::checkAttributes(void)",
+                << "The bounds array of the axis [ id = '" << getId() << "' , context = '" << CObjectFactory::GetCurrentContextId() << "' ] must be of dimension axis size x 2" << endl
+                << "Axis size is " << ni << endl
+                << "Bounds size is "<< bounds.extent(0) << " x " << bounds.extent(1));
+      hasBounds_ = true;
+    }
+    else hasBounds_ = false;
+  }
+
+
   bool CAxis::dispatchEvent(CEventServer& event)
    {
@@ -196 +204 @@
              return true;
              break;
+           case EVENT_ID_INDEX:
+            recvIndex(event);
+            return true;
+            break;
+          case EVENT_ID_DISTRIBUTED_VALUE:
+            recvDistributedValue(event);
+            return true;
+            break;
+          case EVENT_ID_NON_DISTRIBUTED_VALUE:
+            recvNonDistributedValue(event);
+            return true;
+            break;
            default :
              ERROR("bool CContext::dispatchEvent(CEventServer& event)",
@@ -227 +247 @@
      {
        sendServerAttribut(globalDim, orderPositionInGrid, distType);
+       sendValue();
      }
 
      this->isChecked = true;
    }
+
+  void CAxis::sendValue()
+  {
+     if (ni.getValue() == size.getValue())
+     {
+       sendNonDistributedValue();
+     }
+     else
+     {
+       computeConnectedServer();
+       sendDistributedValue();
+     }
+  }
+
+  void CAxis::computeConnectedServer()
+  {
+    CContext* context = CContext::getCurrent();
+    CContextClient* client = context->client;
+    int nbServer = client->serverSize;
+    int range, clientSize = client->clientSize;
+
+    CArray<size_t,1> globalIndexAxis(ni);
+    size_t ibegin = this->ibegin.getValue();
+    int zoom_end = global_zoom_begin+global_zoom_size-1;
+    std::vector<size_t> globalAxisZoom;
+    for (size_t idx = 0; idx < ni; ++idx)
+    {
+      size_t globalIndex = ibegin + idx;
+      globalIndexAxis(idx) = globalIndex;
+      if (globalIndex >= global_zoom_begin && globalIndex <= zoom_end) globalAxisZoom.push_back(globalIndex);
+    }
+
+    std::vector<int> nGlobDomain(1);
+    nGlobDomain[0] = size.getValue();
+
+    size_t globalSizeIndex = 1, indexBegin, indexEnd;
+    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);
+    serverDescription.computeServerGlobalIndexInRange(nbServer, std::make_pair<size_t,size_t>(indexBegin, indexEnd));
+    CClientServerMappingDistributed clientServerMap(serverDescription.getGlobalIndexRange(), client->intraComm);
+    clientServerMap.computeServerIndexMapping(globalIndexAxis);
+    const std::map<int, std::vector<size_t> >& globalIndexAxisOnServer = clientServerMap.getGlobalIndexOnServer();
+
+    std::map<int, std::vector<size_t> >::const_iterator it = globalIndexAxisOnServer.begin(),
+                                                       ite = globalIndexAxisOnServer.end();
+    std::vector<size_t>::const_iterator itbVec = (globalAxisZoom).begin(),
+                                        iteVec = (globalAxisZoom).end();
+    indSrv_.clear();
+    for (; it != ite; ++it)
+    {
+      int rank = it->first;
+      const std::vector<size_t>& globalIndexTmp = it->second;
+      int nb = globalIndexTmp.size();
+
+      for (int i = 0; i < nb; ++i)
+      {
+        if (std::binary_search(itbVec, iteVec, globalIndexTmp[i]))
+        {
+          indSrv_[rank].push_back(globalIndexTmp[i]);
+        }
+      }
+    }
+
+    connectedServerRank_.clear();
+    for (it = globalIndexAxisOnServer.begin(); it != ite; ++it) {
+      connectedServerRank_.push_back(it->first);
+    }
+
+    if (!indSrv_.empty())
+    {
+      connectedServerRank_.clear();
+      for (it = indSrv_.begin(); it != indSrv_.end(); ++it)
+        connectedServerRank_.push_back(it->first);
+    }
+    nbConnectedClients_ = clientServerMap.computeConnectedClients(client->serverSize, client->clientSize, client->intraComm, connectedServerRank_);
+
+  }
+
+  void CAxis::sendNonDistributedValue()
+  {
+    CContext* context = CContext::getCurrent();
+    CContextClient* client = context->client;
+    CEventClient event(getType(),EVENT_ID_NON_DISTRIBUTED_VALUE);
+
+    int zoom_end = global_zoom_begin+global_zoom_size-1;
+    int nb =0;
+    for (size_t idx = 0; idx < ni; ++idx)
+    {
+      size_t globalIndex = ibegin + idx;
+      if (globalIndex >= global_zoom_begin && globalIndex <= zoom_end) ++nb;
+    }
+
+    CArray<double,1> val(nb);
+    nb = 0;
+    for (size_t idx = 0; idx < ni; ++idx)
+    {
+      size_t globalIndex = ibegin + idx;
+      if (globalIndex >= global_zoom_begin && globalIndex <= zoom_end)
+      {
+        val(nb) = value(idx);
+        ++nb;
+      }
+    }
+
+    if (client->isServerLeader())
+    {
+      std::list<CMessage> msgs;
+
+      const std::list<int>& ranks = client->getRanksServerLeader();
+      for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
+      {
+        // Use const int to ensure CMessage holds a copy of the value instead of just a reference
+        msgs.push_back(CMessage());
+        CMessage& msg = msgs.back();
+        msg << this->getId();
+        msg << val;
+        event.push(*itRank,1,msg);
+      }
+      client->sendEvent(event);
+    }
+    else client->sendEvent(event);
+  }
+
+  void CAxis::sendDistributedValue(void)
+  {
+    int ns, n, i, j, ind, nv, idx;
+    CContext* context = CContext::getCurrent();
+    CContextClient* client=context->client;
+
+    // send value for each connected server
+    CEventClient eventIndex(getType(), EVENT_ID_INDEX);
+    CEventClient eventVal(getType(), EVENT_ID_DISTRIBUTED_VALUE);
+
+    list<CMessage> list_msgsIndex, list_msgsVal;
+    list<CArray<int,1> > list_indi;
+    list<CArray<double,1> > list_val;
+    list<CArray<double,2> > list_bounds;
+
+    std::map<int, std::vector<size_t> >::const_iterator it, iteMap;
+    iteMap = indSrv_.end();
+    for (int k = 0; k < connectedServerRank_.size(); ++k)
+    {
+      int nbData = 0;
+      int rank = connectedServerRank_[k];
+      it = indSrv_.find(rank);
+      if (iteMap != it)
+        nbData = it->second.size();
+
+      list_indi.push_back(CArray<int,1>(nbData));
+      list_val.push_back(CArray<double,1>(nbData));
+
+      if (hasBounds_)
+      {
+        list_bounds.push_back(CArray<double,2>(2,nbData));
+      }
+
+      CArray<int,1>& indi = list_indi.back();
+      CArray<double,1>& val = list_val.back();
+
+      for (n = 0; n < nbData; ++n)
+      {
+        idx = static_cast<int>(it->second[n]);
+        ind = idx - ibegin;
+
+        val(n) = value(ind);
+        indi(n) = idx;
+
+        if (hasBounds_)
+        {
+          CArray<double,2>& boundsVal = list_bounds.back();
+          boundsVal(0, n) = bounds(0,n);
+          boundsVal(1, n) = bounds(1,n);
+        }
+      }
+
+      list_msgsIndex.push_back(CMessage());
+      list_msgsIndex.back() << this->getId() << list_indi.back();
+
+      list_msgsVal.push_back(CMessage());
+      list_msgsVal.back() << this->getId() << list_val.back();
+
+      if (hasBounds_)
+      {
+        list_msgsVal.back() << list_bounds.back();
+      }
+
+      eventIndex.push(rank, nbConnectedClients_[rank], list_msgsIndex.back());
+      eventVal.push(rank, nbConnectedClients_[rank], list_msgsVal.back());
+    }
+
+    client->sendEvent(eventIndex);
+    client->sendEvent(eventVal);
+  }
+
+  void CAxis::recvIndex(CEventServer& event)
+  {
+    list<CEventServer::SSubEvent>::iterator it;
+    for (it = event.subEvents.begin(); it != event.subEvents.end(); ++it)
+    {
+      CBufferIn* buffer = it->buffer;
+      string domainId;
+      *buffer >> domainId;
+      get(domainId)->recvIndex(it->rank, *buffer);
+    }
+  }
+
+  void CAxis::recvIndex(int rank, CBufferIn& buffer)
+  {
+    buffer >> indiSrv_[rank];
+  }
+
+  void CAxis::recvDistributedValue(CEventServer& event)
+  {
+    list<CEventServer::SSubEvent>::iterator it;
+    for (it = event.subEvents.begin(); it != event.subEvents.end(); ++it)
+    {
+      CBufferIn* buffer = it->buffer;
+      string domainId;
+      *buffer >> domainId;
+      get(domainId)->recvDistributedValue(it->rank, *buffer);
+    }
+  }
+
+  void CAxis::recvDistributedValue(int rank, CBufferIn& buffer)
+  {
+    CArray<int,1> &indi = indiSrv_[rank];
+    CArray<double,1> val;
+    CArray<double,2> boundsVal;
+
+    buffer >> val;
+    if (hasBounds_) buffer >> boundsVal;
+
+    int i, j, ind_srv;
+    for (int ind = 0; ind < indi.numElements(); ++ind)
+    {
+      i = indi(ind);
+      ind_srv = i - zoom_begin_srv;
+      value_srv(ind_srv) = val(ind);
+      if (hasBounds_)
+      {
+        bound_srv(0,ind_srv) = boundsVal(0, ind);
+        bound_srv(1,ind_srv) = boundsVal(1, ind);
+      }
+    }
+  }
+
+   void CAxis::recvNonDistributedValue(CEventServer& event)
+  {
+    list<CEventServer::SSubEvent>::iterator it;
+    for (it = event.subEvents.begin(); it != event.subEvents.end(); ++it)
+    {
+      CBufferIn* buffer = it->buffer;
+      string domainId;
+      *buffer >> domainId;
+      get(domainId)->recvNonDistributedValue(it->rank, *buffer);
+    }
+  }
+
+  void CAxis::recvNonDistributedValue(int rank, CBufferIn& buffer)
+  {
+    CArray<double,1> val;
+    buffer >> val;
+
+    for (int ind = 0; ind < val.numElements(); ++ind)
+    {
+      value_srv(ind) = val(ind);
+      if (hasBounds_)
+      {
+        bound_srv(0,ind) = bounds(0,ind);
+        bound_srv(1,ind) = bounds(1,ind);
+      }
+    }
+  }
 
   void CAxis::sendServerAttribut(const std::vector<int>& globalDim, int orderPositionInGrid,
@@ -263 +567 @@
         msg << this->getId();
         msg << ni << begin << end;
-        msg<<global_zoom_begin<<global_zoom_size;
+        msg << global_zoom_begin << global_zoom_size;
 
         event.push(*itRank,1,msg);
@@ -304 +608 @@
       zoom_size_srv  = zoom_end_srv - zoom_begin_srv + 1;
     }
+    value_srv.resize(zoom_size_srv);
+    bound_srv.resize(2,zoom_size_srv);
   }