##########################################	The implemention of DFSClient.#	It creates some connections with DFSServer and DFSNodeServer,#	and communicate information and data with them.#	#	@author Kai Yao(yaokai)#	@author Yang Fan(fyabc) ######################################### implement DFSClient; include "sys.m";include "dfsutil.m";include "dfsclient.m"; include "../xml/xmlhandle.m";include "xml.m";include "bufio.m";include "lists.m"; sys : Sys;dfsutil : DFSUtil; xml : Xml;xmlhd : XmlHandle;bufio : Bufio;lists : Lists; DFSFile : import dfsutil;DFSChunk : import dfsutil;DFSNode : import dfsutil;Connection : import sys;Parser : import xml;FD : import sys;Iobuf : import bufio; masterAddr : string;masterPort : int;defaultRep : int;defaultChunkSize : int;defaultTempPath : string; conn : Connection;init() { 	if (sys == nil)		sys = load Sys Sys->PATH;	if (dfsutil == nil)		dfsutil = load DFSUtil DFSUtil->PATH;	if (xml == nil)		xml = load Xml Xml->PATH;	if (xmlhd == nil)		xmlhd = load XmlHandle XmlHandle->PATH;	if (lists == nil)		lists = load Lists Lists->PATH;	if (bufio == nil)		bufio = load Bufio Bufio->PATH; 	xml->init();	dfsutil->init(); 	#masterAddr = "192.168.43.216";	#masterPort = 2333; 	buffer := bufio->open("/appl/MR/config", Bufio->OREAD);	masterAddr = buffer.gets('\n');	masterAddr = masterAddr[:len masterAddr - 1];	masterPort = 2333; 	defaultRep = 3;	defaultChunkSize = 1<<20;	defaultTempPath = "/appl/MR/cli/"; 	ok : int;	(ok, conn) = sys->dial("tcp!" + masterAddr + "!" + string masterPort, nil);	if (ok < 0) 		sys->print("Error: DFSClient--dial failed!\n");} disconnect(){	sys->fprint(conn.dfd, "disconnect");} createFile(fileName : string, replicas : int) : int {	buf := array [Sys->ATOMICIO] of byte;	msg := "create@" + fileName + "@" + string replicas; 	sys->fprint(conn.dfd, "%s", msg);	length := sys->read(conn.dfd, buf, len buf);	if (length <= 0) {		sys->print("Error: DFSClient--createFile failed!\n");			return 1;	}	sys->print("%s", string buf[:length]);	return 0;} createChunk(fileName : string, offset : big, size : int) : int{	buf := array [Sys->ATOMICIO] of byte;	msg := "chunk@" + fileName + "@" + string offset + "@" + string size;	sys->fprint(conn.dfd, "%s", msg);	length := sys->read(conn.dfd, buf, len buf);	ok := int string buf[:length];	if (ok != 0) {		sys->print("Error: DFSClient--createChunk failed: %d !,offset: %bd, size: %d\n", ok, offset, size);		return 1;	}	return 0;} listFiles() : string{	buf := array [Sys->ATOMICIO] of byte;	msg := "list";	sys->fprint(conn.dfd, "%s", msg);	length := sys->read(conn.dfd, buf, len buf);	if (length <= 0)		return nil;	ret := string buf[:length];	return ret;} getFile(fileName : string) : ref DFSFile{	buf := array [Sys->ATOMICIO] of byte;	msg := "get@" + fileName;	sys->fprint(conn.dfd, "%s", msg); 	length := sys->read(conn.dfd, buf, len buf);	if (length <= 0) {		sys->print("Error: DFSClient--deleteFile failed!\n");			return nil;	}	sys->print("%s", string buf[:length]);	if (int buf[0] == 'N')		return nil; 	sys->fprint(conn.dfd, "start");	sys->export(conn.dfd, defaultTempPath, Sys->EXPWAIT);		parser := xmlhd->init(defaultTempPath + fileName + ".xml");	file := xmlhd->xml2file(parser);	sys->remove(defaultTempPath + fileName + ".xml");	return file;} deleteFile(fileName : string) : int{	buf := array [Sys->ATOMICIO] of byte;	msg := "delete@" + fileName;	sys->fprint(conn.dfd, "%s", msg);		length := sys->read(conn.dfd, buf, len buf);	if (length <= 0) {		sys->print("Error: DFSClient--deleteFile failed!\n");			return 1;	}	sys->print("%s", string buf[:length]);	if (int buf[0] == 'N')		return 1; 	return 0;} updateNode(addr : string, port : int, chunkNumber : int) : int{	buf := array [Sys->ATOMICIO] of byte;	msg := "upNode@" + addr + "@" + string port + "@" +  string chunkNumber; 	sys->fprint(conn.dfd, "%s", msg);	length := sys->read(conn.dfd, buf, len buf);	if (length <= 0) {		sys->print("Error: DFSClient--updateNode failed!\n");			return 1;	}	sys->print("%s", string buf[:length]);	return 0;} removeNode(addr : string, port : int) : int{	buf := array [Sys->ATOMICIO] of byte;	msg := "rmNode@" + addr + "@" + string port; 	sys->fprint(conn.dfd, "%s", msg); 	length := sys->read(conn.dfd, buf, len buf);	if (length <= 0) {		sys->print("Error: DFSClient--removeNode failed!\n");			return 1;	}	sys->print("%s", string buf[:length]);	if (int buf[0] == 'N')		return 1; 	return 0;} # fyabc readChunk(chunk : ref DFSChunk) : ref FD{	node : ref DFSNode; 	for(p := chunk.nodes; p != nil; p = tl p)	{		node = hd p;		if((ret := readChunkWithNode(node, chunk.id)) != nil)			return ret;	}		sys->print("All nodes of this chunk %d are failed to read!", chunk.id);	return nil;} writeChunk(chunk : ref DFSChunk, offset : big, size : int, datafd : ref FD) : int{	ret := 0;	node : ref DFSNode; 	for(p := chunk.nodes; p != nil; p = tl p)	{		sys->seek(datafd, offset, Sys->SEEKSTART);		node = hd p;		if(writeChunkWithNode(node, chunk.id, size, datafd) != 0)		{			sys->print("Write chunk %d in node addr: %s, port: %d failed!\n", chunk.id, node.addr, node.port);			ret = -1;		}		else			sys->print("Successed write chunk %d in node addr: %s, port: %d!\n", chunk.id, node.addr, node.port);	}        return ret;} deleteChunk(chunk : ref DFSChunk) : int{	ret := 0;	node : ref DFSNode;	for(p := chunk.nodes; p != nil; p = tl p)	{		node = hd p;		if(deleteChunkWithNode(node, chunk.id)!=0)		{			sys->print("Delete chunk %d in node addr: %s, port: %d failed!\n", chunk.id, node.addr, node.port);			ret = -1;		}		else		sys->print("Successed delete chunk %d in node addr: %s, port: %d!\n", chunk.id, node.addr, node.port);    }        return ret;} lineOffset(fileName : string) : array of big{	file := getFile(fileName);	offsets : list of big; 	chunk : ref DFSChunk;	for(p := file.chunks; p != nil; p = tl p)	{		chunk = hd p;		chunkOffsets := lineOffsetChunk(chunk); 		for(i := 0; i < len chunkOffsets; i++)		{			offsets = chunkOffsets[i] :: offsets;		}	} 	length := len offsets;	ret := array[length] of big;	for(j := length-1; j>=0; j--)	{		ret[j] = hd offsets;		offsets = tl offsets;	}	return ret;} lineOffsetChunk(chunk : ref DFSChunk) : array of big{	node : ref DFSNode; 	for(p := chunk.nodes; p != nil; p = tl p)	{		node = hd p;		(ok, conn) := sys->dial("tcp!" + node.addr + "!" + string node.port, nil);		if (ok >= 0)		{			msg := "lineos@" + string chunk.id;			sys->fprint(conn.dfd, "%s", msg); 			buf := array [Sys->ATOMICIO] of byte;			length := sys->read(conn.dfd, buf, len buf);			(nil,chunkOffsets) := sys->tokenize(string buf[:length], " ") ;			sys->print("%d\n",len chunkOffsets);			ret := array[len chunkOffsets] of big; 			for(i :=0; i< len ret; i++)			{				ret[i] = big (hd chunkOffsets) + chunk.offset;				chunkOffsets = tl chunkOffsets;			} 			return ret;		}	} 	sys->print("All nodes of this chunk %d are failed to lineOffset!", chunk.id);	return nil;} readChunkWithNode(node : ref DFSNode, chunkId : int) : ref FD{	ok : int;	(ok, conn) = sys->dial("tcp!" + node.addr + "!" + string node.port, nil);	if (ok < 0)	{		sys->print("Error: DFSClient->readChunk--dial failed!\n");		return nil;	} 	buf := array [Sys->ATOMICIO] of byte;	msg := "read@" + string chunkId;	sys->fprint(conn.dfd, "%s", msg); 	sys->export(conn.dfd, defaultTempPath, Sys->EXPWAIT); 	ret := sys->open(defaultTempPath + string chunkId, sys->ORDWR); 	return ret;} writeChunkWithNode(node : ref DFSNode, chunkId : int, size : int, datafd : ref FD) : int{	ok : int;	(ok, conn) = sys->dial("tcp!" + node.addr + "!" + string node.port, nil);	if (ok < 0)	{		sys->print("Error: DFSClient->writeChunk--dial failed!\n");		return -1;	} 	buf := array [Sys->ATOMICIO] of byte;	msg := "write" ;	sys->fprint(conn.dfd, "%s", msg); 	sendWriteData(datafd, chunkId, size, conn); 	return 0; } deleteChunkWithNode(node : ref DFSNode, chunkId : int) : int{	ok : int;	(ok, conn) = sys->dial("tcp!" + node.addr + "!" + string node.port, nil);	if (ok < 0)	{		sys->print("Error: DFSClient->writeChunk--dial failed!\n");		return -1;	}		buf := array [Sys->ATOMICIO] of byte;	msg := "delete@" + string chunkId;	sys->fprint(conn.dfd, "%s", msg); 	length := sys->read(conn.dfd, buf, len buf);	return int string buf[:length];	} sendWriteData(datafd : ref FD, chunkId : int, size : int, conn : Connection){	stmsg := array[100] of byte;	n := sys->read(conn.dfd, stmsg, len stmsg);	sys->mount(conn.dfd, nil, defaultTempPath + "remote",Sys->MCREATE, nil);	datacopyfd := sys->create(defaultTempPath + "remote/" + string chunkId, sys->ORDWR, 8r600); 	buf := array [Sys->ATOMICIO] of byte;	length : int;	(nil, dir) := sys->fstat(datafd);	total := dir.length;	offset := sys->seek(datafd, big 0, Sys->SEEKRELA);	if (int(total - offset) < size)		size = int(total - offset);	do {		if (size < len buf) 			length = sys->read(datafd, buf, size);		else			length = sys->read(datafd, buf, len buf);		sys->write(datacopyfd, buf[:length], length);		size -= length; 	}while (size > 0); 	sys->unmount(nil, defaultTempPath + "remote");}