Index: XIOS/dev/dev_ym/XIOS_COUPLING/src/distribution/gatherer_connector.hpp
===================================================================
--- XIOS/dev/dev_ym/XIOS_COUPLING/src/distribution/gatherer_connector.hpp	(revision 2267)
+++ XIOS/dev/dev_ym/XIOS_COUPLING/src/distribution/gatherer_connector.hpp	(revision 2291)
@@ -9,4 +9,5 @@
 #include "distributed_view.hpp"
 #include "context_client.hpp"
+#include "reduction_types.hpp"
 
 
@@ -29,51 +30,126 @@
       
       template<typename T>
-      void transfer(int repeat, int sizeT, map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut)
+      void transfer(int repeat, int sizeT, map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut, EReduction op = EReduction::none)
       {
         // for future, make a specific transfer function for sizeT=1 to avoid multiplication (increasing performance)
+       
         size_t dstSlice = dstSize_*sizeT ;
         dataOut.resize(repeat* dstSlice) ;  
         
-        for(auto& data : dataIn)
-        {
+        
+        if (op == EReduction::none) // tranfer without reduction
+        {
+          for(auto& data : dataIn)
+          {
+            T* output = dataOut.dataFirst() ;
+            int rank=data.first ;
+            auto input  = data.second.dataFirst() ;
+            auto& connector=connector_[rank] ;
+            auto& mask=mask_[rank] ;
+            int size=mask.size() ;
+            size_t srcSlice = size * sizeT ;
+            for(int l=0; l<repeat; l++)
+            {   
+              for(int i=0, j=0 ;i<size;i++)
+              {
+                if (mask[i]) 
+                {
+                  int cj = connector[j]*sizeT ;
+                  int ci = i*sizeT ;
+                  for (int k=0;k<sizeT;k++) output[cj+k] = input[ci+k] ;
+                  j++ ;
+                }
+              }
+              input+=srcSlice ;
+              output+=dstSlice ;
+           }
+          }
+        }
+        else // with reduction, to be optimized, see how to vectorize
+        {
+          vector<int> vcount(dataOut.size(),0) ;
+          int* count = vcount.data() ;
+          T defaultValue = std::numeric_limits<T>::quiet_NaN();
+          for(auto& data : dataIn)
+          {
+            T* output = dataOut.dataFirst() ;
+            int rank=data.first ;
+            auto input  = data.second.dataFirst() ;
+            auto& connector=connector_[rank] ;
+            auto& mask=mask_[rank] ;
+            int size=mask.size() ;
+            size_t srcSlice = size * sizeT ;
+            for(int l=0; l<repeat; l++)
+            {   
+              for(int i=0, j=0 ;i<size;i++)
+              {
+                if (mask[i]) 
+                {
+                  int cj = connector[j]*sizeT ;
+                  int ci = i*sizeT ;
+                  for (int k=0;k<sizeT;k++)
+                  { 
+                    if (!std::isnan(input[ci+k])) // manage missing value
+                    {
+                      if (count[cj+k]==0)  output[cj+k] = input[ci+k] ;
+                      else
+                      {
+                        switch(op)
+                        {
+                          case EReduction::sum :
+                            output[cj+k] += input[ci+k] ;
+                            break ;
+                          case EReduction::min :
+                            output[cj+k]= std::min(output[cj+k],input[ci+k]) ;
+                            break ;
+                          case EReduction::max :
+                            output[cj+k]= std::max(output[cj+k],input[ci+k]) ;
+                            break ;
+                          case EReduction::average :
+                            output[cj+k] += input[ci+k] ;
+                            break ;
+                          default :
+                            ERROR("CGathererConnector::transfer",
+                               <<"reduction operator "<<(int)op<<" is not defined for this operation") ;
+                          break ;
+                        }
+                      }
+                      count[cj+k]++ ; 
+                    }
+                  }
+                  j++ ;
+                }
+              }
+              input+=srcSlice ;
+              output+=dstSlice ;
+              count+=dstSlice ;
+            }
+          }
+
           T* output = dataOut.dataFirst() ;
-          int rank=data.first ;
-          auto input  = data.second.dataFirst() ;
-          auto& connector=connector_[rank] ;
-          auto& mask=mask_[rank] ;
-          int size=mask.size() ;
-          size_t srcSlice = size * sizeT ;
-          for(int l=0; l<repeat; l++)
-          {   
-            for(int i=0, j=0 ;i<size;i++)
-            {
-              if (mask[i]) 
-              {
-                int cj = connector[j]*sizeT ;
-                int ci = i*sizeT ;
-                for (int k=0;k<sizeT;k++) output[cj+k] = input[ci+k] ;
-                j++ ;
-              }
-            }
-            input+=srcSlice ;
-            output+=dstSlice ;
-          }
-        }
-      }
-
-      template<typename T>
-      void transfer(int sizeT, map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut)
-      {
-        transfer(1, sizeT, dataIn, dataOut) ;
+          if (op==EReduction::average)
+            for(int i=0;  i < dataOut.size() ; i++) 
+            {
+              if (count[i]>0) dataOut[i]/=count[i] ;
+              else dataOut[i] = defaultValue ;
+            }
+          else for(int i=0;  i < dataOut.size() ; i++) if (count[i]==0) dataOut[i] = defaultValue ;
+        }
+      }
+
+      template<typename T>
+      void transfer(int sizeT, map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut, EReduction op = EReduction::none)
+      {
+        transfer(1, sizeT, dataIn, dataOut, op) ;
       }
     
       template<typename T>
