mr_worker.cc

#include <iostream>
#include <fstream>
#include <sstream>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <dirent.h>
#include <sys/file.h>

#include <mutex>
#include <string>
#include <vector>
#include <map>

#include <sys/time.h>

#include "rpc.h"
#include "mr_protocol.h"

using namespace std;

struct KeyVal {
    string key;
    string val;
};

//
// The map function is called once for each file of input. The first
// argument is the name of the input file, and the second is the
// file's complete contents. You should ignore the input file name,
// and look only at the contents argument. The return value is a slice
// of key/value pairs.
//
vector<KeyVal> Map(const string &filename, const string &content)
{
	vector<KeyVal> wordsMap;
    string tempString=content;
    for(int i=0;i<(int)tempString.size();i++)
        if(!(tempString[i]>='a'&&tempString[i]<='z')&&!(tempString[i]>='A'&&tempString[i]<='Z'))
            tempString[i]=' ';
    istringstream wordsStream(tempString);
    string tempKey;
    KeyVal keyVal;
    while(wordsStream >> tempKey){
        keyVal.key=tempKey;
        keyVal.val="1";
        wordsMap.push_back(keyVal);
    }
    return wordsMap;
}

//
// The reduce function is called once for each key generated by the
// map tasks, with a list of all the values created for that key by
// any map task.
//
string Reduce(const string &key, const vector < string > &values)
{
	// Copy your code from mr_sequential.cc here.
	// Lab2: Your code here.
	return "";
}


typedef vector<KeyVal> (*MAPF)(const string &key, const string &value);
typedef string (*REDUCEF)(const string &key, const vector<string> &values);

class Worker {
public:
	Worker(const string &dst, const string &dir, MAPF mf, REDUCEF rf);

	void doWork();

private:
	void doMap(int index, const string &filename);
	void doReduce(int index, int mapNum);
	void doSubmit(mr_tasktype taskType, int index);

	mutex mtx;
	int id;

	rpcc *cl;
	std::string basedir;
	MAPF mapf;
	REDUCEF reducef;
};


Worker::Worker(const string &dst, const string &dir, MAPF mf, REDUCEF rf)
{
	this->basedir = dir;
	this->mapf = mf;
	this->reducef = rf;

	sockaddr_in dstsock;
	make_sockaddr(dst.c_str(), &dstsock);
	this->cl = new rpcc(dstsock);
	if (this->cl->bind() < 0) {
		printf("mr worker: call bind error\n");
	}
}

void Worker::doMap(int index, const string &filename)
{
	// Lab2: Your code goes here.
	// cout<<"MAP "<<index<<" "<<filename<<" start"<<endl;
	string content;
	getline(ifstream(filename), content, '\0');
    vector <KeyVal> KVA = Map(filename, content);

	vector<string> outStrings(REDUCER_COUNT,"");
	for(const KeyVal& keyVal : KVA){
		std::hash<std::string> hash_string;
    	int fileIndex = hash_string(keyVal.key)%REDUCER_COUNT;
		outStrings[fileIndex]+=keyVal.key+" "+keyVal.val+"\n";
	}

	for(int j=0; j<REDUCER_COUNT; j++){
		string filenameOut=basedir+"mr-"+to_string(index)+"-"+to_string(j);
		ofstream out(filenameOut, ios::out);
		out<<outStrings[j];
		out.close();
		// cout<<"MAP "<<index<<" output to"<<" "<<filenameOut<<endl;
	}
}

void Worker::doReduce(int index, int mapNum)
{
	// cout<<"REDUCE "<<index<<" start"<<endl;
	map<string, unsigned long> kvmap;
	for(int j=0; j<mapNum; j++){
		string filename="mr-"+to_string(j)+"-"+to_string(index);
		ifstream in(filename, ios::in);
		string key, val;
		while (in>>key>>val)
			kvmap[key]+=stol(val);
		in.close();
		// cout<<"REDUCE "<<index<<" read "<<filename<<endl;

	}
	string outString="";
	for(const pair<string,unsigned long>& kv:kvmap)
		outString+=kv.first+" "+to_string(kv.second)+"\n";
	
	string mr_out=basedir+"mr-out-"+to_string(index);

	ofstream out(mr_out,ios::out|ios::app);

    // class Helper : public std::filebuf {
    // public:
    //   int handle() { return _M_file.fd(); }
    // };

  	// int fd = static_cast<Helper&>(*out.rdbuf()).handle();
	// flock(fd, LOCK_EX);
	out<<outString<<endl;
	out.close();
	// flock(fd, LOCK_UN);
	// cout<<"REDUCE "<<index<<" write to "<<mr_out<<endl;
}

void Worker::doSubmit(mr_tasktype taskType, int index)
{
	bool b;
	mr_protocol::status ret = this->cl->call(mr_protocol::submittask, taskType, index, b);
	if (ret != mr_protocol::OK) {
		fprintf(stderr, "submit task failed\n");
		exit(-1);
	}
}

void Worker::doWork()
{
	for (;;) {
		string reply, type, index, filename, mapNum;
		cl->call(mr_protocol::asktask,id,reply);
		istringstream strstream(reply);
		strstream>>type>>index>>mapNum>>filename;
		// cout<<"get "<<"type: "<<type<<" index: "<<index<<" mapNum: "<<mapNum<<" filename: "<<filename<<endl;
		switch (stoi(type))
		{
		case 1:
			doMap(stoi(index),filename);
			doSubmit(mr_tasktype::MAP,stoi(index));
			break;
		case 2:
			doReduce(stoi(index),stoi(mapNum));
			doSubmit(mr_tasktype::REDUCE,stoi(index));
			break;
		default:
			sleep(1);
			break;
		}
		//
		// Lab2: Your code goes here.
		// Hints: send asktask RPC call to coordinator
		// if mr_tasktype::MAP, then doMap and doSubmit
		// if mr_tasktype::REDUCE, then doReduce and doSubmit
		// if mr_tasktype::NONE, meaning currently no work is needed, then sleep
		//
		

	}
}

int main(int argc, char **argv)
{
	if (argc != 3) {
		fprintf(stderr, "Usage: %s <coordinator_listen_port> <intermediate_file_dir> \n", argv[0]);
		exit(1);
	}

	MAPF mf = Map;
	REDUCEF rf = Reduce;
	
	Worker w(argv[1], argv[2], mf, rf);
	w.doWork();

	return 0;
}