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 <mutex>
#include <string>
#include <vector>
#include <map>

#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)
{
	// Copy your code from mr_sequential.cc here.
    int pos = 0;
    vector<KeyVal> res;

    while (pos < content.size())
    {
        if (!isalpha(content[pos])) {
            ++pos;
            continue;
        }

        KeyVal kv;
        kv.val = "1";
        while(isalpha(content[pos])){
            kv.key += content[pos];
            ++pos;
        }
        res.push_back(kv);
    }

    return res;
}

//
// 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.
    return to_string(values.size());

}


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 vector<string> &filenames);
	void doReduce(int index);
	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 vector<string> &filenames)
{
	// Lab2: Your code goes here.
	vector<KeyVal> kv;
	ifstream in(filenames[0]);
	ofstream out[REDUCER_COUNT];
	string content;
	ostringstream tmp;

	for (int i = 0; i < REDUCER_COUNT; ++i) {
		out[i].open(this->basedir + "mr-" + to_string(index) + "-" + to_string(i));
		if (!out[i].is_open()) {
			fprintf(stderr, "create file error.\n");
			exit(-1);
		}
	}

	tmp << in.rdbuf();
	content = tmp.str();

	kv = this->mapf("nonesense", content);

	hash<string> hash_str;
	for (auto iter : kv) {
		int hashValue = hash_str(iter.key) % REDUCER_COUNT;
		out[hashValue] << iter.key << " " << iter.val << " ";
	}

	in.close();
	for (int i = 0; i < REDUCER_COUNT; ++i) {
		out->close();
	}

	return;
}

void Worker::doReduce(int index)
{
	// Lab2: Your code goes here.
	DIR *dir;
	dirent *ptr;
	vector<string> reduceFile;

	dir = opendir(this->basedir.c_str());
	while ((ptr = readdir(dir)) != nullptr)
	{
		string filename = ptr->d_name;
		int beginPos = filename.find_last_of('-');

		if(filename.substr(beginPos + 1) == to_string(index)){
			reduceFile.push_back(this->basedir + filename);
		}
	}
	closedir(dir);

	ofstream out(this->basedir + "mr-out-" + to_string(index));
	map<string, vector<string>> classifiedKV;
	
	for (auto file : reduceFile) {
		ifstream in(file);

		vector<KeyVal> kv;

		KeyVal tmp;
		while (in >> tmp.key >> tmp.val) {
			kv.push_back(tmp);
		}

		for (auto iter : kv) {
			classifiedKV[iter.key].push_back(iter.val);
		}
		
		in.close();
	}

	for (auto iter = classifiedKV.begin(); iter != classifiedKV.end(); ++iter) {
		out << iter->first << " " << this->reducef(iter->first, iter->second) << "\n";
	}

	out.close();
}

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 (;;) {

		//
		// 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 nonesense = 0;
		mr_protocol::AskTaskResponse reply;

		mr_protocol::status ret = this->cl->call(mr_protocol::asktask, nonesense, reply);
		if(ret != mr_protocol::OK) {
			fprintf(stderr, "ask task failed\n");
			exit(-1);
		}

		vector<string> filenames;

		switch (reply.taskType)
		{
		case mr_tasktype::MAP:
			filenames.push_back(reply.filename);
	
			this->doMap(reply.taskIndex, filenames);
			printf("finish map\n");
			this->doSubmit(mr_tasktype(reply.taskType), reply.taskIndex);
			printf("finish submit map\n");
			break;
		case mr_tasktype::REDUCE:
			this->doReduce(reply.taskIndex);
			this->doSubmit(mr_tasktype(reply.taskType), reply.taskIndex);
			break;
		case mr_tasktype::NONE:
			sleep(1);
			break;
		default:
			fprintf(stderr, "unknown task type.\n");
			exit(-1);
			break;
		}

	}
}

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;
}