- Main Page
- Classes
- Files

- File List

# OptFrame/Parallel/MapReduce++/MaPI/reducer.hpp

    00001 #ifndef MaPI_REDUCER_HPP
    00002 #define MaPI_REDUCER_HPP
    00003
    00005 template<class KeyA, class A, class KeyB, class B, class C>
    00006 class MaPI_Reducer : public Reducer<KeyA,A,KeyB,B,C>
    00007 {
    00008 public:
    00010         MaPI_Reducer(MaPI_MapReduce<KeyA,A,KeyB,B,C> * _mapReduce, MaPI_Serializer<KeyA,A,KeyB,B,C> * _serializer)
    00011                 :mapReduce(_mapReduce),serializer(_serializer){ mapReduce->registry(this); };
    00013 #ifndef MRI_USE_MULTIMAP
    00014         virtual vector< pair<KeyB,C> > run( vector< pair<KeyB,B> > & _mapped)
    00015         {
    00016                 multimap<KeyB,B> mapped;
    00017                 for (int i = 0 ; i < _mapped.size() ; i++) mapped.insert(_mapped[i]);
    00018 #else
    00019         virtual vector< pair<KeyB,C> > run( multimap<KeyB,B> & mapped)
    00020         {
    00021 #endif
    00022
    00023                 vector< pair<KeyB,C> > reduced;
    00024
    00025                 typename std::multimap<KeyB,B>::iterator it = mapped.begin();
    00026                 KeyB lastKey = (*it).first;
    00027                 vector< pair < KeyB,vector<B> > > toReduce(1, pair < KeyB,vector<B> >( lastKey , vector<B>() ) );
    00028                 for( ; it != mapped.end() ; ++it)
    00029                 {
    00030                         if ((*it).first != lastKey) {
    00031                                 toReduce.push_back( pair< KeyB,vector<B> > ( (*it).first , vector<B>() ) );
    00032                                 lastKey = (*it).first;
    00033                         }
    00034                         toReduce.back().second.push_back( (*it).second );
    00035                 }
    00036
    00037                 // Parallel Reduction
    00038
    00039             vector<string> outputs;
    00040
    00041                 vector<string> inputs;
    00042                 typename vector< pair < KeyB,vector<B> > >::iterator itr;
    00043                 for (itr = toReduce.begin() ; itr != toReduce.end() ; ++itr)
    00044                 {
    00045                         stringstream s; s << (*itr).first << "\1";
    00046                         for (int i = 0 ; i < (*itr).second.size() ; i++) s << (*itr).second[i] << "\1";
    00047                         inputs.push_back(s.str());
    00048                 }
    00049
    00050             int numMapProcs = mapReduce->getMPISize()-1;
    00051             int numInputs   = inputs.size();
    00052             int reducerId       = mapReduce->id(this);
    00053                 int operation   = 2;
    00054
    00055             // Caso 1 - Número de entradas menor ou igual ao número de processadores (n-1)
    00056         if (numInputs <= numMapProcs)
    00057                 {
    00058                 // Mensagem
    00059                 // enviar para os processos
    00060                 for (int i = 0; i < numInputs; i++)
    00061                 {   //cout << "Enviando" << i << endl;
    00062                 // enviar
    00063
    00064                 int  stsize = inputs.at(i).size() + 1;
    00065                 char st[stsize];
    00066                 strcpy(st,inputs.at(i).c_str());
    00067
    00068                 MPI_Send(&operation, 1, MPI_INT, i+1, 1, MPI_COMM_WORLD);
    00069                 MPI_Send(&reducerId, 1, MPI_INT, i+1, 1, MPI_COMM_WORLD);
    00070                 MPI_Send(&stsize, 1, MPI_INT, i+1, 1, MPI_COMM_WORLD);
    00071                 MPI_Send(st, stsize, MPI_CHAR, i+1, 1, MPI_COMM_WORLD);
    00072                 }
    00073
    00074                 // enviar _ para os demais processos
    00075             for (int i = numInputs; i < numMapProcs; i++)
    00076             {   //cout << "Enviando_" << i << endl;
    00077                 // enviar
    00078                 int  stsize = 1;
    00079                 char st[] = "_";
    00080                 MPI_Send(&operation, 1, MPI_INT, i+1, 1, MPI_COMM_WORLD);
    00081                 MPI_Send(&reducerId, 1, MPI_INT, i+1, 1, MPI_COMM_WORLD);
    00082                 MPI_Send(&stsize, 1, MPI_INT, i+1, 1, MPI_COMM_WORLD);
    00083                 MPI_Send(st, stsize, MPI_CHAR, i+1, 1, MPI_COMM_WORLD);
    00084             }
    00085
    00086             // receber
    00087             for (int i = 1; i <= numMapProcs; i++)
    00088             {
    00089                 // enviar
    00090                 int stsize;
    00091                 MPI_Recv(&stsize, 1, MPI_INT, i, 1, MPI_COMM_WORLD, mapReduce->getMPIStatus());
    00092                 char st[stsize];
    00093                 MPI_Recv(&st, stsize, MPI_CHAR, i, 1, MPI_COMM_WORLD, mapReduce->getMPIStatus());
    00094                 if (st[0] != '_') outputs.push_back(string(st));
    00095                                 //cout << "st: " << st << endl;
    00096             }
    00097         }
    00098
    00099         else
    00100
    00101         // Caso 2 - Número de entradas maior que o número de processadores (n-1)
    00102             {
    00103                 // Mensagem
    00104             // enviar para os processos
    00105             for (int i = 0; i < numInputs; i++)
    00106             {
    00107                 int dest = i % numMapProcs + 1;
    00108                 //cout << "EnviandoDest " << dest << endl;
    00109                 // enviar
    00110                 int  stsize = inputs.at(i).size();
    00111                 char st[stsize];
    00112                 strcpy(st,inputs.at(i).c_str());
    00113                     MPI_Send(&operation, 1, MPI_INT, dest, 1, MPI_COMM_WORLD);
    00114                 MPI_Send(&reducerId, 1, MPI_INT, dest, 1, MPI_COMM_WORLD);
    00115                 MPI_Send(&stsize, 1, MPI_INT,  dest, 1, MPI_COMM_WORLD);
    00116                     MPI_Send(st, stsize, MPI_CHAR, dest, 1, MPI_COMM_WORLD);
    00117             }
    00118
    00119             // receber
    00120             for (int i = 0; i < numInputs; i++)
    00121             {   int from = i % numMapProcs + 1;
    00122                 int stsize;
    00123                 MPI_Recv(&stsize, 1, MPI_INT, from, 1, MPI_COMM_WORLD, mapReduce->getMPIStatus());
    00124                 char st[stsize];
    00125                 MPI_Recv(&st, stsize, MPI_CHAR, from, 1, MPI_COMM_WORLD, mapReduce->getMPIStatus());
    00126                 if (st[0] != '_') outputs.push_back(string(st));
    00127             }
    00128         }
    00129
    00130                 for (int i = 0 ; i < outputs.size() ; i++)
    00131                 {
    00132                         Scanner scan(outputs[i]);
    00133                         scan.useSeparators("\1");
    00134
    00135                         while (scan.hasNext())
    00136                         {
    00137                                 pair<KeyB,C> p;
    00138                                 p.first = getSerializer()->KeyB_fromString(scan.next());
    00139                                 p.second = getSerializer()->C_fromString(scan.next());
    00140                                 reduced.push_back(p);
    00141                         }
    00142                 }
    00143
    00144                 return reduced;
    00145         };
    00147         virtual pair<KeyB,C> reduce( pair<KeyB, vector<B> > ) = 0 ;
    00148
    00149         MaPI_Serializer<KeyA,A,KeyB,B,C> * getSerializer(){return serializer;};
    00150
    00151         void setSerializer(MaPI_Serializer<KeyA,A,KeyB,B,C> * _serializer){serializer = _serializer;};
    00152
    00153 protected:
    00154         MaPI_MapReduce<KeyA,A,KeyB,B,C> * mapReduce;
    00155         MaPI_Serializer<KeyA,A,KeyB,B,C> * serializer;
    00156 };
    00157
    00159 class MaPI_StrReducer : public MaPI_Reducer<string,string,string,string,string>
    00160 {
    00161 public:
    00163         MaPI_StrReducer(MaPI_MapReduce<string,string,string,string,string> * _mapReduce)
    00164                 :MaPI_Reducer<string,string,string,string,string>(_mapReduce,NULL) { serializer = new MaPI_StrSerializer(); deleteSerializer = true; };
    00166         MaPI_StrReducer(MaPI_MapReduce<string,string,string,string,string> * _mapReduce,
    00167                                         MaPI_Serializer<string,string,string,string,string> * _serializer)
    00168                 :MaPI_Reducer<string,string,string,string,string>(_mapReduce,_serializer) { deleteSerializer = false; };
    00170         virtual ~MaPI_StrReducer(){ if (deleteSerializer) delete serializer; };
    00171 private:
    00172         bool deleteSerializer;
    00173 };
    00174
    00175 #endif /* MaPI_REDUCER_HPP */

All Classes Functions