duo_distributed_vector.h

Go to the documentation of this file.

/* ./bridging/duo_distributed_vector.h 
**********************************
Copyright INRIA and CEA 

author : Guillaume ANCIAUX (anciaux@labri.fr, g.anciaux@laposte.net)

The LibMultiScale is a C++ parallel framework for the multiscale
coupling methods dedicated to material simulations. This framework
provides an API which makes it possible to program coupled simulations
and integration of already existing codes.

This Project is done in a collaboration between INRIA Futurs Bordeaux
within ScAlApplix team and CEA/DPTA Ile de France. 

This software is governed by the CeCILL-C license under French law and
abiding by the rules of distribution of free software.  You can  use, 
modify and/ or redistribute the software under the terms of the CeCILL-C
license as circulated by CEA, CNRS and INRIA at the following URL
"http://www.cecill.info". 

As a counterpart to the access to the source code and  rights to copy,
modify and redistribute granted by the license, users are provided only
with a limited warranty  and the software's author,  the holder of the
economic rights,  and the successive licensors  have only  limited
liability. 

In this respect, the user's attention is drawn to the risks associated
with loading,  using,  modifying and/or developing or reproducing the
software by the user in light of its specific status of free software,
that may mean  that it is complicated to manipulate,  and  that  also
therefore means  that it is reserved for developers  and  experienced
professionals having in-depth computer knowledge. Users are therefore
encouraged to load and test the software's suitability as regards their
requirements in conditions enabling the security of their systems and/or 
data to be ensured and,  more generally, to use and operate it in the 
same conditions as regards security. 

The fact that you are presently reading this means that you have had
knowledge of the CeCILL-C license and that you accept its terms.
***********************************/

#ifndef DUODISTRIBUTEDVECTEUR_H
#define DUODISTRIBUTEDVECTEUR_H

#include "../communicator/communicator.h"
#include <list>

typedef struct Myblocks_ {
  unsigned int start;
  unsigned int end;
}blocks;


class DuoDistributedVecteur{
 public:

  DuoDistributedVecteur(int lsize,unsigned int tsize){
    DUMP("Creating duodistributed vector of size " << lsize << "," << tsize,DBG_INFO_STARTUP);
    //init des flags et indicateurs
    totalsize = tsize;
    localsize = lsize;
    compressed = false;

    //allocation du tableau principal
    //qui sera compressé plus tard
    indirection = new int[localsize];
    
    // mise a zero du vecteur
    memset(indirection,-1,sizeof(int)*localsize);

    schNew_index = 1;
    sch_index = 0;

    something_changed = true;
    step = 0;
  };

  void SetNeighbor(int i, int proc){
    if (indirection[i]!= -1){
      FATAL("distant owner for index " << i << " was already attributed to " << indirection[i] << " i try to change it for  " << proc);
    }
    indirection[i] = proc;
  };

  int GetNeighbor(int i){
    blocks b;
    int proc;
    findBlock(i,b,proc,sch_index);
    return proc;
  }

  int GetUncompressedNeighbor(int i){
    return indirection[i];
  }


  //les methodes ci dessous stocke les envois
  //pour le faire en une fois de maniere globale
  //par la methode SynchronizeMigration
  void receiveAtom(int i,int neighbor){
    //    int dist_index;

   AddAtomInBlocks(i,neighbor,schNew_index);
    // findDistIndex(i,neighbor,dist_index,schNew_index);
   DUMP("atom " << i << " received with new neighbor " << neighbor << " newpos = " << i << " number " 
        <<  received[neighbor].size(),DBG_INFO);
   //    received[neighbor].push_back(dist_index);
   received[neighbor].push_back(i);
  }

  void sendAtom(int i , int toproc){
    DUMP("atom " << i << " migrate to " << toproc,DBG_INFO);
    //je le cherche pour voir qd il est envoye au voisin FE
    int neigh,dist_index=0;
    dist_index = findDistIndex(i,neigh,sch_index);
    RemoveAtomFromBlocks(i,schNew_index);
    //j'informe le nouveau proprietaire de qui est sont homologue
    //    if (dist_index == sent[neigh][sent[neigh].size()-1].first) FATAL("detection peu couteuse : double declaration de migration d'atome : index" 
    //                                                               << dist_index); 
    sent[neigh].push_back(std::make_pair<int,int>(dist_index,toproc));
    DUMP("actually size to notify to proc " << neigh << " size = " << sent[neigh].size(),DBG_INFO);
  }


