20060303a

1 files changed, 239 insertions, 0 deletions

diff --git a/appl/grid/register.b b/appl/grid/register.b
new file mode 100644
index 00000000..bd0d8265
--- /dev/null
+++ b/appl/grid/register.b
@@ -0,0 +1,239 @@
+implement Register;
+
+#
+# Copyright © 2003 Vita Nuova Holdings Limited.  All rights reserved.
+#
+
+
+include "sys.m";
+	sys: Sys;
+include "draw.m";
+include "sh.m";
+include "registries.m";
+	registries: Registries;
+	Registry, Attributes, Service: import registries;
+include "grid/announce.m";
+	announce: Announce;
+include "arg.m";
+
+registered: ref Registries->Registered;
+
+Register: module {
+	init: fn(ctxt: ref Draw->Context, argv: list of string);
+};
+
+init(ctxt: ref Draw->Context, argv: list of string)
+{
+	sys = load Sys Sys->PATH;
+	sys->pctl(sys->FORKNS | sys->NEWPGRP, nil);
+	registries = load Registries Registries->PATH;
+	if (registries == nil)
+		badmod(Registries->PATH);
+	registries->init();
+	announce = load Announce Announce->PATH;
+	if (announce == nil)
+		badmod(Announce->PATH);
+	announce->init();
+	arg := load Arg Arg->PATH;
+	if (arg == nil)
+		badmod(Arg->PATH);
+
+	attrs := Attributes.new(("proto", "styx") :: ("auth", "none") :: ("resource","Cpu Pool") :: nil);
+	maxusers := -1;
+	autoexit := 0;
+	myaddr := "";
+	arg->init(argv);
+	arg->setusage("register [-u maxusers] [-e exit threshold] [-a attributes] { program }");
+	while ((opt := arg->opt()) != 0) {
+		case opt {
+		'm' =>
+			attrs.set("memory", memory());
+		'u' =>
+			if ((maxusers = int arg->earg()) <= 0)
+				arg->usage();
+		'e' =>
+			if ((autoexit = int arg->earg()) < 0)
+				arg->usage();
+		'A' =>
+			myaddr = arg->earg();
+		'a' =>
+			attr := arg->earg();
+			val := arg->earg();
+			attrs.set(attr, val);
+		}
+	}
+	argv = arg->argv();
+	if (argv == nil)
+		arg->usage();
+	(nil, plist) := sys->tokenize(hd argv, "{} \t\n");
+	arg = nil;	
+	sysname := readfile("/dev/sysname");
+	reg: ref Registry;
+	reg = Registry.new("/mnt/registry");
+	if (reg == nil)
+		reg = Registry.connect(nil, nil, nil);
+	if (reg == nil)
+		error(sys->sprint("Could not find registry: %r\nMake sure that ndb/cs has been started and there is a registry announcing on the machine specified in /lib/ndb/local"));
+
+	c : sys->Connection;
+	if (myaddr == nil) {
+		(addr, conn) := announce->announce();
+		if (addr == nil)
+			error(sys->sprint("cannot announce: %r"));
+		myaddr = addr;
+		c = *conn;
+	}
+	else {
+		n: int;
+		(n, c) = sys->announce(myaddr);
+		if (n == -1)
+			error(sys->sprint("cannot announce: %r"));
+		(n, nil) = sys->tokenize(myaddr, "*");
+		if (n > 1) {
+			(nil, lst) := sys->tokenize(myaddr, "!");
+			if (len lst >= 3)
+				myaddr = "tcp!" + sysname +"!" + hd tl tl lst;
+		}
+	}
+	persist := 0;
+	if (attrs.get("name") == nil)
+		attrs.set("name", sysname);
+	err: string;
+	(registered, err) = reg.register(myaddr, attrs, persist);
+	if (err != nil) 
+		error("could not register with registry: "+err);
+
+	mountfd := popen(ctxt, plist);
+	spawn listener(c, mountfd, maxusers);
+}
+
+listener(c: Sys->Connection, mountfd: ref sys->FD, maxusers: int)
+{
+	for (;;) {
+		(n, nc) := sys->listen(c);
+		if (n == -1)
+			error(sys->sprint("listen failed: %r"));
+		dfd := sys->open(nc.dir + "/data", Sys->ORDWR);
+		if (maxusers != -1 && nusers >= maxusers)
+			sys->fprint(stderr(), "register: maxusers (%d) exceeded!\n", nusers);
+		else if (dfd != nil) {
+			sync := chan of int;
+			addr := readfile(nc.dir + "/remote");
+			if (addr == nil)
+				addr = "unknown";
+			if (addr[len addr - 1] == '\n')
+				addr = addr[:len addr - 1];
+			spawn proxy(sync, dfd, mountfd, addr);
+			<-sync;
+		}
+	}
+}
+
+proxy(sync: chan of int, dfd, mountfd: ref sys->FD, addr: string)
+{
+	pid := sys->pctl(Sys->NEWFD | Sys->NEWNS, 1 :: 2 :: mountfd.fd :: dfd.fd :: nil);
+	dfd = sys->fildes(dfd.fd);
+	mountfd = sys->fildes(mountfd.fd);
+	sync <-= 1;
+	done := chan of int;
+	spawn exportit(dfd, done);
+	if (sys->mount(mountfd, nil, "/", sys->MREPL | sys->MCREATE, addr) == -1)
+		sys->fprint(stderr(), "register: proxy mount failed: %r\n");
+	nusers++;
+	<-done;
+	nusers--;
+}
+
+nusers := 0;
+clock(tick: chan of int)
+{
+	for (;;) {
+		sys->sleep(2000);
+		tick <-= 1;
+	}
+}
+
+exportit(dfd: ref sys->FD, done: chan of int)
+{
+	sys->export(dfd, "/", sys->EXPWAIT);
+	done <-= 1;
+}
+
+popen(ctxt: ref Draw->Context, argv: list of string): ref Sys->FD
+{
+	sync := chan of int;
+	fds := array[2] of ref Sys->FD;
+	sys->pipe(fds);
+	spawn runcmd(ctxt, argv, fds[0], sync);
+	<-sync;
+	return fds[1];
+}
+
+runcmd(ctxt: ref Draw->Context, argv: list of string, stdin: ref Sys->FD, sync: chan of int)
+{
+	pid := sys->pctl(Sys->FORKFD, nil);
+	sys->dup(stdin.fd, 0);
+	stdin = nil;
+	sync <-= 0;
+	sh := load Sh Sh->PATH;
+	sh->run(ctxt, argv);
+}
+
+error(e: string)
+{
+	sys->fprint(stderr(), "register: %s\n", e);
+	raise "fail:error";
+}
+
+user(): string
+{
+	if ((s := readfile("/dev/user")) == nil)
+		return "none";
+	return s;
+}
+
+readfile(f: string): string
+{
+	fd := sys->open(f, sys->OREAD);
+	if(fd == nil)
+		return nil;
+
+	buf := array[8192] of byte;
+	n := sys->read(fd, buf, len buf);
+	if(n < 0)
+		return nil;
+
+	return string buf[0:n];	
+}
+
+stderr(): ref Sys->FD
+{
+	return sys->fildes(2);
+}
+
+badmod(path: string)
+{
+	sys->fprint(stderr(), "Register: cannot load %s: %r\n", path);
+	exit;
+}
+
+killg(pid: int)
+{
+	if ((fd := sys->open("/prog/" + string pid + "/ctl", Sys->OWRITE)) != nil) {
+		sys->fprint(fd, "killgrp");
+		fd = nil;
+	}
+}
+
+memory(): string
+{
+	buf := array[1024] of byte;
+	s := readfile("/dev/memory");
+	(nil, lst) := sys->tokenize(s, " \t\n");
+	if (len lst > 2) {
+		mem := int hd tl lst;
+		mem /= (1024*1024);
+		return string mem + "mb";
+	}
+	return "not known";
+}