ref: 1018d7a7091fd5e0433f501a3474cdc3fec322cc
dir: /appl/grid/cpupool.b/
implement CpuPool; # # Copyright © 2003 Vita Nuova Holdings Limited. All rights reserved. # include "sys.m"; sys : Sys; include "daytime.m"; daytime: Daytime; include "styx.m"; styx: Styx; Rmsg, Tmsg: import styx; include "styxservers.m"; styxservers: Styxservers; Fid, Navigator, Navop: import styxservers; Styxserver: import styxservers; nametree: Nametree; Tree: import nametree; include "draw.m"; include "dial.m"; dial: Dial; include "sh.m"; include "arg.m"; include "registries.m"; registries: Registries; Registry, Attributes, Service: import registries; include "grid/announce.m"; announce: Announce; include "readdir.m"; readdir: Readdir; TEST: con 0; RUN : con "#!/dis/sh\n" + "load std\n" + "if {~ $#* 0} {\n" + " echo usage: run.sh cmd args\n"+ " raise usage\n" + "}\n"+ "CMD = $*\n" + "{echo $CMD; dir=`{read -o 0}; cat <[0=3] > $dir/data& catpid=$apid;"+ " cat $dir/data >[1=4]; kill $catpid >[2] /dev/null} <[3=0] >[4=1] <> clone >[1=0]\n"; EMPTYDIR: con "#//dev"; rootpath := "/tmp/cpupool/"; rstyxreg: ref Registry; registered: ref Registries->Registered; CpuSession: adt { proxyid, fid, cpuid, omode, written, finished: int; stdoutopen, stdinopen: int; stdinchan, stdoutchan: chan of array of byte; closestdin,closestdout, readstdout, sync: chan of int; rcmdfinishedstdin, rcmdfinishedstdout: chan of int; fio: ref sys->FileIO; pids: list of int; }; NILCPUSESSION: con CpuSession (-1, -1,-1, 0, 0, 0, 0, 0, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil); cpusession: array of CpuSession; poolchanin : chan of string; poolchanout : chan of int; conids : array of int; CpuPool: module { init: fn (nil : ref Draw->Context, argv: list of string); }; init(nil : ref Draw->Context, argv: list of string) { sys = load Sys Sys->PATH; if (sys == nil) badmod(Sys->PATH); daytime = load Daytime Daytime->PATH; if (daytime == nil) badmod(Daytime->PATH); dial = load Dial Dial->PATH; if (dial == nil) badmod(Dial->PATH); styx = load Styx Styx->PATH; if (styx == nil) badmod(Styx->PATH); styx->init(); styxservers = load Styxservers Styxservers->PATH; if (styxservers == nil) badmod(Styxservers->PATH); styxservers->init(styx); nametree = load Nametree Nametree->PATH; if (nametree == nil) badmod(Nametree->PATH); nametree->init(); 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(); readdir = load Readdir Readdir->PATH; if (readdir == nil) badmod(Readdir->PATH); arg := load Arg Arg->PATH; if (arg == nil) badmod(Arg->PATH); sys->pctl(Sys->FORKNS | sys->NEWPGRP, nil); sys->unmount(nil, "/n/remote"); getuid(); sys->chdir(EMPTYDIR); cpusession = array[500] of { * => NILCPUSESSION }; attrs := Attributes.new(("proto", "styx") :: ("auth", "none") :: ("resource","Cpu Pool") :: nil); arg->init(argv); arg->setusage("cpupool [-a attributes] [rootdir]"); while ((opt := arg->opt()) != 0) { case opt { 'a' => attr := arg->earg(); val := arg->earg(); attrs.set(attr, val); * => arg->usage(); } } argv = arg->argv(); arg = nil; if (argv != nil) rootpath = hd argv; if (rootpath[len rootpath - 1] != '/') rootpath[len rootpath] = '/'; (n, dir) := sys->stat(rootpath); if (n == -1 || !(dir.mode & sys->DMDIR)) error("Invalid tmp path: "+rootpath); rstyxreg = Registry.new("/mnt/rstyxreg"); if (rstyxreg == nil) error("Could not find Rstyx Registry"); reg := Registry.connect(nil, nil, nil); if (reg == nil) error("Could not find registry"); (myaddr, c) := announce->announce(); if (myaddr == nil) error(sys->sprint("cannot announce: %r")); persist := 0; err: string; (registered, err) = reg.register(myaddr, attrs, persist); if (err != nil) error("could not register with registry: "+err); conids = array[200] of { * => -1 }; poolchanin = chan of string; poolchanout = chan of int; userchan := chan of int; spawn listener(c); spawn cpupoolloop(poolchanin, poolchanout); } attrval(s: string): (string, string) { for (i := 0; i < len s; i++) { if (s[i] == '=') return (s[:i], s[i+1:]); } return (nil, s); } uid: string; Qroot : con 0; Qclone: con 1; Qdata: con 2; Qsh: con 3; Qrun: con 4; Qcpu: con 5; Qsessdir: con 6; Qsessdat: con 7; getuid() { buf := array [100] of byte; fd := sys->open("/dev/user", Sys->OREAD); uidlen := sys->read(fd, buf, len buf); uid = string buf[0: uidlen]; } dir(name: string, perm: int, length: int, qid: int): Sys->Dir { d := sys->zerodir; d.name = name; d.uid = uid; d.gid = uid; d.qid.path = big qid; if (perm & Sys->DMDIR) d.qid.qtype = Sys->QTDIR; else { d.qid.qtype = Sys->QTFILE; d.length = big length; } d.mode = perm; d.atime = d.mtime = daytime->now(); return d; } defaultdirs := array[] of { ("dis", 1), ("dev", 1), ("fonts", 1), ("mnt", 0), ("prog", 0), }; serveloop(fd : ref sys->FD, cmdchan: chan of (int, string, chan of int), exitchan, sync: chan of int, proxyid: int) { if (TEST) sys->fprint(sys->fildes(2), "starting serveloop"); tchan: chan of ref Tmsg; srv: ref Styxserver; (tree, treeop) := nametree->start(); tree.create(big Qroot, dir(".",8r555 | sys->DMDIR,0,Qroot)); tree.create(big Qroot, dir("clone",8r666,0,Qclone)); tree.create(big Qroot, dir("run.sh",8r555,0,Qrun)); tree.create(big Qroot, dir("cpu",8r444,0,Qcpu)); tree.create(big Qroot, dir("data",8r777 | sys->DMDIR,0,Qdata)); tree.create(big Qroot, dir("runtime",8r444 | sys->DMDIR,0,Qsh)); for (i := 0; i < len defaultdirs; i++) tree.create(big Qroot, dir(defaultdirs[i].t0,8r555 | sys->DMDIR ,0,8 + (i<<4))); (tchan, srv) = Styxserver.new(fd,Navigator.new(treeop), big Qroot); fd = nil; datafids : list of Datafid = nil; sync <-= 1; gm: ref Tmsg; loop: for (;;) { alt { <-exitchan => break loop; gm = <-tchan => if (gm == nil) break loop; # sys->fprint(sys->fildes(2), "Got new GM %s tag: %d\n", gm.text(), gm.tag); pick m := gm { Readerror => sys->fprint(sys->fildes(2), "cpupool: fatal read error: %s\n", m.error); exit; Clunk => deldf: Datafid; (datafids, deldf) = delfid(datafids, m.fid); if (deldf.sessid != -1) { if (deldf.omode == sys->OREAD || deldf.omode == sys->ORDWR) cpusession[deldf.sessid].sync <-= STDOUTCLOSE; else if (deldf.omode == sys->OWRITE || deldf.omode == sys->ORDWR) cpusession[deldf.sessid].sync <-= STDINCLOSE; } else { sessid := getsession(m.fid); if (sessid != -1) cpusession[sessid].sync <-= CLONECLOSE; } srv.default(gm); Open => (f, nil, d, err) := srv.canopen(m); if(f == nil) { srv.reply(ref Rmsg.Error(m.tag, err)); break; } ind := int f.uname; mode := m.mode & 3; case int f.path & 15 { Qclone => if (mode == sys->OREAD) { srv.reply(ref Rmsg.Error(m.tag, "ctl cannot be open as read only")); break; } poolchanin <-= "request"; cpuid := <-poolchanout; if (cpuid == -1) srv.reply(ref Rmsg.Error(m.tag, "no free resources")); else { sessid := getsession(-1); cpusession[sessid].fid = m.fid; cpusession[sessid].cpuid = cpuid; cpusession[sessid].omode = mode; cpusession[sessid].sync = chan of int; cpusession[sessid].proxyid = proxyid; spawn sessionctl(sessid, tree); Qdir := Qsessdir | (sessid<<4); tree.create(big Qroot, dir(string sessid, 8r777 | sys->DMDIR,0, Qdir)); tree.create(big Qdir, dir("data", 8r666,0, Qsessdat | (sessid<<4))); if (TEST) sys->fprint(sys->fildes(2), "New Session %d\n\tcpuid: %d\n" ,sessid,cpuid); srv.default(gm); } Qsessdat => err = ""; sessid := (int f.path)>>4; datafids = addfid(datafids, Datafid(sessid, m.fid, mode)); if (cpusession[sessid].finished) err = "session already finished"; else if (mode == sys->OREAD || mode == sys->ORDWR) { if (cpusession[sessid].stdoutopen == -1) err = "pipe closed"; else cpusession[sessid].sync <-= STDOUTOPEN; } else if (mode == sys->OWRITE || mode == sys->ORDWR) { if (cpusession[sessid].stdinopen == -1) err = "pipe closed"; else cpusession[sessid].sync <-= STDINOPEN; } # sys->fprint(sys->fildes(2), # "Open: Data: sessid %d, stdout %d stdin %d: err: '%s'\n", # sessid,cpusession[sessid].stdoutopen, # cpusession[sessid].stdinopen, err); if (err == nil) srv.default(gm); else srv.reply(ref Rmsg.Error(m.tag, err)); * => # sys->print("Open: %s tag: %d\n", gm.text(), gm.tag); srv.default(gm); } Write => (f,e) := srv.canwrite(m); if(f == nil) { # sys->print("breaking! %r\n"); break; } case int f.path & 15 { Qsessdat => sessid := (int f.path)>>4; # sys->fprint(sys->fildes(2), "Write: Data %d len: %d\n", # sessid,len m.data); spawn datawrite(sessid,srv,m); Qclone => sessid := getsession(m.fid); # sys->fprint(sys->fildes(2), "Write: clone %d\n",sessid); spawn clonewrite(sessid,srv, m, cmdchan); * => srv.default(gm); } Read => (f,e) := srv.canread(m); if(f == nil) break; case int f.path & 15 { Qclone => sessid := getsession(m.fid); # sys->fprint(sys->fildes(2), "Read: clone %d\n",sessid); srv.reply(styxservers->readbytes(m, array of byte (string sessid + "\n"))); Qsessdat => sessid := (int f.path)>>4; # sys->fprint(sys->fildes(2), "Read: data session: %d\n",sessid); if (cpusession[sessid].finished) srv.reply(ref Rmsg.Error(m.tag, "session finished")); else spawn dataread(sessid, srv, m); Qrun => srv.reply(styxservers->readbytes(m, array of byte RUN)); Qcpu => poolchanin <-= "refresh"; s := (string ncpupool) + "\n"; srv.reply(styxservers->readbytes(m, array of byte s)); * => srv.default(gm); } * => srv.default(gm); } } } if (TEST) sys->fprint(sys->fildes(2), "leaving serveloop...\n"); tree.quit(); for (i = 0; i < len cpusession; i++) { if (cpusession[i].proxyid == proxyid) { #Tear it down! if (TEST) sys->fprint(sys->fildes(2), "Killing off session %d\n",i); poolchanin <-= "free "+string cpusession[i].cpuid; for (; cpusession[i].pids != nil; cpusession[i].pids = tl cpusession[i].pids) kill(hd cpusession[i].pids); cpusession[i] = NILCPUSESSION; } } if (TEST) sys->fprint(sys->fildes(2), "serveloop exited\n"); } dataread(sessid: int, srv: ref Styxserver, m: ref Tmsg.Read) { cpusession[sessid].readstdout <-= 1; data := <- cpusession[sessid].stdoutchan; srv.reply(ref Rmsg.Read(m.tag, data)); } datawrite(sessid: int, srv: ref Styxserver, m: ref Tmsg.Write) { # sys->fprint(sys->fildes(2), "Writing to Stdin %d (%d)\n'%s'\n", # len m.data, m.tag, string m.data); cpusession[sessid].stdinchan <-= m.data; # sys->fprint(sys->fildes(2), "Written to Stdin %d!\n",m.tag); srv.reply(ref Rmsg.Write(m.tag, len m.data)); } clonewrite(sessid: int, srv: ref Styxserver, m: ref Tmsg.Write, cmdchan: chan of (int, string, chan of int)) { if (cpusession[sessid].written) { srv.reply(ref Rmsg.Error(m.tag, "session already started")); return; } rc := chan of int; cmdchan <-= (sessid, string m.data, rc); i := <-rc; # sys->fprint(sys->fildes(2), "Sending write\n"); srv.reply(ref Rmsg.Write(m.tag, i)); } badmod(path: string) { sys->fprint(sys->fildes(1), "error CpuPool: failed to load: %s\n",path); exit; } listener(c: ref Sys->Connection) { for (;;) { nc := dial->listen(c); if (nc == nil) error(sys->sprint("listen failed: %r")); dfd := dial->accept(nc); if (dfd != nil) { sync := chan of int; sys->print("got new connection!\n"); spawn proxy(sync, dfd); <-sync; } } } proxy(sync: chan of int, dfd: ref Sys->FD) { proxypid := sys->pctl(0, nil); sys->pctl(sys->FORKNS, nil); sys->chdir(EMPTYDIR); sync <-= 1; sync = chan of int; fds := array[2] of ref sys->FD; sys->pipe(fds); cmdchan := chan of (int, string, chan of int); exitchan := chan of int; killsrvloop := chan of int; spawn serveloop(fds[0], cmdchan, killsrvloop, sync, proxypid); <-sync; if (sys->mount(fds[1], nil, "/n/remote", Sys->MREPL | sys->MCREATE, nil) == -1) error(sys->sprint("cannot mount mountfd: %r")); conid := getconid(-1); conids[conid] = 1; setupworkspace(conid); spawn exportns(dfd, conid, exitchan); for (;;) alt { (sessid, cmd, reply) := <-cmdchan => spawn runit(conid, sessid, cmd, reply); e := <-exitchan => killsrvloop <-= 1; return; } } getconid(id: int): int { for (i := 0; i < len conids; i++) if (conids[i] == id) return i; return -1; } exportns(dfd: ref Sys->FD, conid: int, exitchan: chan of int) { sys->export(dfd, "/n/remote", sys->EXPWAIT); if (TEST) sys->fprint(sys->fildes(2), "Export Finished!\n"); conids[conid] = -1; exitchan <-= 1; } error(e: string) { sys->fprint(sys->fildes(2), "CpuPool: %s: %r\n", e); raise "fail:error"; } setupworkspace(pathid: int) { path := rootpath + string pathid; sys->create(path, sys->OREAD, 8r777 | sys->DMDIR); delpath(path, 0); sys->create(path + "/data", sys->OREAD, 8r777 | sys->DMDIR); if (sys->bind(path+"/data", "/n/remote/data", sys->MREPL | sys->MCREATE) == -1) sys->fprint(sys->fildes(2), "data bind error %r\n"); sys->create(path + "/runtime", sys->OREAD, 8r777 | sys->DMDIR); if (sys->bind(path+"/runtime", "/n/remote/runtime", sys->MREPL) == -1) sys->fprint(sys->fildes(2), "runtime bind error %r\n"); for (i := 0; i < len defaultdirs; i++) { if (defaultdirs[i].t1 == 1) { sys->create(path+"/"+defaultdirs[i].t0, sys->OREAD, 8r777 | sys->DMDIR); if (sys->bind("/"+defaultdirs[i].t0, "/n/remote/"+defaultdirs[i].t0, sys->MREPL) == -1) sys->fprint(sys->fildes(2), "dir bind error %r\n"); } } } delpath(path: string, incl: int) { if (path[len path - 1] != '/') path[len path] = '/'; (dirs, n) := readdir->init(path, readdir->NONE | readdir->COMPACT); for (i := 0; i < n; i++) { if (dirs[i].mode & sys->DMDIR) delpath(path + dirs[i].name, 1); else sys->remove(path + dirs[i].name); } if (incl) sys->remove(path); } runit(id, sessid: int, cmd: string, sync: chan of int) { # sys->print("got runit!\n"); cpusession[sessid].sync <-= PID; cpusession[sessid].sync <-= sys->pctl(sys->FORKNS, nil); if (!TEST && sys->bind("/net.alt", "/net", sys->MREPL) == -1) { sys->fprint(sys->fildes(2), "cpupool net.alt bind failed: %r\n"); sync <-= -1; return; } path := rootpath + string id; runfile := "/runtime/start"+string cpusession[sessid].cpuid+".sh"; sh := load Sh Sh->PATH; if(sh == nil) { sys->fprint(sys->fildes(2), "Failed to load sh: %r\n"); sync <-= -1; return; } sys->remove(path+runfile); fd := sys->create(path+runfile, sys->OWRITE, 8r777); if (fd == nil) { sync <-= -1; return; } sys->fprint(fd, "#!/dis/sh\n"); sys->fprint(fd, "bind /prog /n/client/prog\n"); sys->fprint(fd, "bind /n/client /\n"); sys->fprint(fd, "cd /\n"); sys->fprint(fd, "%s\n", cmd); if (sys->bind("#s", "/n/remote/runtime", Sys->MBEFORE|Sys->MCREATE) == -1) { sys->fprint(sys->fildes(2), "cpupool: %r\n"); return; } cpusession[sessid].fio = sys->file2chan("/n/remote/runtime", "mycons"); if (cpusession[sessid].fio == nil) { sys->fprint(sys->fildes(2), "cpupool: file2chan failed: %r\n"); return; } if (sys->bind("/n/remote/runtime/mycons", "/n/remote/dev/cons", sys->MREPL) == -1) sys->fprint(sys->fildes(2), "cons bind error %r\n"); cpusession[sessid].written = 1; cpusession[sessid].stdinchan = chan of array of byte; cpusession[sessid].closestdin = chan of int; cpusession[sessid].rcmdfinishedstdin = chan of int; spawn devconsread(sessid); cpusession[sessid].stdoutchan = chan of array of byte; cpusession[sessid].closestdout = chan of int; cpusession[sessid].readstdout = chan of int; cpusession[sessid].rcmdfinishedstdout = chan of int; spawn devconswrite(sessid); # Let it know that session channels have been created & can be listened on... sync <-= len cmd; # would prefer that it were authenticated if (TEST) sys->print("ABOUT TO RCMD\n"); sh->run(nil, "rcmd" :: "-A" :: "-e" :: "/n/remote" :: cpupool[cpusession[sessid].cpuid].srvc.addr :: "sh" :: "-c" :: "/n/client"+runfile :: nil); if (TEST) sys->print("DONE RCMD\n"); sys->remove(path+runfile); sys->unmount(nil, "/n/remote/dev/cons"); cpusession[sessid].rcmdfinishedstdin <-= 1; cpusession[sessid].rcmdfinishedstdout <-= 1; cpusession[sessid].sync <-= FINISHED; } CLONECLOSE: con 0; FINISHED: con 1; STDINOPEN: con 2; STDINCLOSE: con 3; STDOUTOPEN: con 4; STDOUTCLOSE: con 5; PID: con -2; sessionctl(sessid: int, tree: ref Nametree->Tree) { cpusession[sessid].pids = sys->pctl(0, nil) :: nil; clone := 1; closed := 0; main: for (;;) { i := <-cpusession[sessid].sync; case i { PID => pid := <-cpusession[sessid].sync; if (TEST) sys->fprint(sys->fildes(2), "adding PID: %d\n", pid); cpusession[sessid].pids = pid :: cpusession[sessid].pids; STDINOPEN => cpusession[sessid].stdinopen++; if (TEST) sys->fprint(sys->fildes(2), "%d: Open stdin: => %d\n", sessid, cpusession[sessid].stdinopen); STDOUTOPEN => cpusession[sessid].stdoutopen++; if (TEST) sys->fprint(sys->fildes(2), "%d: Open stdout: => %d\n", sessid, cpusession[sessid].stdoutopen); STDINCLOSE => cpusession[sessid].stdinopen--; if (TEST) sys->fprint(sys->fildes(2), "%d: Close stdin: => %d\n", sessid, cpusession[sessid].stdinopen); if (cpusession[sessid].stdinopen == 0) { cpusession[sessid].stdinopen = -1; cpusession[sessid].closestdin <-= 1; } # sys->fprint(sys->fildes(2), "Clunk: stdin (in %d: out %d\n", # cpusession[sessid].stdinopen, cpusession[sessid].stdoutopen); STDOUTCLOSE => cpusession[sessid].stdoutopen--; if (TEST) sys->fprint(sys->fildes(2), "%d: Close stdout: => %d\n", sessid, cpusession[sessid].stdoutopen); if (cpusession[sessid].stdoutopen == 0) { cpusession[sessid].stdoutopen = -1; cpusession[sessid].closestdout <-= 1; } #sys->fprint(sys->fildes(2), "Clunk: stdout (in %d: out %d\n", # cpusession[sessid].stdinopen, cpusession[sessid].stdoutopen); CLONECLOSE => if (TEST) sys->fprint(sys->fildes(2), "%d: Close clone\n", sessid); clone = 0; #sys->fprint(sys->fildes(2), "Clunk: clone (in %d: out %d\n", # cpusession[sessid].stdinopen, cpusession[sessid].stdoutopen); FINISHED => if (TEST) sys->fprint(sys->fildes(2), "%d: Rcmd finished", sessid); cpusession[sessid].finished = 1; poolchanin <-= "free "+string cpusession[sessid].cpuid; if (closed) break main; } if (cpusession[sessid].stdinopen <= 0 && cpusession[sessid].stdoutopen <= 0 && clone == 0) { closed = 1; tree.remove(big (Qsessdir | (sessid<<4))); tree.remove(big (Qsessdat | (sessid<<4))); if (cpusession[sessid].finished || !cpusession[sessid].written) break main; } } if (!cpusession[sessid].finished) # ie never executed anything poolchanin <-= "free "+string cpusession[sessid].cpuid; cpusession[sessid] = NILCPUSESSION; if (TEST) sys->fprint(sys->fildes(2), "closing session %d\n",sessid); } devconswrite(sessid: int) { cpusession[sessid].sync <-= PID; cpusession[sessid].sync <-= sys->pctl(0, nil); stdouteof := 0; file2chaneof := 0; rcmddone := 0; main: for (;;) alt { <-cpusession[sessid].rcmdfinishedstdout => rcmddone = 1; if (file2chaneof) break main; <-cpusession[sessid].closestdout => stdouteof = 1; (offset, d, fid, wc) := <-cpusession[sessid].fio.write => if (wc != nil) { # sys->fprint(sys->fildes(2), "stdout: '%s'\n", string d); if (stdouteof) { # sys->fprint(sys->fildes(2), "stdout: sending EOF\n"); wc <-= (0, nil); continue; } alt { <-cpusession[sessid].closestdout => # sys->print("got closestdout\n"); wc <-= (0, nil); stdouteof = 1; <-cpusession[sessid].readstdout => cpusession[sessid].stdoutchan <-= d; wc <-= (len d, nil); } } else { # sys->fprint(sys->fildes(2), "got nil wc\n"); file2chaneof = 1; if (rcmddone) break main; } } # No more input at this point as rcmd has finished; if (stdouteof || cpusession[sessid].stdoutopen == 0) { # sys->print("leaving devconswrite\n"); return; } for (;;) alt { <-cpusession[sessid].closestdout => # sys->print("got closestdout\n"); # sys->print("leaving devconswrite\n"); return; <- cpusession[sessid].readstdout => cpusession[sessid].stdoutchan <-= nil; } } devconsread(sessid: int) { cpusession[sessid].sync <-= PID; cpusession[sessid].sync <-= sys->pctl(0, nil); stdineof := 0; file2chaneof := 0; rcmddone := 0; main: for (;;) alt { <-cpusession[sessid].rcmdfinishedstdin => rcmddone = 1; if (file2chaneof) break main; <-cpusession[sessid].closestdin => # sys->print("got stdin close\n"); stdineof = 1; (offset, count, fid, rc) := <-cpusession[sessid].fio.read => if (rc != nil) { # sys->fprint(sys->fildes(2), "devconsread: '%d %d'\n", count, offset); if (stdineof) { rc <-= (nil, nil); continue; } alt { data := <-cpusession[sessid].stdinchan => # sys->print("got data len %d\n", len data); rc <-= (data, nil); <-cpusession[sessid].closestdin => # sys->print("got stdin close\n"); stdineof = 1; rc <-= (nil, nil); } } else { # sys->print("got nil rc\n"); file2chaneof = 1; if (rcmddone) break main; } } if (!stdineof && cpusession[sessid].stdinopen != 0) <-cpusession[sessid].closestdin; # sys->fprint(sys->fildes(2), "Leaving devconsread\n"); } Srvcpool: adt { srvc: ref Service; inuse: int; }; cpupool: array of Srvcpool; ncpupool := 0; cpupoolloop(chanin: chan of string, chanout: chan of int) { cpupool = array[200] of Srvcpool; for (i := 0; i < len cpupool; i++) cpupool[i] = Srvcpool (nil, 0); wait := 0; for (;;) { inp := <-chanin; # sys->print("poolloop: '%s'\n",inp); (nil, lst) := sys->tokenize(inp, " \t\n"); case hd lst { "refresh" => if (daytime->now() - wait >= 60) { refreshcpupool(); wait = daytime->now(); } "request" => if (daytime->now() - wait >= 60) { refreshcpupool(); wait = daytime->now(); } found := -1; # sys->print("found %d services...\n", ncpupool); for (i = 0; i < ncpupool; i++) { if (!cpupool[i].inuse) { found = i; cpupool[i].inuse = 1; break; } } # sys->print("found service %d\n", found); chanout <-= found; "free" => if (TEST) sys->print("freed service %d\n", int hd tl lst); cpupool[int hd tl lst].inuse = 0; } } } refreshcpupool() { (lsrv, err) := rstyxreg.find(("resource", "Rstyx resource") :: nil); # sys->print("found %d resources\n",len lsrv); if (err != nil) return; tmp := array[len cpupool] of Srvcpool; ntmp := len lsrv; i := 0; for (;lsrv != nil; lsrv = tl lsrv) tmp[i++] = Srvcpool(hd lsrv, 0); min := 0; for (i = 0; i < ntmp; i++) { for (j := min; j < ncpupool; j++) { if (tmp[i].srvc.addr == cpupool[j].srvc.addr) { if (j == min) min++; tmp[i].inuse = cpupool[j].inuse; } } } ncpupool = ntmp; for (i = 0; i < ntmp; i++) cpupool[i] = tmp[i]; # sys->print("ncpupool: %d\n",ncpupool); } getsession(fid: int): int { for (i := 0; i < len cpusession; i++) if (cpusession[i].fid == fid) return i; return -1; } kill(pid: int) { if ((fd := sys->open("/prog/" + string pid + "/ctl", Sys->OWRITE)) != nil) sys->fprint(fd, "kill"); } killg(pid: int) { if ((fd := sys->open("/prog/" + string pid + "/ctl", Sys->OWRITE)) != nil) sys->fprint(fd, "killgrp"); } delfid(datafids: list of Datafid, fid: int): (list of Datafid, Datafid) { rdf := Datafid (-1, -1, -1); tmp : list of Datafid = nil; for (; datafids != nil; datafids = tl datafids) { testdf := hd datafids; if (testdf.fid == fid) rdf = testdf; else tmp = testdf :: tmp; } return (tmp, rdf); } addfid(datafids: list of Datafid, df: Datafid): list of Datafid { (datafids, nil) = delfid(datafids, df.fid); return df :: datafids; } Datafid: adt { sessid, fid, omode: int; };