- Main Page
- Classes
- Files

- File List

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

    00001 #ifndef MaPI_MAPREDUCE_HPP
    00002 #define MaPI_MAPREDUCE_HPP
    00003
    00005
    00008 template<class KeyA, class A, class KeyB, class B, class C>
    00009 class MaPI_MapReduce : public MapReduce<KeyA,A,KeyB,B,C>
    00010 {
    00011 public:
    00013         virtual vector< pair<KeyB,C> > run( Mapper<KeyA,A,KeyB,B,C> & mapper, Reducer<KeyA,A,KeyB,B,C> & reducer, vector< pair<KeyA,A> > & as)
    00014         {
    00015                 if (mappers.size() == 0 || reducers.size() == 0 ) {
    00016                         cout << "Aborting: It is necessary to registry almost one mapper and almost one reducer\n";
    00017                         return vector< pair<KeyB,C> >();
    00018                 }
    00019                 //cout << "Input  :\t" << as << endl;
    00020 #ifndef MRI_USE_MULTIMAP
    00021                 vector< pair<KeyB,B> > bs = mapper.run( as );
    00022 #else
    00023                 multimap<KeyB,B> bs = mapper.run( as );
    00024 #endif
    00025                 //cout << "Mapped :\t" << bs << endl;
    00026                 vector< pair<KeyB,C> > cs = reducer.run( bs );
    00027                 //cout << "Reduced:\t" << cs << endl;
    00028                 return cs;
    00029         };
    00031         MaPI_MapReduce(){};
    00033         MaPI_MapReduce(MaPI_Mapper<KeyA,A,KeyB,B,C> * _mapper, MaPI_Reducer<KeyA,A,KeyB,B,C> * _reducer)
    00034         {
    00035                 mappers.push_back(_mapper);
    00036                 reducers.push_back(_reducer);
    00037         };
    00039         void registry(MaPI_Mapper<KeyA,A,KeyB,B,C> * _mapper, MaPI_Reducer<KeyA,A,KeyB,B,C> * _reducer)
    00040         {
    00041                 mappers.push_back(_mapper);
    00042                 reducers.push_back(_reducer);
    00043         };
    00045         void registry(MaPI_Mapper<KeyA,A,KeyB,B,C> * _mapper) { mappers.push_back(_mapper); };
    00047         void registry(MaPI_Reducer<KeyA,A,KeyB,B,C> * _reducer) { reducers.push_back(_reducer); };
    00049         void initServers(int argc, char **argv)
    00050         {
    00051             MPI_Init(&argc,&argv);
    00052             MPI_Comm_size(MPI_COMM_WORLD,&MRSize);
    00053             MPI_Comm_rank(MPI_COMM_WORLD,&MRRank);
    00054             if (MRRank!=0) server();
    00055         };
    00057         virtual ~MaPI_MapReduce(){ finalize(); };
    00059         int id(MaPI_Mapper<KeyA,A,KeyB,B,C> * _mapper){ int i=0; while (mappers[i]!=_mapper) i++; return i; };
    00061         int id(MaPI_Reducer<KeyA,A,KeyB,B,C> * _reducer){ int i=0; while (reducers[i]!=_reducer) i++; return i; };
    00063         int getMPISize(){return MRSize;};
    00065         int getMPIRank(){return MRRank;};
    00067         MPI_Status * getMPIStatus(){return &MRStat;};
    00068 protected:
    00069         vector<MaPI_Mapper<KeyA,A,KeyB,B,C> * > mappers;
    00070         vector<MaPI_Reducer<KeyA,A,KeyB,B,C> * > reducers;
    00071         int MRSize,MRRank;
    00072         MPI_Status MRStat;
    00073         void server()
    00074         {
    00075             while (true)
    00076             {
    00077                 int operation; // 0: to terminate ; 1: to map ; 2: to reduce ;
    00078                 MPI_Recv(&operation, 1, MPI_INT, 0, 1, MPI_COMM_WORLD, &MRStat);
    00079                 if (operation == 0) { MPI_Finalize(); exit(0); }
    00080
    00081                 int object_id; // where the object is an instance of Mapper or an instance of Reducer
    00082                 MPI_Recv(&object_id, 1, MPI_INT, 0, 1, MPI_COMM_WORLD, &MRStat);
    00083
    00084                 int input_size;
    00085                 MPI_Recv(&input_size, 1, MPI_INT, 0, 1, MPI_COMM_WORLD, &MRStat);
    00086
    00087                 char input[input_size]; // Receive a serialization of "pair<KeyA,A>" or "pair<KeyB, vector<B> >"
    00088                 MPI_Recv(&input, input_size, MPI_CHAR, 0, 1, MPI_COMM_WORLD, &MRStat);
    00089
    00090                         if (operation == 1) // Mapping
    00091                         {
    00092                                 if (object_id >= mappers.size()) { cout << "\nERROR: Overflow at map server\n"; exit(1); }
    00093
    00094                                 //printf("Server mapper called\n");
    00095
    00096                                 stringstream _output;
    00097                                 if (input[0]=='_')
    00098                                         _output << "_";
    00099                                 else {
    00100                                         pair<KeyA,A> a;
    00101
    00102                                         Scanner scan(input);
    00103                                         scan.useSeparators("\1");
    00104
    00105                                         a.first  = mappers[object_id]->getSerializer()->KeyA_fromString(scan.next());
    00106                                         a.second = mappers[object_id]->getSerializer()->A_fromString(scan.next());
    00107
    00108                                         //cout << "#" << a << endl;
    00109
    00110                                         vector< pair<KeyB,B> > mapped = mappers[object_id]->map( a );
    00111
    00112                                         typename vector< pair<KeyB,B> >::iterator it;
    00113                                         for (it = mapped.begin() ; it != mapped.end() ; ++it)
    00114                                         {
    00115                                                 _output  << mappers[object_id]->getSerializer()->KeyB_toString((*it).first) << "\1"
    00116                                                                  << mappers[object_id]->getSerializer()->B_toString((*it).second) << "\1" ;
    00117                                         }
    00118
    00119                                         //cout << "_output: " << _output.str() << endl;
    00120                                 }
    00121
    00122                         int output_size = _output.str().size() + 1;
    00123                 char output[output_size];
    00124                 strcpy(output,_output.str().c_str());
    00125                         MPI_Send(&output_size, 1, MPI_INT, 0, 1, MPI_COMM_WORLD);
    00126                         MPI_Send(output, output_size, MPI_CHAR, 0, 1, MPI_COMM_WORLD);
    00127                         }
    00128                         else
    00129                         if (operation == 2) // Reducing
    00130                         {
    00131                                 if (object_id >= reducers.size()) { cout << "\nERROR: Overflow at reduce server\n"; exit(1); }
    00132
    00133                                 //printf("Server reductor called\n");
    00134
    00135                                 //cout << input << endl;
    00136
    00137                                 stringstream _output;
    00138                                 if (input[0]=='_')
    00139                                         _output << "_";
    00140                                 else {
    00141                                         pair < KeyB,vector<B> > b;
    00142
    00143                                         Scanner scan(input);
    00144                                         scan.useSeparators("\1");
    00145
    00146                                         b.first = reducers[object_id]->getSerializer()->KeyB_fromString(scan.next());
    00147                                         while (scan.hasNext()) b.second.push_back(reducers[object_id]->getSerializer()->B_fromString(scan.next()));
    00148
    00149                                         //cout << b.first << endl;
    00150                                         //for (int i = 0 ; i < b.second.size() ; i++) cout << "#" << b.second[i] << endl;
    00151
    00152                                         pair < KeyB , C > reduced = reducers[object_id]->reduce( b );
    00153                                         _output << reducers[object_id]->getSerializer()->KeyB_toString(reduced.first) << "\1"
    00154                                                         << reducers[object_id]->getSerializer()->C_toString(reduced.second) << "\1";
    00155                                         //cout << "_output: " << _output.str() << endl;
    00156                                 }
    00157
    00158                         int output_size = _output.str().size() + 1;
    00159                 char output[output_size];
    00160                 strcpy(output,_output.str().c_str());
    00161                         MPI_Send(&output_size, 1, MPI_INT, 0, 1, MPI_COMM_WORLD);
    00162                         MPI_Send(output, output_size, MPI_CHAR, 0, 1, MPI_COMM_WORLD);
    00163                         }
    00164             }
    00165         };
    00166         void finalize()
    00167         {
    00168             int operation = 0;
    00169             for (int i = 1; i < MRSize; i++) MPI_Send(&operation, 1, MPI_INT,  i, 1, MPI_COMM_WORLD);
    00170             MPI_Finalize();
    00171         }
    00172 };
    00173
    00175 class MaPI_StrMapReduce : public MaPI_MapReduce<string,string,string,string,string> {};
    00176
    00177 #endif /* MaPI_MAPREDUCE_HPP */

All Classes Functions