##########################################	The implemention of the MapperWorker module.#	the mapreduce module and inputFileBlock are downloaded from DFS.#	collect() collect the pair of (key , value) into OutputCollector;#	saveToLocal() sort and combine the pairs in OutputCollector and write it into a local file.##	@author Yang Fan(fyabc) #	@author Guanji Gao(ggj)######################################### implement MapperWorker; include "sys.m";include "draw.m";include "lists.m";include "tables.m";include "sort.m";include "bufio.m"; include "mapperworker.m";include "mapreduce.m";include "mrutil.m";include "ioutil.m"; include "../dfs/dfsutil.m";include "../dfs/dfsclient.m"; sys : Sys;lists : Lists;sort : Sort;bufio : Bufio;tables : Tables; mapreduce : MapReduce;ioutil : IOUtil;mrutil : MRUtil;dfsutil : DFSUtil;dfsclient : DFSClient; FD : import sys;Iobuf : import bufio;MapperTask : import mrutil;OutputCollector : import ioutil;DFSFile : import dfsutil;DFSChunkCmp : import dfsutil;Strhash : import tables; mapperPath := "/appl/MR/task/";mapperAddr : string; downloadMutex : chan of int;downloadMutex2 : chan of int; init(){	sys = load Sys Sys->PATH;	lists = load Lists Lists->PATH;	sort = load Sort Sort->PATH;	bufio = load Bufio Bufio->PATH; 	ioutil = load IOUtil IOUtil->PATH;	mrutil = load MRUtil MRUtil->PATH; 	dfsclient = load DFSClient DFSClient->PATH;	dfsutil = load DFSUtil DFSUtil->PATH; 	tables = load Tables Tables->PATH; 	ioutil->init();	mrutil->init();	dfsutil->init(); 	buffer := bufio->open("/appl/MR/config", Bufio->OREAD);	buffer.gets('\n');	mapperAddr = buffer.gets('\n');	mapperAddr = mapperAddr[: len mapperAddr - 1]; 	downloadMutex = chan [1] of int;	downloadMutex2 = chan [1] of int;} run(mapperTask : ref MapperTask){	downloadMutex <-= 0;	getmr(mapperTask);	<-downloadMutex; 	folderName := mapperPath + "tasks_" + string mapperTask.id + "/";	sys->create(folderName, sys->OREAD, sys->DMDIR + 8r777); 	downloadMutex2 <-= 0;	getFileBlock(mapperTask);	<-downloadMutex2; 	buffer := bufio->open(folderName + mapperTask.inputFileBlock.fileName + "_inputFileBlock", Bufio->OREAD); 	collector := collect(buffer); 	saveToLocal(mapperTask, collector, folderName); 	spawn ioutil->sendRemoteFile(70000 + mapperTask.id,folderName);}  collect(buffer : ref Iobuf) : ref OutputCollector{	collector := ref OutputCollector(nil);	collector.collection = Strhash[ref IOUtil->KVs].new(10000, nil);	line : string; 	while((line = buffer.gets('\n')) != nil)	{		kvList := mapreduce->filt(line);		for ( ; kvList != nil ; kvList = tl kvList)		{			(key , value) := hd kvList;			mapreduce->map(key, value, collector);		}	} 	return collector;} saveToLocal(mapperTask : ref MapperTask, collector : ref OutputCollector, folderName : string){	fds := array[mapperTask.reducerAmount] of ref FD;	buffers := array[mapperTask.reducerAmount] of ref Iobuf;	for(i := 0; i < mapperTask.reducerAmount; i++)	{		fds[i] = sys->create(folderName + "tcp!" +  mapperAddr + "!" + string (70000 + mapperTask.id) + "_part_" + string i, sys->ORDWR, 8r600);		buffers[i] = bufio->open(folderName + "tcp!" +  mapperAddr + "!" + string (70000 + mapperTask.id) + "_part_" + string i, bufio->OWRITE);	} 	recordMap := collector.getMap(); 	mapSize := len recordMap;	keyRange : int;	if(mapreduce->keySpaceSize() % mapperTask.reducerAmount == 0)		keyRange = mapreduce->keySpaceSize() / mapperTask.reducerAmount;	else		keyRange = mapreduce->keySpaceSize() / mapperTask.reducerAmount + 1; 	for(i = 0; i < mapSize; i++)	{		key := recordMap[i].key;		values := recordMap[i].values; 		j := mapreduce->hashKey(key) / keyRange; 		if (mapperTask.combinable == 1) {			(key_cb, value_cb) := mapreduce->combine(key, values);			buffers[j].puts(key_cb + " " + value_cb + "\n");		}		else {			for( ; values != nil; values = tl values)			{				value := hd values;				buffers[j].puts(key + " " + value + "\n");			}		}	} 	for(i = 0; i < mapperTask.reducerAmount; i++)	{		buffers[i].flush();		buffers[i].close();	}} getmr(mapperTask : ref MapperTask){	if(sys->open(mapperPath + mapperTask.mrClassName, Sys->OREAD) == nil)	{		dfsclient->init();		file := dfsclient->getFile(mapperTask.mrClassName);		total := (lists->last(file.chunks)).offset + big (lists->last(file.chunks)).size;		fd := sys->create(mapperPath + mapperTask.mrClassName, Sys->ORDWR, 8r600);		dd(fd, file,big 0,total);	}	mapreduce = load MapReduce (mapperPath + mapperTask.mrClassName);	mapreduce->init();} getFileBlock(mapperTask : ref MapperTask){	folderName := mapperPath + "tasks_" + string mapperTask.id + "/";	dfsclient->init();	file := dfsclient->getFile(mapperTask.inputFileBlock.fileName);	fd := sys->create(folderName + mapperTask.inputFileBlock.fileName + "_inputFileBlock", Sys->ORDWR, 8r600);	dd(fd, file, mapperTask.inputFileBlock.offset, big mapperTask.inputFileBlock.size);} dd(fd : ref Sys->FD, file : ref DFSFile, offset : big, size : big){	if (dfsclient == nil)		dfsclient = load DFSClient DFSClient->PATH;	if (dfsutil == nil) {		dfsutil = load DFSUtil DFSUtil->PATH;		dfsutil->init();	}	if (sys == nil)		sys == load Sys Sys->PATH;	sort = load Sort Sort->PATH; 	dfsclient->init();	chunks := file.getChunks();	cmp := ref DFSChunkCmp();	sort->sort(cmp, chunks);	i := 0; 	while (i < len chunks && chunks[i].offset + big chunks[i].size < offset)		i++;	while (i < len chunks && chunks[i].offset < offset + size) {		tempfd := dfsclient->readChunk(chunks[i]);		start := big 0;		buf := array [Sys->ATOMICIO] of byte; 		if (chunks[i].offset < offset)			start = sys->seek(tempfd, offset - chunks[i].offset, Sys->SEEKSTART); 		if (chunks[i].offset + big chunks[i].size > offset + size) {			total := int (offset + size - chunks[i].offset - start);				while (total > len buf) {				sys->read(tempfd, buf, len buf);				sys->write(fd, buf, len buf);				total -= len buf;			}			if (total > 0) {				sys->read(tempfd, buf[:total], total);				sys->write(fd, buf[:total], total);			}		}		else while ((length := sys->read(tempfd, buf, len buf)) != 0)			sys->write(fd, buf, length);		sys->remove(sys->fd2path(tempfd));		i++;	}}