shithub: dddb

Info  •  Files  •  Log  •  Branches

added a basic node registry

--- a/appl/cmd/config.b

+++ b/appl/cmd/config.b

@@ -7,6 +7,7 @@

 DCFGPATH:	con "/lib/ndb/dddbcfg";

 DADDR:		con "tcp!*!dddbctl";

+DPSIZE:		con 3;

 DFSWRKS:	con 10;

 # Attrdb constants 

@@ -13,7 +14,9 @@

 KNAME:		con "nodename";

 KSYSNAME:	con "nodesysn";

 KADDR:		con "addr";

+KKEYFILE:	con "keyfile";

 KSTORAGE:	con "storage";

+KPSIZE:		con "psize";

 KFSWRKS:	con "readworkers";

 Config.open(nodename: string, path: string): Config

@@ -95,8 +98,11 @@

 	sysname := entry.findfirst(KSYSNAME);

 	addr := entry.findfirst(KADDR);

 	storage := entry.findfirst(KSTORAGE);

+	keyfile := entry.findfirst(KKEYFILE);

+	psize_s := entry.findfirst(KPSIZE);

 	fswrks_s := entry.findfirst(KFSWRKS);

+	psize := DPSIZE;

 	fswrks := DFSWRKS;

 	if(len name == 0)

@@ -111,8 +117,14 @@

 			error(sys->sprint("malformed fs workers count: %s", fswrks_s));

 		fswrks = fswrks_i;

 	}

+	if(len psize_s != 0) {

+		(psize_i, rm) := strm->toint(psize_s, 10);

+		if(rm != "")

+			error(sys->sprint("malformed pool size: %s", psize_s));

+		psize = psize_i;

+	}

 	return NodeConfig(

-		name, sysname, addr, storage,	# basic information

-		fswrks);						# tunable options

+		name, sysname, addr, keyfile, storage,	# basic information

+		psize, fswrks);							# tunable options

 }

--- a/appl/cmd/ctlfs.b

+++ b/appl/cmd/ctlfs.b

@@ -4,6 +4,9 @@

 include "security.m";

 	auth: Auth;

+include "keyring.m";

+	keyring: Keyring;

+

 include "styx.m";

 	styx: Styx;

 	Tmsg, Rmsg: import Styx;

@@ -10,100 +13,115 @@

 include "styxservers.m";

 	styxservers: Styxservers;

-	Styxserver, Fid, Navigator,

-	Navop, Enotfound, Enotdir: import styxservers;

+	nametree: Nametree;

+	Tree: import nametree;

+	Styxserver, Fid, Navigator, Navop,

+	Eperm, Ecount, Eoffset: import styxservers;

-# FS file index

-Qroot, Qctl, Qstats, Qmax: con iota;

-tab := array[] of {

-	(Qroot, ".", Sys->DMDIR|8r555),

-	(Qctl, "ctl", 8r222),

-	(Qstats, "stats", 8r111),

-};

+# Database features

+dbfeatures: list of string;

+# Initial fs files

+Qroot, Qctl, Qname, Qstatus: con big iota;

+

 # create ctlfs and the appropriate listeners

-init_ctlfs(cfg: Config, keyfile: string, algs: list of string)