-      void transfer(map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut)
-      {
-        transfer(1,dataIn,dataOut) ;
-      }
-
-      template<typename T>
-      void transfer(int rank,  shared_ptr<CGathererConnector>* connectors, int nConnectors, const T* input, T* output)
+      void transfer(map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut, EReduction op = EReduction::none)
+      {
+        transfer(1,dataIn,dataOut, op) ;
+      }
+
+      template<typename T>
+      void transfer(int rank,  shared_ptr<CGathererConnector>* connectors, int nConnectors, const T* input, T* output, EReduction op = EReduction::none, int* count=nullptr)
       {
         auto& connector = connector_[rank] ; // probably costly, find a better way to avoid the map
@@ -83,10 +159,75 @@
         if (nConnectors==0)
         {
-          for(int i=0, j=0; i<srcSize; i++)
-            if (mask[i]) 
-            {
-              *(output+connector[j]) = *(input + i) ;
-              j++ ;
-            }
+          if (op == EReduction::none)
+          {
+            for(int i=0, j=0; i<srcSize; i++)
+              if (mask[i]) 
+              {
+                *(output+connector[j]) = *(input + i) ;
+                j++ ;
+              }
+          }
+          else
+          {
+            switch(op)
+            {
+              case EReduction::sum :
+                for(int i=0, j=0; i<srcSize; i++)
+                  if (mask[i]) 
+                  {
+                    if (!std::isnan(*(input + i)))
+                    {
+                      if (*(count+connector[j])==0) *(output+connector[j]) = *(input + i) ;
+                      else *(output+connector[j]) += *(input + i) ;
+                      (*(count+connector[j]))++ ;
+                    } 
+                    j++ ;
+                  }
+                break ;
+              case EReduction::min :
+                for(int i=0, j=0; i<srcSize; i++)
+                  if (mask[i]) 
+                  {
+                    if (!std::isnan(*(input + i)))
+                    {
+                      if (*(count+connector[j])==0) *(output+connector[j]) = *(input + i) ;
+                      else *(output+connector[j]) = std::min(*(output+connector[j]),*(input + i)) ;
+                      (*(count+connector[j]))++ ;
+                    } 
+                    j++ ;
+                  }
+                break ;
+              case EReduction::max :
+                for(int i=0, j=0; i<srcSize; i++)
+                  if (mask[i]) 
+                  {
+                    if (!std::isnan(*(input + i)))
+                    {
+                      if (*(count+connector[j])==0) *(output+connector[j]) = *(input + i) ;
+                      else *(output+connector[j]) = std::max(*(output+connector[j]),*(input + i)) ;
+                      (*(count+connector[j]))++ ;
+                    } 
+                    j++ ;
+                  }
+                break ;
+              case EReduction::average :
+                for(int i=0, j=0; i<srcSize; i++)
+                  if (mask[i]) 
+                  {
+                    if (!std::isnan(*(input + i)))
+                    {
+                      if (*(count+connector[j])==0) *(output+connector[j]) = *(input + i) ;
+                      else *(output+connector[j]) += *(input + i) ;
+                      (*(count+connector[j]))++ ;
+                    } 
+                    j++ ;
+                  }
+                break ;
+              default :
+                ERROR("CGathererConnector::transfer",
+                     <<"reduction operator "<<(int)op<<" is not defined for this operation") ;
+                break ;
+            }
+          }
 
         }
@@ -101,5 +242,5 @@
             if (mask[i]) 
             {
-              (*(connectors-1))->transfer(rank, connectors-1, nConnectors-1, in, output+connector[j]*dstSliceSize) ; // the multiplication must be avoid in further optimization
+              (*(connectors-1))->transfer(rank, connectors-1, nConnectors-1, in, output+connector[j]*dstSliceSize, op, count+connector[j]*dstSliceSize) ; // the multiplication must be avoid in further optimization
               j++ ;
             }
@@ -149,25 +290,25 @@
 
       template<typename T>
-      void transfer(map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut, T missingValue)
-      {
-        transfer(1, 1, dataIn, dataOut, missingValue);
-      }
-      
-      template<typename T>
-      void transfer(int sizeT, map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut, T missingValue)
-      {
-         transfer(1, sizeT, dataIn, dataOut, missingValue) ;
-      }
-
-      template<typename T>
-      void transfer(int repeat , int sizeT, map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut, T missingValue)
+      void transfer(map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut, T missingValue, EReduction op = EReduction::none)
+      {
+        transfer(1, 1, dataIn, dataOut, missingValue, op);
+      }
+      
+      template<typename T>
+      void transfer(int sizeT, map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut, T missingValue, EReduction op = EReduction::none)
+      {
+         transfer(1, sizeT, dataIn, dataOut, missingValue, op) ;
+      }
+
+      template<typename T>
+      void transfer(int repeat , int sizeT, map<int, CArray<T,1>>& dataIn, CArray<T,1>& dataOut, T missingValue, EReduction op = EReduction::none)
       {
         dataOut.resize(repeat*dstSize_*sizeT) ;
         dataOut=missingValue ;
-        transfer(repeat, sizeT, dataIn, dataOut) ;
-      }
-
-      template<typename T>
-      void transfer(CEventServer& event, int sizeT, CArray<T,1>& dataOut)
+        transfer(repeat, sizeT, dataIn, dataOut, op) ;
+      }
+
+      template<typename T>
+      void transfer(CEventServer& event, int sizeT, CArray<T,1>& dataOut, EReduction op = EReduction::none)
       {
         map<int, CArray<T,1>> dataIn ;
@@ -177,15 +318,15 @@
           (*subEvent.buffer) >> data ;
         }