  int findDistIndex(unsigned int i,int & neighbor,int schIndex){
    DUMP("request findDistIndex "  << i,DBG_DETAIL);
    std::map<int,std::list<blocks> > & scheme = compressed_scheme[schIndex];

    neighbor = -2;
    std::map<int,std::list<blocks> >::iterator it = scheme.begin();
    std::map<int,std::list<blocks> >::iterator last = scheme.end();
    
    while (it != last){
      int proc = (*it).first;
      std::list<blocks> & lblocks = (*it).second;
      std::list<blocks>::iterator it_block = lblocks.begin();
      std::list<blocks>::iterator last_block = lblocks.end();
      int dist_index = 0;
  
      for (it_block = lblocks.begin(); it_block != last_block ; ++it_block){ 
        blocks & b = (*it_block);
        if (i < b.end && i>= b.start)
          {
            neighbor = proc;
            dist_index += i - b.start;
            return dist_index;
          }
        dist_index += b.end-b.start;
      }
      ++it;
    }
   
    if (neighbor == -2) FATAL("not found atom " << i << " in any blocks");
    return -1;
  }


  int findRealIndex(unsigned int i,int neighbor,int schIndex){
    DUMP("request findRealIndex "  << i << "," << neighbor,DBG_DETAIL);
    std::map<int,std::list<blocks> > & scheme = compressed_scheme[schIndex];

    std::list<blocks> & lblocks = scheme[neighbor];
    std::list<blocks>::iterator it_block = lblocks.begin();
    std::list<blocks>::iterator last_block = lblocks.end();
    unsigned int dist_index = 0;
    
    for (it_block = lblocks.begin(); it_block != last_block ; ++it_block){ 
      blocks & b = (*it_block);
      
      if (i < dist_index + b.end - b.start && i>= dist_index)
        {
          return (b.start + i - dist_index);
        }
      else if (i >= b.start && i < b.end) FATAL("insertion d'atom impossible sur une position " << i 
                                                << " alors qu'il existe un bloc " << b.start << " - " << b.end);
    }
    FATAL("distant index not existant for proc " << neighbor << " dist_index " << i);
  }



  void AddAtomInBlocks(unsigned int i,int neighbor,int schIndex){
    std::map<int,std::list<blocks> > & scheme = compressed_scheme[schIndex];

    std::list<blocks> & lblocks = scheme[neighbor];
    //je creer systematiquement un bloc a la fin pour la coherence du schemas de com (+ simple)
    blocks new_b; 
    new_b.start = i; 
    new_b.end = i+1; 
    lblocks.push_back(new_b);
    

/*     std::list<blocks>::iterator it_block = lblocks.begin(); */
/*     std::list<blocks>::iterator last_block = lblocks.end(); */
/*     int done_flag=false; */
/*     blocks new_b; */

/*     if (i+1>localsize) */
/*       localsize = i+1; */
    
/*     for (it_block = lblocks.begin(); it_block != last_block ; ++it_block){  */
/*       blocks & b = (*it_block); */
/*       if (i == b.end){ */
/*      DUMP("cas 1 " << i,DBG_DETAIL); */
/*      //on etend le bloc */
/*      //      ++b.end;  */
/*      new_b.start = i; */
/*      new_b.end = i+1; */
/*      ++it_block; */
/*      lblocks.insert(it_block,new_b); */
/*      done_flag = true; */
/*      break; */
/*       } */
/*       else if (i+1 == b.start){ */
/*      DUMP("cas 2 " << i,DBG_DETAIL); */
/*      //on etend le bloc par devant */
/*      //      --b.start;  */
/*      new_b.start = i; */
/*      new_b.end = i+1; */
/*      lblocks.insert(it_block,new_b); */
/*      done_flag = true; */
/*      break; */
/*       } */
/*       else if (i < b.start) */
/*      { */
/*        DUMP("cas 3 " << i,DBG_DETAIL); */
/*        //c l'endroit parfait pour creer le nouveau bloc de taille 1 */
/*        new_b.start = i; */
/*        new_b.end = i+1; */
/*        lblocks.insert(it_block,new_b); */
/*        done_flag = true; */
/*        break; */
/*      } */
/*       else if (i >= b.start && i < b.end) FATAL("insertion d'atom impossible sur une position " << i  */
/*                                              << " alors qu'il existe un bloc " << b.start << " - " << b.end); */
/*     } */

/*     if (!done_flag){ */
/*       DUMP("cas 4 " << i,DBG_DETAIL); */
/*       //je creer un nouveau bloc apres tout les autres */
/*       new_b.start = i; */
/*       new_b.end = i+1; */
/*       lblocks.insert(it_block,new_b); */
/*     } */

  }