+run_ctlfs(cfg: Config, dbreg: ref DbRegistry, keyfile: string, algs: list of string)

 {

-	dial = load Dial Dial->PATH;

-	auth = load Auth Auth->PATH;

+	sys->fprint(stderr, "setting up ctlfs\n");

+

+	dbfeatures = DBVER :: dbfeatures;

+

 	styx = load Styx Styx->PATH;

+	styxservers = load Styxservers Styxservers->PATH;

+	nametree = load Nametree Nametree->PATH;

-	if(dial == nil)

-		error("ctlfs: dial module not found");

-	if(auth == nil)

-		error("ctlfs: auth module not found");

+	if(debug)

+		sys->fprint(stderr, "ctlfs: checking if modules are loaded\n");

+

 	if(styx == nil)

 		error("ctlfs: styx module not found");

+	if(styxservers == nil)

+		error("ctlfs: styxservers module not found");

+	if(nametree == nil)

+		error("ctlfs: nametree module not found");

+	if(debug)

+		sys->fprint(stderr, "ctlfs: initializing modules\n");

+

 	auth->init();

 	styx->init();

 	styxservers->init(styx);

-	styxservers->traceset(chatty);

+	nametree->init();

+

 	# authinfo init

-	if(debug)

-		sys->fprint(stderr, "ctlfs: reading authinfo");

+

 	authinfo: ref Keyring->Authinfo;

-	if (doauth) {

-		if (keyfile == nil)

-			keyfile = "/usr/" + user() + "/keyring/default";

-		authinfo = keyring->readauthinfo(keyfile);

-		if (authinfo == nil)

-			error(sys->sprint("ctlfs: cannot read %s: %r", keyfile));

-	}

+	if (keyfile == nil)

+		keyfile = "/usr/" + user() + "/keyring/default";

+	if(debug)

+		sys->fprint(stderr, "ctlfs: reading authinfo %s\n", keyfile);

+	authinfo = keyring->readauthinfo(keyfile);

+	if (authinfo == nil)

+		error(sys->sprint("ctlfs: cannot read %s: %r", keyfile));

 	# announcing

 	if(debug)

-		sys->fprint(stderr, "ctlfs: announcing dddbctl");

-	addr := dial->netmkaddr(cfg.addr, "tcp", "dddbctl");

-	c := dial->announce(addr);

+		sys->fprint(stderr, "ctlfs: announcing dddbctl\n");

+	# addr := dial->netmkaddr(cfg.addr, "tcp", "dddbctl");

+	c := dial->announce(cfg.addr);

 	if(c == nil)

-		error(sys->sprint("ctlfs: cannot listen on %s\n", addr));

+		error(sys->sprint("ctlfs: cannot listen on %s\n", cfg.addr));

 	# bootstrapping

 	if(debug)

-		sys->fprint

+		sys->fprint(stderr, "ctlfs: bootstrapping\n");

 	sys->unmount(nil, "/mnt/keys");

+	sys->unmount(nil, "/mnt");

-	navch := chan of ref Navop;

-	spawn ctlfs_navigator(navch);

+	# nametree; this is shared across all attachees

+	(tree, treeop) := nametree->start();

+	tree.create(Qroot, dir(".", 8r555|Sys->DMDIR, Qroot));

+	tree.create(Qroot, dir("ctl", 8r666, Qctl));

+	tree.create(Qroot, dir("status", 8r444, Qstatus));

-	nav := Navigator.new(navch);

-	(tc, srv) := Styxserver.new(fildes(0), nav, big Qroot);

+	sys->fprint(stderr, "ctlfs: finished setting up; starting\n");

 	# listener entrypoint

-	listener(c, authinfo, algs);

+	ctlfs_listener(cfg, dbreg, c, treeop, authinfo, algs);

+	tree.quit();

 }

 # dddbctl listener loop

-ctlfs_listener(c: ref Dial->Connection, authinfo: ref Keyring->Authinfo, algs: list of string)

