- Main Page
- Classes
- Files

- File List

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

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

All Classes Functions