  void AddAtomInBlocksWithDistIndex(unsigned int distant_index,unsigned int real_index,int neighbor,int schIndex){
    DUMP("request  AddAtomInBlocksWithSpecifiedOrder "  << distant_index << "," << real_index << "," << neighbor,DBG_DETAIL);
    std::map<int,std::list<blocks> > & scheme = compressed_scheme[schIndex];

    std::list<blocks> & lblocks = scheme[neighbor];
    std::list<blocks>::iterator it_block = lblocks.begin();
    std::list<blocks>::iterator last_block = lblocks.end();
    int done_flag=false;
    blocks new_b;

    if (real_index+1 > localsize){
      DUMP("AAAAAAAAAAAAAA this should not append ?",DBG_INFO);
      localsize = real_index+1;
    }
    unsigned int current_index = 0;
    for (it_block = lblocks.begin(); it_block != last_block ; ++it_block){
      blocks & b = (*it_block);
      if (distant_index >= current_index && distant_index < current_index + b.end-b.start){
        //c'est a l'interieur de ce bloc qu'il faut inserer le nouveau bloc
        if(distant_index == current_index){
          FATAL("AAAA");
          //j'insere le nouveau bloc juste avant le suivant
          new_b.start = real_index;
          new_b.end = real_index+1;
          
          lblocks.insert(it_block,new_b);
          done_flag=true;
          DUMP("insertion d'un bloc qui envoie la coordonnee " << real_index << " a " << neighbor,DBG_INFO);
        }
        else{
          FATAL("BBBB");
          //sinon je coupe le bloc en deux
          int index_to_remove = distant_index - current_index + b.start;
          new_b.start = index_to_remove + 1;
          new_b.end = b.end;
          b.end = index_to_remove;
          ++it_block;
          lblocks.insert(it_block,new_b);
          new_b.start = real_index;
          new_b.end = real_index +1 ;
          DUMP("splitting bloc - new blocs -> " << b.start << " - " << b.end << " , " << new_b.start << " - " << new_b.end,DBG_INFO);
          lblocks.insert(it_block,new_b);
          DUMP("adding new bloc at the end referencing index " << real_index,DBG_INFO);
          done_flag=true;
          break;
        }
      }
      current_index += b.end-b.start;
    }
    if (!done_flag){
      //je creer un nouveau bloc apres tout les autres
      new_b.start = real_index;
      new_b.end = real_index+1;
      lblocks.insert(it_block,new_b);
      DUMP("adding new bloc at the end referencing index " << real_index << " associated with " << neighbor,DBG_INFO) ;
    }

  }


