implement TaskTrackerServer;

include "sys.m";
include "draw.m";
include "bufio.m";
include "tables.m";

include "ioutil.m";
include "mrutil.m";
include "../logger/logger.m";
include "tasktracker.m";


TaskTrackerServer : module {
	init : fn(ctxt : ref Draw->Context, args : list of string);
};

sys : Sys;
bufio : Bufio;
ioutil : IOUtil;
mrutil : MRUtil;
logger : Logger;
tasktracker : TaskTracker;

Connection : import sys;
Iobuf : import bufio;

MapperTask : import mrutil;
ReducerTask : import mrutil;
TaskTrackerInfo : import mrutil;

localAddr : string;
localPort := 66667;
hostAddr : string;
hostPort := 66666;
mapperFilePort := 70000;

mutex : chan of int;


init(ctxt : ref Draw->Context, args : list of string)
{
	sys = load Sys Sys->PATH;
	bufio = load Bufio Bufio->PATH;
	mrutil = load MRUtil MRUtil->PATH;
	logger = load Logger Logger->PATH;
	tasktracker = load TaskTracker TaskTracker->PATH;

	mrutil->init();
	tasktracker->init();

	logger->init();
	logger->setFileName("log_tasktrackerserver");

	mutex = chan[1] of int;

	buffer := bufio->open("/appl/MR/config", Bufio->OREAD);
	hostAddr = buffer.gets('\n');
	hostAddr = hostAddr[:len hostAddr - 1];
	localAddr = buffer.gets('\n');
	localAddr = localAddr[:len localAddr - 1];


	(ok, conn) := sys->announce("tcp!*!" + string localPort);
	if (ok < 0) {
		logger->log("TaskTrackerServer: announce failed!", Logger->ERROR);
		logger->scrlog("TaskTrackerServer: announce failed!", Logger->ERROR);
		exit;	
	}

	spawn heartBeat();

	while(1) {
		listen(conn);
	}
}

listen(conn : Connection)
{
	buf := array [Sys->ATOMICIO] of byte;
	(ok, c) := sys->listen(conn);
	if (ok < 0) {
		logger->log("TaskTrackerServer: listen failed!", Logger->ERROR);
		logger->scrlog("TaskTrackerServer: listen failed!", Logger->ERROR);
		exit;
	}
	rdf := sys->open(conn.dir + "/remote", Sys->OREAD);
	n := sys->read(rdf, buf, len buf);
	connHandle(c);
}

connHandle(conn : Connection)
{
	addr := array [Sys->ATOMICIO] of byte;
	msgStr := array [Sys->ATOMICIO] of byte;
	
	rdfd := sys->open(conn.dir + "/data", Sys->OREAD);
	wdfd := sys->open(conn.dir + "/data", Sys->OWRITE);
	rfd := sys->open(conn.dir + "/remote", Sys->OREAD);

	addrlen := sys->read(rfd, addr, len addr);
	logger->logInfo("Message from: " + string addr[:addrlen-1]);
	
	msglen := sys->read(rdfd, msgStr, len msgStr);
	receive : while (msglen > 0) {
		(nil, msg) := sys->tokenize(string msgStr[:msglen], "@");
		op := hd msg;
		msg = tl msg;
		case (op) {
			"shootMapper" => {
				mapper := mrutil->msg2mapper(msg);
				spawn runMapper(mapper);
			}
			"shootReducer" => {
				mapperFileAddr := hd msg;
				msg = tl msg;
				reducer := mrutil->msg2reducer(msg);
				spawn runReducer(mapperFileAddr, reducer);
			}
		}
	msglen = sys->read(rdfd, msgStr, len msgStr);
	}
}

heartBeat()
{
	while (1) {
		sys->sleep(3000);
		(n, conn) := sys->dial("tcp!" + hostAddr + "!" + string hostPort, nil);	
		tracker := ref TaskTrackerInfo(localAddr, localPort, 0, 0, 1, 0);
		msg := "updateTaskTracker" + "@" + mrutil->tracker2msg(tracker);		
		sys->fprint(conn.dfd, "%s", msg);
	}
}

runMapper(mapper : ref MapperTask)
{
	mutex <- = 0;
	ok := tasktracker->runMapperTask(mapper);
	<- mutex;
	msg : string;
	if (ok == 0) {
		logger->logInfo("MapperTask " + string mapper.id + " from job " + string mapper.jobId + " succeed!");
		logger->scrlogInfo("MapperTask " + string mapper.id + " from job " + string mapper.jobId + " succeed!");

		msg = "mapperSucceed";
	}
	else {
		logger->logInfo("MapperTask " + string mapper.id + " from job " + string mapper.jobId + " failed!");
		logger->scrlogInfo("MapperTask " + string mapper.id + " from job " + string mapper.jobId + " failed!");
		
		msg = "mapperFailed";
	}

	(n, conn) := sys->dial("tcp!" + hostAddr + "!" + string hostPort, nil);	
	msg = msg + "@" + "tcp!" + localAddr + "!" + string (mapperFilePort + mapper.id)+ "@" + mrutil->mapper2msg(mapper);
	sys->fprint(conn.dfd, "%s", msg);

	buf := array [Sys->ATOMICIO] of byte;
	length := sys->read(conn.dfd, buf, len buf);
	logger->logInfo("Feed back " + string buf[:length]);
	logger->scrlogInfo("Feed back " + string buf[:length]);
}

runReducer(mapperFileAddr : string, reducer : ref ReducerTask)
{
	mutex <- = 0;
	(ok, failedAddr) := tasktracker->runReducerTask(mapperFileAddr, reducer);
	<- mutex;
	msg : string;
	if (ok == 0) {  	#succeed
		logger->logInfo("ReducerTask " + string reducer.id + " from job " + string reducer.jobId + " succeed!");
		logger->scrlogInfo("ReducerTask " + string reducer.id + " from job " + string reducer.jobId + " succeed!");

		msg = "reducerSucceed";
	}
	else if (ok == 1) { 	#still pending
		logger->logInfo("ReducerTask " + string reducer.id + " from job " + string reducer.jobId + " get mapper address!");
		logger->scrlogInfo("ReducerTask " + string reducer.id + " from job " + string reducer.jobId + " get mapper address!");

		return;
	}
	else {
		logger->logInfo("ReducerTask " + string reducer.id + " from job " + string reducer.jobId + " failed!");
		logger->scrlogInfo("ReducerTask " + string reducer.id + " from job " + string reducer.jobId + " failed!");
		
		msg = "reducerFailed" + "@" + failedAddr;
	}

	(n, conn) := sys->dial("tcp!" + hostAddr + "!" + string hostPort, nil);	
	msg = msg + "@" + mrutil->reducer2msg(reducer);
	sys->fprint(conn.dfd, "%s", msg);

	buf := array [Sys->ATOMICIO] of byte;
	length := sys->read(conn.dfd, buf, len buf);
	logger->logInfo("Feed back " + string buf[:length]);
	logger->scrlogInfo("Feed back " + string buf[:length]);
}