+ctlfs_listener(cfg: Config, dbreg: ref DbRegistry, c: ref Dial->Connection, treeop: chan of ref Styxservers->Navop, authinfo: ref Keyring->Authinfo, algs: list of string)

 {

-	for (;;) {

+	loop: for (;;) {

 		nc := dial->listen(c);

 		if (nc == nil)

 			error(sys->sprint("listen failed: %r"));

 		if (debug)

-			sys->fprint(stderr, "ctlfs: got connection from %s\n",

+			sys->fprint(stderr, "ctlfs: got connection from %s",

 						readfile(nc.dir + "/remote"));

 		dfd := dial->accept(nc);

-		if (dfd != nil) {

-			if(nc.cfd != nil)

-				sys->fprint(nc.cfd, "keepalive");

-			hostname: string;

-			if(passhostnames){

-				hostname = readfile(nc.dir + "/remote");

-				if(hostname != nil)

-					hostname = hostname[0:len hostname - 1];

-			}

+		if (dfd == nil)

+			continue loop;

-			spawn ctlfs_authenticator(dfd, authinfo, algs, hostname);

-		}

+		if(nc.cfd != nil)

+			sys->fprint(nc.cfd, "keepalive");

+

+		hostname: string;

+		hostname = readfile(nc.dir + "/remote");

+		if(hostname != nil)

+			hostname = hostname[0:len hostname - 1];

+

+		regchan := dbreg.changen();

+		spawn ctlfs_authenticator(cfg, regchan, dfd, treeop, authinfo, algs, hostname);

 	}

 }

 # authenticate a connection and set the user id.

-ctlfs_authenticator(dfd: ref Sys->FD, authinfo: ref Keyring->Authinfo,

+ctlfs_authenticator(cfg: Config, regchan: (chan of ref RegTMsg, chan of ref RegRMsg), dfd: ref Sys->FD, treeop: chan of ref Styxservers->Navop, authinfo: ref Keyring->Authinfo,

 		algs: list of string, hostname: string)

 {

 	# authenticate and change user id appropriately

@@ -110,113 +128,160 @@

 	(fd, err) := auth->server(algs, authinfo, dfd, 1);

 	if (fd == nil) {

 		if (debug)

-			sys->fprint(stderr(), "ctlfs: authentication failed: %s\n", err);

+			sys->fprint(stderr, "ctlfs: authentication failed: %s\n", err);

 		return;

 	}

 	if (debug)

-		sys->fprint(stderr(), "ctlfs: client authenticated as %s\n", err);

+		sys->fprint(stderr, "ctlfs: client authenticated as %s\n", err);

-	spawn exportproc(sync, mfd, err, hostname, fd);

+	spawn ctlfs_loop(cfg, regchan, fd, treeop, hostname);

 }

-ctlfs_loop()

+# filesystem loop; nb: hostname will be later used for stats

+ctlfs_loop(cfg: Config, regchan: (chan of ref RegTMsg, chan of ref RegRMsg), fd: ref Sys->FD, treeop: chan of ref Styxservers->Navop, nil: string)

 {

+	(tc, srv) := Styxserver.new(fd, Navigator.new(treeop), big Qroot);

+

+	# registry rx/tx

+	(tx, rx) := regchan;

+

 	# Primary server loop

 	loop:

-	while((tmsg := <-tc) != nil) {

+	while((tm := <-tc) != nil) {

 		# Switch on operations being performed on a given Fid

-		pick msg := tmsg {

+		pick t := tm {

+		# Open operation

 		Open =>

-			srv.default(msg);

-		Read =>

-			fid := srv.getfid(msg.fid);

-

-			if(fid.qtype & Sys->QTDIR) {

-				# This is a directory read

-				srv.default(msg);

+			(f, mode, d, err) := srv.canopen(t);

+			if(f == nil){

+				srv.reply(ref Rmsg.Error(t.tag, err));

 				continue loop;

 			}

+			f.open(mode, d.qid);

-			case int fid.path {

-			Qlog =>

-				# A read on our log file, tell them what they've already said ?

-				s := "";

+			case f.path {

-				for(l := log; l != nil; l = tl l)

-					s = hd l + s;

+			# Qroot

+			Qroot =>

+				if(t.mode != Sys->OREAD) {

+					srv.reply(ref Rmsg.Error(t.tag, Eperm));

+					continue loop;

+				}

+				srv.reply(ref Rmsg.Open(t.tag, d.qid, srv.iounit()));

-				srv.reply(styxservers->readstr(msg, s));

+			# Qctl

+			Qctl =>

+				if((t.mode & (Sys->OTRUNC | Sys->ORCLOSE | Sys->OEXCL)) != 0) {

+					srv.reply(ref Rmsg.Error(t.tag, Eperm));

+					continue loop;

+				}

+				srv.reply(ref Rmsg.Open(t.tag, d.qid, srv.iounit()));

+			# Qname

+			Qname =>

+				if((t.mode & (Sys->OTRUNC | Sys->ORCLOSE | Sys->OEXCL)) != 0) {

+					srv.reply(ref Rmsg.Error(t.tag, Eperm));

+					continue loop;

+				}

+				srv.reply(ref Rmsg.Open(t.tag, d.qid, srv.iounit()));

-			* =>

-				srv.default(msg);

+			# Qstatus

+			Qstatus => 

+				if(t.mode != Sys->OREAD) {

+					srv.reply(ref Rmsg.Error(t.tag, Eperm));

+					continue loop;

+				}

+				srv.reply(ref Rmsg.Open(t.tag, d.qid, srv.iounit()));

+

+			# Default reply

+			* => srv.default(t);

 			}

-		Write =>

-			fid := srv.getfid(msg.fid);

+		# Read operation

+		Read =>

+			(f, err) := srv.canread(t);

+			if(f == nil) {

+				srv.reply(ref Rmsg.Error(t.tag, err));

+				break;

+			}

+			if(f.qtype & Sys->QTDIR){

+				srv.read(t);

+				continue loop;

+			}

-			case int fid.path {

+			case f.path {

+

+			# Qctl

 			Qctl =>

-				# Don't care about offset

-				cmd := string msg.data;

+				ctlmsg := joinstr(dbfeatures, "\n") + "\n";

+				ctlmsgbuf := array of byte ctlmsg;

+				rend := int t.offset + t.count;

+				if(rend > len ctlmsg)

+					rend = len ctlmsg;

+				srv.reply(ref Rmsg.Read(t.tag, ctlmsgbuf[(int t.offset):rend]));

-				reply: ref Rmsg = ref Rmsg.Write(msg.tag, len msg.data);

+			# Qname

+			Qname =>

+				namemsg := cfg.name + "\n";

+				namemsgbuf := array of byte namemsg;

+				rend := int t.offset + t.count;

+				if(rend > len namemsg)

+					rend = len namemsg;

+				srv.reply(ref Rmsg.Read(t.tag, namemsgbuf[(int t.offset):rend]));

+			# Qstatus

+			Qstatus =>

+				info: list of string;

+				info = "name		" + cfg.name :: info;

+				info = "sysname		" + cfg.sysn :: info;

+				info = "addr		" + cfg.addr :: info;

+				info = "storage		" + cfg.storage :: info;

+				info = "fsworkers	" + sys->sprint("%d", cfg.fswrks) :: info;

+				info = "" :: info;

+				info = "nodes" :: info;

-				case cmd {

-				* =>

-					# Ignore empty writes

-					if(cmd != nil)

-						log = cmd :: log;

-					else

-						reply = ref Rmsg.Error(msg.tag, "empty write!");

-				}

-				srv.reply(reply);

-				

-			* =>

-				srv.default(msg);

-			}

+				tx <-= ref RegTMsg.GetNodes();

+				reply := <- rx;

-		* =>

-			srv.default(msg);

-		}

-	}

+				pick r := reply {

+					Error => srv.reply(ref Rmsg.Error(t.tag, r.err));

+					NodeList =>

+						names := lists->reverse(r.names);

+						while(len names != 0) {

+							node := hd names;

+							sline := "";

-	exit;

-}

+							tx <-= ref RegTMsg.Check(node);

+							crep := <- rx;

+							pick cr := crep {

+								Error => sline = cr.err;

+								Status =>

+									up := cr.count;

+									ps := cr.poolsize;

+									sline = sys->sprint("%d	%d", up, ps);

+								* => sline = "unsupported message";

+							}

-# Navigator function for moving around under /

-ctlfs_navigator(c: chan of ref Navop) {

-	loop: 

-	for(;;) {

-		navop := <-c;

-		pick op := navop {

-		Stat =>

-			op.reply <-= (dir(int op.path), nil);

-			

-		Walk =>

-			if(op.name == "..") {

-				op.reply <-= (dir(Qroot), nil);

-				continue loop;

-			}

+							info = node + "		" +  sline :: info;

+							names = tl names;

+						}

+					* => srv.reply(ref Rmsg.Error(t.tag, "unsupported version"));

+				}

-			case int op.path&16rff {

+				statusmsg := joinstr(lists->reverse(info), "\n") + "\n";

+				statusmsgbuf := array of byte statusmsg;

+				rend := int t.offset + t.count;

+				if(rend > len statusmsg)

+					rend = len statusmsg;

+				srv.reply(ref Rmsg.Read(t.tag, statusmsgbuf[(int t.offset):rend]));

-			Qroot =>

-				for(i := 1; i < Qmax; i++)

-					if(tab[i].t1 == op.name) {

-						op.reply <-= (dir(i), nil);

-						continue loop;

-					}

-

-				op.reply <-= (nil, Enotfound);

-			* =>

-				op.reply <-= (nil, Enotdir);

+			# Default reply

+			* => srv.default(t);

 			}

-			

-		Readdir =>

-			for(i := 0; i < op.count && i + op.offset < (len tab) - 1; i++)

-				op.reply <-= (dir(Qroot+1+i+op.offset), nil);

+		# Write operation

+		Write =>

+			srv.default(t);

-			op.reply <-= (nil, nil);

+		# Default action

+		* => srv.default(t);

 		}

 	}

 }

--- a/appl/cmd/dddb.b

+++ b/appl/cmd/dddb.b

@@ -6,13 +6,18 @@

 include "draw.m";

 include "string.m";

 	strm: String;

+include "lists.m";

+	lists: Lists;

 include "config.b";

 include "ctlfs.b";

+include "nodereg.b";

 stderr: ref Sys->FD;

 debug: int;

+DBVER: con "v0.1.0";

+

 error(s: string)

 {

 	sys->fprint(stderr, "dddb: %s\n", s);

@@ -19,10 +24,13 @@

 	raise "dddb:error";

 }

+Epoolnotfound	: con "pool not found";

+Epoolavail		: con "pool not available";

+

 Dddb: module {

 	init: fn(nil: ref Draw->Context, args: list of string);

-	run_fs: fn(cfg: Config);

+	# Configuration section

 	Config: adt {

 		name:		string;

 		sysn:		string;

@@ -31,7 +39,7 @@

 		fswrks:		int;

 		nodes:		list of NodeConfig;

-		open:		fn(nodename: string, path: string): Config;

+		open:		fn(nodename: string, mtpt: string): Config;

 	};

 	NodeConfig: adt {

@@ -38,12 +46,58 @@

 		name:		string;

 		sysn:		string;

 		addr:		string;

+		keyfile:	string;

 		storage:	string;

+		psize:		int;

 		fswrks:		int;

 		new:		fn(entry: ref Dbentry): NodeConfig;

 	};

+	# Registry section

+	RegTMsg: adt {

+		pick {

+		GetNodes =>

+		ChanClose =>

+		Check or Refresh or Close =>

+			nodename:	string;

+		}

+	};

+

+	RegRMsg: adt {

+		pick {

+		Error =>

+			err:		string;

+		Status =>

+			count:		int;

+			poolsize:	int;

+		NodeList =>

+			names:		list of string;

+		}

+	};

+

+	NodePool: adt {

+		cfg:		NodeConfig;

+		instances:	list of string;

+

+		init:		fn(r: self ref NodePool): int;

+		check:		fn(r: self ref NodePool): int;

+		refresh:	fn(r: self ref NodePool): int;

+		newinst:	fn(r: self ref NodePool, mtpt: string): string;

+		close:		fn(r: self ref NodePool);

+	};

+

+	DbRegistry: adt {

+		nodepools:	list of ref NodePool;

+		# rchans:		list of chan of ref RegRMsg;

+		# tchans:		list of chan of ref RegTMsg;

+

+		new:		fn(cfgs: list of NodeConfig): ref DbRegistry;

+		init:		fn(r: self ref DbRegistry);

+		# run:		fn(r: self ref DbRegistry);

+		changen:	fn(r: self ref DbRegistry): (chan of ref RegTMsg, chan of ref RegRMsg);

+		close:		fn(r: self ref DbRegistry);

+	};

 };

 init(nil: ref Draw->Context, args: list of string)

@@ -51,19 +105,43 @@

 	sys = load Sys Sys->PATH;

 	arg := load Arg Arg->PATH;

 	strm = load String String->PATH;

+	lists = load Lists Lists->PATH;

+	dial = load Dial Dial->PATH;

+	auth = load Auth Auth->PATH;

+	keyring = load Keyring Keyring->PATH;

+	if(sys == nil)

+		error("dddb: sys module not found");

+	if(arg == nil)

+		error("dddb: arg module not found");

+	if(strm == nil)

+		error("dddb: strm module not found");

+	if(lists == nil)

+		error("dddb: lists module not found");

+	if(dial == nil)

+		error("dddb: dial module not found");

+	if(auth == nil)

+		error("dddb: auth module not found");

+	if(keyring == nil)

+		error("dddb: keyring module not found");

+

 	stderr = sys->fildes(2);

 	cfgpath: string = "";

+	keyfile: string = nil;

+	algs: list of string = nil;

 	arg->init(args);

-	arg->setusage(arg->progname()+ " [-d] [-c config] nodename");

+	arg->setusage(arg->progname()+ " [-d] [-k keyfile] [-C algs] [-c config] nodename");

 	while((c := arg->opt()) != 0)

 		case c {

 		'd' => debug++;

-		'c' =>

-			cfgpath = arg->earg();

+		'c' => cfgpath = arg->earg();

+		'k' => keyfile = arg->earg();

+		'C' =>

+			algsstr := arg->earg();

+			(nil, algs) = sys->tokenize(algsstr, ",");

 		* =>

-			sys->fprint(sys->fildes(2), "bad option: -%c\n", c);

+			sys->fprint(stderr, "bad option: -%c\n", c);

 			arg->usage();

 		}

@@ -88,7 +166,86 @@

 		sys->fprint(stderr, "cfg.fswrks: %d\n", cfg.fswrks);

 	}

-	run_fs(cfg);

+	if(debug)

+		sys->fprint(stderr, "dddb: creating and running node registry\n");

+

+	sys->pctl(Sys->NEWPGRP, nil);

+

+	dbreg := DbRegistry.new(cfg.nodes);

+	spawn dbreg.init();

+	# spawn dbreg.run();

+

+	if(debug)

+		sys->fprint(stderr, "dddb: running ctlfs\n");

+

+	run_ctlfs(cfg, dbreg, keyfile, algs);

+

+	sys->fprint(stderr, "dddb: performing shutdown\n");

+	dbreg.close();

+	sys->fprint(stderr, "dddb: all components shut off\n");

 }

+user(): string

+{

+	user := readfile("#c/user");

+	if(user == nil)

+		return "none";

+	return user;

+}

+

+readfile(file: string): string

+{

+	fd := sys->open(file, Sys->OREAD);

+	if(fd == nil)

+		return nil;

+

+	buf := array[1024] of byte;

+	n := sys->read(fd, buf, len buf);

+	if(n < 0)

+		return nil;

+

+	return string buf[0:n];

+}

+

+writefile(file: string, s: string): int

+{

+	fd := sys->open(file, Sys->OWRITE);

+	if(fd == nil)

+		return -1;

+

+	buf := array of byte s;

+	n := sys->write(fd, buf, len buf);

+

+	return n;

+}

+

+dir(name: string, perm: int, qid: big): Sys->Dir

+{

+	d := sys->zerodir;

+	user := user();

+	d.name = name;

+	d.uid = user;

+	d.gid = user;

+	d.qid.path = qid;

+	if (perm & Sys->DMDIR)

+		d.qid.qtype = Sys->QTDIR;

+	else

+		d.qid.qtype = Sys->QTFILE;

+	d.mode = perm;

+	return d;

+}

+

+joinstr(items: list of string, sep: string): string

+{

+	s := "";

+	citem := hd items;

+	items = tl items;

+	s = s + citem;

+	while(items != nil) {

+		citem = hd items;

+		items = tl items;

+		s = s + sep + citem;

+	}

+	return s;

+}