  void RemoveAtomFromBlocks(unsigned int i,int schIndex){
    int done_flag = 0;

    std::map<int,std::list<blocks> > & scheme = compressed_scheme[schIndex];

    std::map<int,std::list<blocks> >::iterator it = scheme.begin();
    std::map<int,std::list<blocks> >::iterator last = scheme.end();
    
    while (it != last){
      int proc = (*it).first;
      std::list<blocks> & lblocks = (*it).second;
      std::list<blocks>::iterator it_block = lblocks.begin();
      std::list<blocks>::iterator last_block = lblocks.end();
  
      for (it_block = lblocks.begin(); it_block != last_block ; ++it_block){ 
        blocks & b = (*it_block);
        if (i < b.end && i>= b.start)
          {
            if (i == b.start && i+1 == b.end){
              lblocks.erase(it_block);
              done_flag = 1;
              DUMP("found a block with one element (" << i << "," << i+1 << ") -> erased",DBG_INFO);
            }
            else if (i == b.start && i+1 != b.end){
              b.start = i+1;
              done_flag = 1;
              DUMP("index specified was first of block - none created only first index removed from block",DBG_INFO);
            }
            else if (i != b.start && i+1 == b.end){
              b.end = i;
              done_flag = 1;
              DUMP("index specified was last of block - none created only last index removed from block",DBG_INFO);
            }
            else{
              done_flag = 1;
              DUMP("splitting bloc " << b.start << " - " << b.end << " associated with " << proc << " (" << &b << ")",DBG_INFO);
              //si besoin je creer un nouveau bloc      
              blocks new_b;
              new_b.end = b.end;
              new_b.start = i+1;
              b.end = i;
              ++it_block;
              lblocks.insert(it_block,new_b);
              DUMP("the new blocs " << b.start << " - " << b.end << " , " << new_b.start << " - " << new_b.end,DBG_INFO);
            }
            break;
          }
        
      }
      if (done_flag) break;
      ++it;
    }
  }

  void findBlock(unsigned int i,blocks & bl,int & neighbor,int index){
    neighbor = -2;
    std::map<int,std::list<blocks> > & scheme = compressed_scheme[index];

    std::map<int,std::list<blocks> >::iterator it = scheme.begin();
    std::map<int,std::list<blocks> >::iterator last = scheme.end();
    
    while (it != last){
      int proc = (*it).first;
      std::list<blocks> & lblocks = (*it).second;
      std::list<blocks>::iterator it_block = lblocks.begin();
      std::list<blocks>::iterator last_block = lblocks.end();
      
      for (it_block = lblocks.begin(); it_block != last_block ; ++it_block){ 
        blocks b = (*it_block);
        DUMP("testing if " << i << " is between " << b.start << " and " << b.end << " neigh = " << proc,DBG_DETAIL);
        if (i < b.end && i>= b.start)
          {
            bl = b;
            neighbor = proc;
            break;
          }
      }
      if (neighbor != -2)
        break;
      ++it;
    }
    if (neighbor == -2){
      FATAL("block not found for index " << i);
    }
  }