-        transfer(1, sizeT, dataIn, dataOut) ;
-      }
-      
-      template<typename T>
-      void transfer(CEventServer& event, CArray<T,1>& dataOut)
-      {
-        transfer(event, 1, dataOut) ;
-      }
-
-      template<typename T>
-      void transfer(CEventServer& event, int sizeT, CArray<T,1>& dataOut, T missingValue)
+        transfer(1, sizeT, dataIn, dataOut, op) ;
+      }
+      
+      template<typename T>
+      void transfer(CEventServer& event, CArray<T,1>& dataOut, EReduction op = EReduction::none)
+      {
+        transfer(event, 1, dataOut, op) ;
+      }
+
+      template<typename T>
+      void transfer(CEventServer& event, int sizeT, CArray<T,1>& dataOut, T missingValue, EReduction op = EReduction::none)
       {
         map<int, CArray<T,1>> dataIn ;
@@ -195,9 +336,9 @@
           (*subEvent.buffer) >> data ;
         }
-        transfer(1, sizeT, dataIn, dataOut, missingValue) ;
-      }
-
-      template<typename T>
-      void transfer(CEventServer& event, CArray<T,1>& dataOut, T missingValue)
+        transfer(1, sizeT, dataIn, dataOut, missingValue, op) ;
+      }
+
+      template<typename T>
+      void transfer(CEventServer& event, CArray<T,1>& dataOut, T missingValue, EReduction op = EReduction::none)
       {
         map<int, CArray<T,1>> dataIn ;
@@ -207,5 +348,5 @@
           (*subEvent.buffer) >> data ;
         }
-        transfer(1, 1, dataIn, dataOut, missingValue) ;
+        transfer(1, 1, dataIn, dataOut, missingValue, op) ;
       }
 
