########################################
#
#	The implemention of DFSNodeServer.
#	DFSNodeServer is the service interface that the DFS data node can provide to DFSClient.
#	Every data node in dfs use dfsnodeserver to transport file chunks and communicate with master.
#	DFSNodeServer announce a Connection,
#	listen the message from other computing equipment,
#	and do appropriate operation,such as read, write, delete or lineoffset the DFSChunk.
#
#	@author Yang Fan(fyabc) 
#
########################################

implement DFSNodeServer;

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

include "dfsutil.m";
include "dfsclient.m";


Connection : import sys;

DFSFile : import dfsutil;
DFSChunk : import dfsutil;
DFSNode : import dfsutil;
FD : import sys;
Iobuf : import bufio;

sys : Sys;
bufio : Bufio;
dfsutil : DFSUtil;
dfsclient : DFSClient;

dataPath := string "/appl/MR/ser/";
homePath := string "/appl/MR/";
localPort := int 2334;
chunkNumber := int 0;


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

init(ctxt : ref Draw->Context, args : list of string)
{
	sys = load Sys Sys->PATH;
	bufio = load Bufio Bufio->PATH;
	dfsutil = load DFSUtil DFSUtil->PATH;
	dfsclient = load DFSClient DFSClient->PATH;

	dfsutil->init();

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

	spawn heartBeat(localAddr);

	(n, conn) := sys->announce("tcp!*!" + string localPort);
	if (n < 0) {
		sys->print("DFSServer: announce failed %r\n");
		exit;
	}
	while(1) {
		listen(conn, ctxt);
	}
}

heartBeat(localAddr : string) {
	while (1) {
		dfsclient->init();
		dfsclient->updateNode(localAddr, localPort, chunkNumber); 
		dfsclient->disconnect();
		sys->sleep(30000);
	}
}

listen(conn : Connection, ctxt : ref Draw->Context) 
{
	buf := array [sys->ATOMICIO] of byte;
	(ok, c) := sys->listen(conn);
	if (ok < 0) {
		sys->print("DFSSlaveServer: listen failed %r\n");
	    exit;
	}
	rdf := sys->open(conn.dir + "/remote", sys->OREAD);
	n := sys->read(rdf, buf, len buf);
	connHandle(c, ctxt);
}

connHandle(conn : Connection, ctxt : ref Draw->Context)
{
	addr := array [sys->ATOMICIO] of byte;
	msgStr := array [sys->ATOMICIO] of byte;
	msg : list of string;
	
	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);
	sys->print("Message from: %s\n", string addr[:addrlen]);

	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)
		{
			"disconnect" => break receive;
			"read" => {
				chunkId := int hd msg;

				datafd := sys->open(dataPath + string chunkId, Sys->OREAD);

				if(datafd != nil)
					sendReadData(datafd, chunkId, conn);

				break receive;
			}
			
			"write" => {
				dfd := sys->open(conn.dir + "/data", sys->ORDWR);
				sys->fprint(dfd, "start");
				sys->export(dfd, dataPath, Sys->EXPWAIT);	
				chunkNumber++;
				break receive;
			}
			
			"delete" => {
				chunkId := hd msg;
				ok := sys->remove(dataPath + chunkId);	
				sys->fprint(wdfd, "%d", ok);	
				chunkNumber--;
			}
			
			"lineos" => {
			}
			
			"lineosc" =>{
			}
			* =>
				sys->fprint(wdfd, "Unknown message!\n");
		}
		msglen = sys->read(rdfd, msgStr, len msgStr);
	}

}

sendReadData(datafd : ref FD, chunkId : int, conn : Connection)
{
	dfd := sys->open(conn.dir + "/data", sys->ORDWR);
	sys->mount(dfd, nil, dataPath + "remote",Sys->MCREATE, nil);
	datacopyfd := sys->create(dataPath + "remote/" + string chunkId, sys->ORDWR, 8r600);

	buf := array [Sys->ATOMICIO] of byte;
	length : int;
	do {
		length = sys->read(datafd, buf, len buf);
		sys->write(datacopyfd, buf[:length], length);
	}while (length == len buf);

	sys->unmount(nil, dataPath + "remote");
}