  int SynchronizeMigration(int groupeAtomic,int groupeFE,Communicator & com){
    DUMP("SynchronizeMigration start",DBG_INFO);
    // en premier je parcours la map des "sent" pour avertir des coms

    std::map<int,std::list<blocks> > & scheme = compressed_scheme[sch_index];

    if (scheme.size() == 0)
      return 0;

    something_changed = false;

    int proc;
    int nb_migrated;
    std::vector<int> buffer;

    //atoms side
    if(com.amIinGroup(groupeAtomic)){

      {
        std::map<int,std::list<blocks> >::iterator it = scheme.begin();
        std::map<int,std::list<blocks> >::iterator last = scheme.end();
        
        while (it != last){
          proc = (*it).first;
          if (proc != -1){
            nb_migrated = sent[proc].size();
            DUMP("envoie le nombre d'atomes qui ont migres to proc "  << com.RealRank(proc,groupeFE) << " (" << nb_migrated << ")",DBG_INFO);
            com.SendIntegers(&nb_migrated,1,proc,groupeFE);
            com.waitForPendingComs();
            if (nb_migrated){
              something_changed = true;
              DUMP("i warn proc " << com.RealRank(proc,groupeFE) << " of migration of " << nb_migrated << " atoms",DBG_INFO);
              //je construit maintenant le vecteur d'informationd sur les migrations
              buffer.resize(2*nb_migrated);
              //je parcours les atomes migres
              for (int i = 0 ; i < nb_migrated ; ++i){
                buffer[2*i] = sent[proc][i].first;
                buffer[2*i+1] = sent[proc][i].second;
              }
              
              DUMP("i send the migration information to " << com.RealRank(proc,groupeFE),DBG_INFO);
              com.SendIntegers(&buffer[0],2*nb_migrated,proc,groupeFE);
              com.waitForPendingComs();
            }
          }
          ++it;
        }
      }
      

      std::map<int,std::vector<int> >::iterator it = received.begin();      
      std::map<int,std::vector<int> >::iterator last = received.end();      


      // first converting all the local indexes in subset to blks indexes
      while (it != last){
        int proc = (*it).first;
        vector<int> & new_pos = (*it).second;
        int temp;
        for (unsigned int cpt = 0 ; cpt < new_pos.size() ; ++cpt){
          DUMP("find distant index for index " << cpt,DBG_DETAIL);
          int toto = proc;
          temp = findDistIndex(new_pos[cpt],toto,schNew_index);
          if (toto != proc) FATAL("inconstistency in block management " << toto << " != " << proc); 
          DUMP("result is " << temp << " neighbor detected = " << proc,DBG_DETAIL);
          
          new_pos[cpt] = temp;
        }
        ++it;
      }

      //dont't forget to reset iterator
      it = received.begin();      
      // then send all the data for received atoms indexes to FE nodes
      while (it != last){
        int proc = (*it).first;
        vector<int> & new_pos = (*it).second;
        DUMP("i send " << new_pos.size() << " indexes of new positions to proc " << com.RealRank(proc,groupeFE),DBG_INFO);
        if (proc != -1){
          something_changed = true;
#ifdef DEBUG_BRIDGE
          {
            int nb_send = new_pos.size();
            com.SendIntegers(&nb_send,1,proc,groupeFE);
          }
#endif
          com.SendIntegers(&new_pos[0],new_pos.size(),proc,groupeFE);
        }
        com.waitForPendingComs();
        ++it;
      }
    }
    

    // FE side
    if(com.amIinGroup(groupeFE)){
      std::map<int,std::vector<int> > pending_coms;

      {
        std::map<int,std::list<blocks> >::iterator it = scheme.begin();
        std::map<int,std::list<blocks> >::iterator last = scheme.end();
        
        while (it != last){
          int proc = (*it).first;
          //      std::list<blocks> & lblocks = (*it).second;
          //      std::list<blocks>::iterator it_block = lblocks.begin();
          //std::list<blocks>::iterator last_block = lblocks.end();
          //je recois le nombre de migration
          DUMP("recois le nombre d'atomes qui ont migres de " << com.RealRank(proc,groupeAtomic),DBG_INFO);
          if (proc == -1){++it;continue;}

          com.ReceiveIntegers(&nb_migrated,1,proc,groupeAtomic);
          com.waitForPendingComs();
          if (nb_migrated){
            something_changed = true;
            DUMP("i am warned from proc " << com.RealRank(proc,groupeAtomic) << " of migration of " << nb_migrated << " atoms",DBG_INFO);
            buffer.resize(2*nb_migrated);
            DUMP("i receive the migration information from " << com.RealRank(proc,groupeAtomic),DBG_INFO);
            com.ReceiveIntegers(&buffer[0],2*nb_migrated,proc,groupeAtomic);
            com.waitForPendingComs();
            
            //maintenant que j'ai recupere les informations je peux parcourir mon schemas de com
            // et le modifier
            //je parcours les infos recues
            
            for (int i = 0 ; i < nb_migrated ; ++i){    
              unsigned int distant_index = buffer[2*i];
              unsigned int new_neigh = buffer[2*i+1];
              int real_index = 0;
              //              int old_neigh;
              
              DUMP("splitting blocks from distant index " << distant_index,DBG_INFO);
              real_index = findRealIndex(distant_index,proc,sch_index);
              RemoveAtomFromBlocks(real_index,schNew_index);
              pending_coms[new_neigh].push_back(real_index);
            }
          }         
          ++it;
        }
      }
      {
        //maintenant je receptionne les coms des nouveaux voisins pour chaque atomes
        std::map<int,std::vector<int> >::iterator it = pending_coms.begin();
        std::map<int,std::vector<int> >::iterator last = pending_coms.end();
        
        while (it != last){
          int proc = (*it).first;
          int nb_coms = (*it).second.size();
          if (nb_coms){
            buffer.resize(nb_coms);
            DUMP("reception of " << nb_coms << " atoms new indexes from " << com.RealRank(proc,groupeAtomic),DBG_INFO);
#ifdef DEBUG_BRIDGE
            {
              int nb_recv = 0;
              com.ReceiveIntegers(&nb_recv,1,proc,groupeAtomic);
              com.waitForPendingComs();
              if (nb_recv != nb_coms)
                FATAL("redistribution failed : proc " << com.RealRank(proc,groupeAtomic) << " want to send me " << nb_recv << " integers but i planed " <<  nb_coms);
            }
#endif
            com.ReceiveIntegers(&buffer[0],nb_coms,proc,groupeAtomic);
            com.waitForPendingComs();
            if (buffer.size() != pending_coms[proc].size()) FATAL("inconstistency");
            //      doubleQuickSort(&buffer[0],&(pending_coms[proc][0]),0,pending_coms[proc].size()-1);
            for (int i = 0 ; i < nb_coms ; ++i){
              if (i > 0 && buffer[i-1] > buffer[i]) FATAL("inconstistency in duo blocks "  
                                                          << buffer[i-1] << " " << buffer[i]); 
              AddAtomInBlocksWithDistIndex(buffer[i],pending_coms[proc][i],proc,schNew_index);
            }
          }
          ++it;
        }
      }

    }
    
    sent.clear();
    received.clear();
    
    swapSchemes();

    return something_changed;

    DUMP("done sync",DBG_INFO);
    return 0;
  }

  void swapSchemes(){
    if (sch_index == 0){
      sch_index = 1;
      schNew_index = 0;
    }
    else{
      sch_index = 0;
      schNew_index = 1;
    }
    
    std::map<int,std::list<blocks> > & scheme = compressed_scheme[sch_index];
    std::map<int,std::list<blocks> > & schemeNew = compressed_scheme[schNew_index];    

    schemeNew.clear();
    std::map<int,std::list<blocks> >::iterator it = scheme.begin();
    std::map<int,std::list<blocks> >::iterator last = scheme.end();
    
    while (it != last){
      int proc = (*it).first;
      std::list<blocks> & lblocks = (*it).second;
      std::list<blocks>::iterator it_block = lblocks.begin();
      std::list<blocks>::iterator last_block = lblocks.end();
      blocks b;

      while (it_block != last_block){
         b = (*it_block);
        schemeNew[proc].push_back(b);
        ++it_block;
      }
      ++it;
    }
  }

  void compress(){
    std::map<int,std::list<blocks> > & scheme = compressed_scheme[sch_index];
    std::map<int,std::list<blocks> > & schemeNew = compressed_scheme[schNew_index];

    scheme.clear();

    unsigned int sindex=0;
    int current_ind=indirection[0];

    for(unsigned int i = 1; i < localsize ; ++i)
      {
        if (current_ind == indirection[i]){
          continue;
        }

        blocks b;
        //j'init ses valeurs de block
        b.start = sindex;
        b.end = i;
        //      b.indirection = current_ind;

        //je rajoute un block   
        scheme[current_ind].push_back(b);
        //je met a jour le nouveau start
        sindex = i;

        //je met a jour l'indirection courante
        current_ind = indirection[i];
      }

    //je rentre le dernier block
    blocks b;
    b.start = sindex;
    b.end = localsize;
    //    b.indirection = current_ind;
    scheme[current_ind].push_back(b);

    DUMP("compressed to " << scheme.size() << " blocks",DBG_INFO_STARTUP);
    compressed = true;

    nb_par_interlocuteurs.clear();

    std::map<int,std::list<blocks> >::iterator it = scheme.begin();
    std::map<int,std::list<blocks> >::iterator last = scheme.end();

    while (it != last){
      int proc = (*it).first;
      std::list<blocks> & lblocks = (*it).second;
      std::list<blocks>::iterator it_block = lblocks.begin();
      std::list<blocks>::iterator last_block = lblocks.end();

      while (it_block != last_block){
        blocks b = (*it_block);
        nb_par_interlocuteurs[proc]+=b.end-b.start;
        schemeNew[proc].push_back(b);

        ++it_block;
        DUMP("dans la compression initiale on a le block (" << b.start << " - " << b.end << ") pour le proc " << proc,DBG_WARNING);
      }

      ++it;
    }
    
  };

#ifdef DEBUG_BRIDGE
  void SynchronizeVecteurBySum(char * name_vec,double * vec,unsigned int nmax,int groupe1,int groupe2,Communicator & com,int stride=1){
#else
  void SynchronizeVecteurBySum(char * name_vec,double * vec,int groupe1,int groupe2,Communicator & com,int stride=1){
#endif
    if(com.amIinGroup(groupe1)){
      DUMP("distribue le vecteur de " << groupe1 << " vers " << groupe2,DBG_INFO);
#ifdef DEBUG_BRIDGE
      DistributeVecteur(name_vec,vec,nmax,groupe1,groupe2,com,stride);
#else
      DistributeVecteur(name_vec,vec,groupe1,groupe2,com,stride);
#endif
    }
    else if(com.amIinGroup(groupe2)){
      DUMP("receptionne la somme partielle de " << groupe1 << " vers " << groupe2,DBG_INFO);
      tmp.resize(localsize*stride); 
#ifdef DEBUG_BRIDGE      
      DistributeVecteur("tmp vector for dist-reduction",&tmp[0],localsize*stride,groupe1,groupe2,com,stride);
      if ( localsize * stride > nmax ) FATAL("inconstistency found " << localsize*stride << " " << nmax << " " << name_vec); 
#else
      DistributeVecteur("tmp vector for dist-reduction",&tmp[0],groupe1,groupe2,com,stride);
#endif
      //compute sum
      for(unsigned int i=0; i < localsize*stride;++i)
        vec[i] += tmp[i];
    }
    
    //renvoie le resultat
    DUMP("retour du resultat",DBG_INFO);
#ifdef DEBUG_BRIDGE      
    DistributeVecteur(name_vec,vec,nmax,groupe2,groupe1,com,stride);
#else
    DistributeVecteur(name_vec,vec,groupe2,groupe1,com,stride);
#endif
  }

#ifdef DEBUG_BRIDGE
  void DistributeVecteur(char * name_vec,double * vec,unsigned int nmax,int groupe1,int groupe2,Communicator & com,int stride=1){
#else
  void DistributeVecteur(char * name_vec,double * vec,int groupe1,int groupe2,Communicator & com,int stride=1){
#endif
    std::map<int,std::list<blocks> > & scheme = compressed_scheme[sch_index];

    std::map<int,std::list<blocks> >::iterator it = scheme.begin();
    std::map<int,std::list<blocks> >::iterator last = scheme.end();

    while (it != last){
      int proc = (*it).first;
      std::list<blocks> & lblocks = (*it).second;
      std::list<blocks>::iterator it_block = lblocks.begin();
      std::list<blocks>::iterator last_block = lblocks.end();
      if (proc == -1){++it;continue;}

      //je fait une boucle sur les block a transférer
      for (it_block = lblocks.begin(); it_block!=last_block;++it_block){
        blocks & b = *it_block;
        if(com.amIinGroup(groupe1)){
          //if (b.end > localsize){FATAL("y'a un souci dans la mise ajour des blocks du duo");}
          DUMP("sending bloc " << b.start*stride << " - " << b.end*stride << " to proc " << proc << "(" << localsize*stride << ")",DBG_INFO);
#ifdef DEBUG_BRIDGE
          if (b.end*stride > nmax) FATAL("depassement detecte " << b.end*stride << " " << nmax << " " << name_vec); 
#endif
#ifdef DEBUG_BRIDGE
          {
            int nb_send = (b.end-b.start)*stride;
            com.SendIntegers(&nb_send,1,proc,groupe2);
          }
#endif
          com.SendDoubles(vec+b.start*stride,(b.end-b.start)*stride,proc,groupe2);
          DUMP("sent bloc " << b.start*stride << " - " << b.end*stride << " to proc " << com.RealRank(proc,groupe2) << "(" << localsize*stride << ")",DBG_INFO);
        }
        if(com.amIinGroup(groupe2)){
          //      if (b.end > localsize){FATAL("y'a un souci dans la mise ajour des blocks du duo");}
          //      sleep(1);
          DUMP("recepting bloc " << b.start*stride << " - " << b.end*stride << " from proc " << com.RealRank(proc,groupe1) << "(" << localsize*stride << ") writing to " << vec,DBG_INFO); 
        
#ifdef DEBUG_BRIDGE
          if (b.end*stride > nmax) FATAL("depassement detecte " << b.end*stride << " " << nmax << " " << name_vec); 
          {
            int nb_recv = 0;
            com.ReceiveIntegers(&nb_recv,1,proc,groupe1);
            if (nb_recv != (int)((b.end-b.start)*stride))
              FATAL("redistribution failed : proc " << com.RealRank(proc,groupe1) << " want to send me " << nb_recv << " doubles and i planed " <<  (b.end-b.start)*stride);
          }
#endif
          com.ReceiveDoubles(vec+b.start*stride,(b.end-b.start)*stride,proc,groupe1);
          //      sleep(1);
          DUMP("reception bloc " << b.start*stride << " - " << b.end*stride << " from proc " << com.RealRank(proc,groupe1) << "(" << localsize*stride << ")",DBG_INFO); 
        }
      }
      ++it;
    }
    com.waitForPendingComs();
  }

  void print(char * prefix){
    char name[255];

/*     if(something_changed == false){ */
/*       ++step; */
/*       return; */
/*     } */

    sprintf(name,"%s-redistrib-scheme%d-proc%d.mat",prefix,step,my_proc_id);
    FILE * fichier = fopen(name,"wb+");

    std::map<int,std::list<blocks> > & scheme = compressed_scheme[sch_index];

    std::map<int,std::list<blocks> >::iterator it = scheme.begin();
    std::map<int,std::list<blocks> >::iterator last = scheme.end();
    
    while (it != last){
      int proc = (*it).first;
      std::list<blocks> & lblocks = (*it).second;
      std::list<blocks>::iterator it_block = lblocks.begin();
      std::list<blocks>::iterator last_block = lblocks.end();
      
      int distant_index_start = 0;
      int distant_index_end = 0;
      //je fait une boucle sur les block a transférer
      for (it_block = lblocks.begin(); it_block!=last_block;++it_block){
        blocks & b = *it_block;
        distant_index_end += b.end-b.start;
        fprintf(fichier,"%d-%d %d %d %d\n",b.start,b.end,distant_index_start,distant_index_end,proc);
        distant_index_start += b.end-b.start;
      }
      ++it;
    }

    fprintf(fichier,"\n");
    fclose(fichier);

    ++step;
    something_changed = false;
/*     for (unsigned int i=0; i < localsize;++i) */
/*       std::cerr << i << " " << indirection[i] << "\n"; */
  }

  bool isInBlock(unsigned int i,blocks b){ 
    return b.start<=i && b.end>i; 
  } 

  void setLocalSize(unsigned int size){ 
    localsize = size;
  } 

/*   blocks & block(int i){return compressed_scheme[i];} */
/*   int nb_blocks(){return compressed_scheme.size();} */

/*   void deleteTmps(){ */
/*     delete [] tmp; */
/*     tmp = NULL; */
/*   }; */

  std::map<int,std::list<blocks> > & GetCompressedScheme(){
    return compressed_scheme[sch_index];
  }
  std::map<int,std::list<blocks> > & GetNewCompressedScheme(){
    return compressed_scheme[schNew_index];
  }


 private:

  bool compressed;
  unsigned int totalsize;
  unsigned int localsize;

  int * indirection;
  std::map<int,int> nb_par_interlocuteurs;

  
  int schNew_index;
  int sch_index;

  std::map<int,std::list<blocks> > compressed_scheme[2];

  std::map<int,std::vector<std::pair<int,int> > > sent;
  std::map<int,std::vector<int> > received;

  
  std::vector<double> tmp;

  int step;
  bool something_changed;
};

#endif //DUODISTRIBUTEDVECTEUR_H

---