gang2fs: initial decomposition

WARNING: does not build

ericvh committed 15 years ago

Commit 7e7008e1 parent caad9483 files · +1365 −1370Browse files

Unified · Split

3 files

sys/src/cmd/uem/gang.c +913 −0
sys/src/cmd/uem/gang2fs.c +8 −1370
sys/src/cmd/uem/gangmon.c +444 −0

@@ -0,0 +1,913 @@
	1	+/* remote support */
	2	+static void
	3	+startimport(void *arg)
	4	+{
	5	+ char addr = (char ) arg;
	6	+ char *name = smprint("%s/%s", amprefix, addr);
	7	+
	8	+ threadsetname("gangfs-startimport");
	9	+
	10	+ // FIXME: should arg be replaced with addr
	11	+ DPRINT(DCUR, "startimport: importing name=%s\n");
	12	+ procexecl(nil, "/bin/import", "import", "-A", arg, "/",
	13	+ name, nil);
	14	+
	15	+ DPRINT(DERR, "ERROR: startimport: procexecl returned: %r\n");
	16	+ threadexits(nil);
	17	+}
	18	+
	19	+/* checkmount: check to see if a remote system is already, if not mount it */
	20	+static int
	21	+checkmount(char *addr)
	22	+{
	23	+ char *dest = estrdup9p(addr);
	24	+ int retries = 0;
	25	+ int err = 0;
	26	+ int fd;
	27	+ char *srvpt = smprint("/srv/%s", addr);
	28	+ char *srvtg = smprint("%s/%s", amprefix, addr);
	29	+ char *mtpt = smprint("%s/proc", srvtg);
	30	+ Dir *tmp = dirstat(mtpt);
	31	+
	32	+ DPRINT(DCUR, "checkmount: checking mount point (%s)\n", srvtg);
	33	+
	34	+ if(tmp != nil) {
	35	+ DPRINT(DEXE, "\tsrv already mounted on (%s)\n", mtpt);
	36	+ goto out;
	37	+ } else
	38	+
	39	+ DPRINT(DFID, "\tmount point (%s) does not exist... mounting\n", srvtg);
	40	+ DPRINT(DCUR, "\tattempting to mount %s on %s\n", srvpt, srvtg);
	41	+ fd = open(srvpt, ORDWR);
	42	+ if(fd > 0) {
	43	+ DPRINT(DCUR, "\tfd=(%d)\n", fd);
	44	+ if(mount(fd, -1, srvtg, MREPL, "") < 0) {
	45	+ // FIXME: what should we do here?
	46	+ DPRINT(DERR, "\tERROR: srv mount failed: %r\n");
	47	+ close(fd);
	48	+ } else {
	49	+ DPRINT(DFID, "\tsrv mount succeeded");
	50	+ DPRINT(DFID, "\t\tMOUNT");
	51	+ close(fd);
	52	+ goto out;
	53	+ }
	54	+ }
	55	+
	56	+ /* no dir, spawn off an import */
	57	+ DPRINT(DFID, "\tno dir, spawn off an import +++\n");
	58	+ proccreate(startimport, dest, STACK);
	59	+
	60	+ /* we need to wait for this to be done */
	61	+ while((tmp = dirstat(mtpt)) == nil) {
	62	+ // start a mount with a timeout...
	63	+ // use wait....
	64	+ if(retries++ > 100) {
	65	+ DPRINT(DERR, "ERROR: checkmount: import not complete after 10 seconds, giving up\n");
	66	+ sleep(100);
	67	+ err = -1;
	68	+ break;
	69	+ }
	70	+ }
	71	+
	72	+out:
	73	+ if(srvpt)
	74	+ free(srvpt);
	75	+ if(srvtg)
	76	+ free(srvtg);
	77	+ if(tmp)
	78	+ free(tmp);
	79	+ free(dest);
	80	+ free(mtpt);
	81	+ return err;
	82	+}
	83	+
	84	+static void
	85	+gangrefinc(Gang g, char where)
	86	+{
	87	+ g->refcount++;
	88	+ DPRINT(DREF, "\t\tGang(%p).refcount++ = %d -> %d[%s]\n", g, g->refcount-1, g->refcount, where);
	89	+}
	90	+
	91	+static void
	92	+gangrefdec(Gang g, char where)
	93	+{
	94	+ g->refcount--;
	95	+ // reality check
	96	+ if(g->refcount < 0) {
	97	+ DPRINT(DERR, "%s: WARNING gangrefdec is less than 0. Resetting to 0\n", where);
	98	+ g->refcount = 0;
	99	+ }
	100	+ DPRINT(DREF, "\t\tGang(%p).refcount-- = %d -> %d[%s]\n", g, g->refcount+1, g->refcount, where);
	101	+}
	102	+
	103	+static int
	104	+mpipe(char path, char name)
	105	+{
	106	+ int fd, ret;
	107	+ char *srvpt = smprint("/srv/%s", srvmpipe);
	108	+ fd = open(srvpt, ORDWR);
	109	+ if(fd<0) {
	110	+ DPRINT(DERR, "ERROR: couldn't open %s: %r\n", srvpt);
	111	+ free(srvpt);
	112	+ return -1;
	113	+ }
	114	+ free(srvpt);
	115	+
	116	+ //wunlock(&glock);
	117	+ DPRINT(DFID, "mpipe: mounting name=(%s) on path=(%s) pid=(%d) fd=%d\n",
	118	+ name, path, getpid(), fd);
	119	+ DPRINT(DCUR, "\tfrom mysysname=(%s)\n", mysysname);
	120	+ ret = mount(fd, -1, path, MBEFORE, name);
	121	+ //wunlock(&glock);
	122	+ if(ret < 0) {
	123	+ DPRINT(DERR, "mount of multipipe failed: %r\n");
	124	+ } else
	125	+ DPRINT(DFID, "\t\tMOUNT");
	126	+
	127	+ close(fd);
	128	+ return ret;
	129	+}
	130	+
	131	+static int
	132	+flushmp(char *src)
	133	+{
	134	+ int fd = open(src, OWRITE);
	135	+ int n;
	136	+ char hdr[255];
	137	+ ulong tag = ~0; /* header byte is at offset ~0 */
	138	+ char pkttype='.';
	139	+
	140	+ if(fd < 0) {
	141	+ DPRINT(DPIP, "flushmp: open failed: %r\n");
	142	+ return fd;
	143	+ }
	144	+
	145	+ n = snprint(hdr, 255, "%c\n%lud\n%lud\n%s\n",
	146	+ pkttype, (ulong)0, (ulong)0, "");
	147	+ DPRINT(DCUR, "flushmp: src=(%s) pid=(%d)\n", src, getpid());
	148	+ DPRINT(DCUR, "\tfd=(%d)\n", fd);
	149	+
	150	+ n = pwrite(fd, hdr, n, tag);
	151	+ close(fd);
	152	+ return n;
	153	+}
	154	+
	155	+static int
	156	+spliceto(char src, char dest)
	157	+{
	158	+ int fd = open(src, OWRITE);
	159	+ int n;
	160	+ char hdr[255];
	161	+ ulong tag = ~0; /* header byte is at offset ~0 */
	162	+ char pkttype='>';
	163	+
	164	+ if(fd < 0) {
	165	+ DPRINT(DPIP, "spliceto: open %s failed: %r\n", src);
	166	+ return fd;
	167	+ }
	168	+
	169	+ DPRINT(DCUR, "spliceto: fd=(%d)\n", fd);
	170	+ n = snprint(hdr, 255, "%c\n%lud\n%lud\n%s\n", pkttype, (ulong)0, (ulong)0, dest);
	171	+ n = pwrite(fd, hdr, n, tag);
	172	+ close(fd);
	173	+ return n;
	174	+}
	175	+
	176	+static int
	177	+splicefrom(char dest, char src)
	178	+{
	179	+ int fd = open(dest, OWRITE);
	180	+ int n;
	181	+ char hdr[255];
	182	+ ulong tag = ~0; /* header byte is at offset ~0 */
	183	+ char pkttype='<';
	184	+
	185	+ if(fd < 0) {
	186	+ DPRINT(DERR, "ERROR: splicefrom: open %s failed: %r\n", dest);
	187	+ return fd;
	188	+ }
	189	+
	190	+ DPRINT(DCUR, "splicefrom: fd=(%d)\n", fd);
	191	+ DPRINT(DEXE, "just so we know where we are splicing: [%s]\n", src);
	192	+
	193	+ n = snprint(hdr, 255, "%c\n%lud\n%lud\n%s\n", pkttype, (ulong)0, (ulong)0, src);
	194	+ n = pwrite(fd, hdr, n, tag);
	195	+ close(fd);
	196	+ return n;
	197	+}
	198	+
	199	+static Gang *
	200	+newgang(void)
	201	+{
	202	+ Gang *mygang = emalloc9p(sizeof(Gang));
	203	+
	204	+ if(mygang == nil)
	205	+ return nil;
	206	+ memset(mygang, 0, sizeof(Gang));
	207	+ mygang->chan = chancreate(sizeof(void *), 0);
	208	+
	209	+ wlock(&glock);
	210	+ if(glist == nil) {
	211	+ glist = mygang;
	212	+ mygang->index = 0;
	213	+ } else {
	214	+ int last = -1;
	215	+ Gang *current;
	216	+ for(current=glist; current->next != nil; current = current->next) {
	217	+ if(current->index != last+1) {
	218	+ break;
	219	+ }
	220	+ last++;
	221	+ }
	222	+ mygang->next = current->next;
	223	+ current->next = mygang;
	224	+ mygang->index = current->index+1;
	225	+ }
	226	+
	227	+ mygang->ctlref = 1;
	228	+ mygang->refcount = 1;
	229	+ mygang->imode = CMbcast; /* broadcast mode default */
	230	+ mygang->status = GANG_NEW;
	231	+ wunlock(&glock);
	232	+
	233	+ return mygang;
	234	+}
	235	+
	236	+static Gang *
	237	+findgang(int index)
	238	+{
	239	+ Gang *current = nil;
	240	+
	241	+ rlock(&glock);
	242	+ if(glist == nil) {
	243	+ goto out;
	244	+ } else {
	245	+ for(current=glist; current != nil; current = current->next) {
	246	+ if((current->index == index)&&(current->status != GANG_CLOSED)) {
	247	+ // gangrefinc(current, "findgang");
	248	+ goto out;
	249	+ }
	250	+ }
	251	+ }
	252	+out:
	253	+ runlock(&glock);
	254	+ return current;
	255	+}
	256	+
	257	+
	258	+
	259	+static Gang *
	260	+findgangnum(int which)
	261	+{
	262	+ Gang *current = nil;
	263	+ rlock(&glock);
	264	+ if(glist == nil) {
	265	+ goto out;
	266	+ } else {
	267	+ int count = 0;
	268	+ for(current=glist; current != nil; current = current->next) {
	269	+ if(current->status == GANG_CLOSED)
	270	+ continue;
	271	+ if(count == which) {
	272	+ // gangrefinc(current, "findgangnum");
	273	+ goto out;
	274	+ }
	275	+ count++;
	276	+ }
	277	+ }
	278	+out:
	279	+ runlock(&glock);
	280	+ return current;
	281	+}
	282	+
	283	+
	284	+static void
	285	+releasesessions(Gang *g)
	286	+{
	287	+ int count;
	288	+
	289	+ DPRINT(DEXE, "releasesessions:\n");
	290	+ for(count = 0; count < g->size; count++) {
	291	+ DPRINT(DEXE, "\tsess[%d].path=(%s)\n", count, g->sess[count].path);
	292	+ DPRINT(DEXE, "\tsess[%d].chan=(%d)\n", count, g->sess[count].chan);
	293	+ DPRINT(DEXE, "\tsess[%d].fd=(%d)\n", count, g->sess[count].fd);
	294	+ free(g->sess[count].path);
	295	+ chanfree(g->sess[count].chan);
	296	+ close(g->sess[count].fd);
	297	+ }
	298	+
	299	+ free(g->sess);
	300	+ g->sess = nil;
	301	+}
	302	+
	303	+static void
	304	+flushgang(void *arg)
	305	+{
	306	+ Gang *g = arg;
	307	+ char fname[255];
	308	+
	309	+ DPRINT(DCLN, "flushgang: flushing stdin, stdout, and stderr\n");
	310	+ snprint(fname, 255, "%s/g%d/stdin", gangpath, g->index);
	311	+ flushmp(fname); /* flush pipe to be sure */
	312	+ snprint(fname, 255, "%s/g%d/stdout", gangpath, g->index);
	313	+ flushmp(fname);
	314	+ snprint(fname, 255, "%s/g%d/stderr", gangpath, g->index);
	315	+ flushmp(fname);
	316	+}
	317	+
	318	+
	319	+static void
	320	+printgang(Gang *g)
	321	+{
	322	+ rlock(&glock);
	323	+
	324	+ DPRINT(DCUR, "\tprintgang:\n");
	325	+ DPRINT(DCUR, "\t\tpath=%s\n", g->path);
	326	+ DPRINT(DCUR, "\t\trefcount=%d\n", g->refcount);
	327	+ DPRINT(DCUR, "\t\tctlref=%d\n", g->ctlref);
	328	+ DPRINT(DCUR, "\t\tstatus=%d\n", g->status);
	329	+ DPRINT(DCUR, "\t\tindex=%d\n", g->index);
	330	+ DPRINT(DCUR, "\t\tsize=%d\n", g->size);
	331	+ DPRINT(DCUR, "\t\timode=%d\n", g->imode);
	332	+ DPRINT(DCUR, "\t\t*** sessions not reported\n");
	333	+
	334	+ runlock(&glock);
	335	+}
	336	+
	337	+static void
	338	+printganglist(void)
	339	+{
	340	+ Gang *g;
	341	+
	342	+ DPRINT(DCUR, "printganglist:\n");
	343	+ for(g = glist; g != nil; g = g->next)
	344	+ printgang(g);
	345	+}
	346	+
	347	+static void
	348	+cleanupgang(void *arg)
	349	+{
	350	+ Gang *g = arg;
	351	+ char fname[255];
	352	+
	353	+ /* Clean up subsessions first so they flush data to us */
	354	+ releasesessions(g);
	355	+
	356	+ /* make sure that the std files are completely flushed */
	357	+ flushgang(arg);
	358	+
	359	+ /* unmount the std files and remove their references */
	360	+ DPRINT(DCLN, "cleanupgang: unmounting stdin, stdout, and stderr\n");
	361	+ snprint(fname, 255, "%s/g%d/stdin", gangpath, g->index);
	362	+ if(unmount(0, fname) == -1)
	363	+ DPRINT(DERR, "cleanupgang: ERROR unable to unmount %s: %r\n", fname);
	364	+
	365	+ snprint(fname, 255, "%s/g%d/stdout", gangpath, g->index);
	366	+ if(unmount(0, fname) == -1)
	367	+ DPRINT(DERR, "cleanupgang: ERROR unable to unmount %s: %r\n", fname);
	368	+
	369	+ snprint(fname, 255, "%s/g%d/stderr", gangpath, g->index);
	370	+ if(unmount(0, fname) == -1)
	371	+ DPRINT(DERR, "cleanupgang: ERROR unable to unmount %s: %r\n", fname);
	372	+
	373	+ /* close it out */
	374	+ g->status=GANG_CLOSED;
	375	+ /* now what? */
	376	+
	377	+printgang(g);
	378	+}
	379	+
	380	+
	381	+static void
	382	+releasegang(Gang *g)
	383	+{
	384	+ wlock(&glock);
	385	+ DPRINT(DEXE, "releasegang:\n");
	386	+
	387	+//need to decrement the ref counter here
	388	+ gangrefdec(g, "releasegang");
	389	+// if((g->ctlref == 0)&&(g->refcount == 0)) { /* clean up */
	390	+ if(g->ctlref == 0) { /* clean up */
	391	+ Gang *c;
	392	+ Gang *p = glist;
	393	+
	394	+ if(g->refcount != 0)
	395	+ DPRINT(DCUR, "WARNING: releasegang, ctlrefcount=0, gang refcount=%d\n", g->refcount);
	396	+
	397	+ if(glist == g) {
	398	+ glist = g->next;
	399	+ } else {
	400	+ for(c = glist; c != nil; c = c->next) {
	401	+ if(c == g) {
	402	+ p->next = g->next;
	403	+ break;
	404	+ }
	405	+ p = c;
	406	+ }
	407	+ }
	408	+
	409	+ chanfree(g->chan);
	410	+ free(g);
	411	+ }
	412	+
	413	+ wunlock(&glock);
	414	+}
	415	+
	416	+static void
	417	+cleanup(Srv *)
	418	+{
	419	+ DPRINT(DCUR, "cleanup: pid=(%d)\n", getpid());
	420	+ nbsendp(iochan, 0); /* flush any blocked readers */
	421	+
	422	+ // FIXME: is this necessary? the iothread is being held open...
	423	+ DPRINT(DFID, "cleanup: iothread's pid=(%d)\n", threadpid(iothread_id));
	424	+ threadkill(iothread_id);
	425	+
	426	+ chanclose(iochan);
	427	+ closestatus(&mystats);
	428	+ sleep(5);
	429	+ DPRINT(DCUR, "cleanup: killing remaing threads\n");
	430	+ threadexitsall("done");
	431	+}
	432	+
	433	+/* mypath is the canonical path to my gproc */
	434	+/* path is the path to the target proc */
	435	+static char *
	436	+setupstdio(Session *s)
	437	+{
	438	+ Gang *g = s->g;
	439	+ char buf[255];
	440	+ char dest[255];
	441	+ int n;
	442	+
	443	+ DPRINT(DEXE, "setupstdio: path=(%s) remote=(%d)\n", s->path, s->remote);
	444	+ DPRINT(DEXE, "\tgangpath=(%s) xgangpath=(%s)\n", gangpath, xgangpath);
	445	+
	446	+ /* BUG: won't work for gang because gang pid is not an int */
	447	+ if(s->remote)
	448	+ snprint(buf, 255, "%s/g%d", s->path, s->pid);
	449	+ else
	450	+ snprint(buf, 255, "%s/%d", s->path, s->pid);
	451	+
	452	+ snprint(dest, 255, "%s/g%d/%d", gangpath, g->index, s->index);
	453	+ n = bind(buf, dest, MREPL);
	454	+ DPRINT(DEXE, "\tbind buf=(%s) dest=(%s) ret=%d\n", buf, dest, n);
	455	+ if(n < 0)
	456	+ return smprint("couldn't bind %s %s: %r\n", buf, dest);
	457	+
	458	+// FIXME: the rest of the gangs are actually xgangs...
	459	+ if(s->remote)
	460	+ snprint(buf, 255, "%s/g%d/stdin", s->path, s->pid);
	461	+ else
	462	+ snprint(buf, 255, "%s/%d/stdin", s->path, s->pid);
	463	+ snprint(dest, 255, "%s/g%d/stdin", xgangpath, g->index);
	464	+ DPRINT(DEXE, "\tsplicefrom buf=(%s) dest=(%s)\n", buf, dest);
	465	+ n = splicefrom(buf, dest); /* execfs initiates splicefrom gangfs */
	466	+ DPRINT(DEXE, "\t\tret=%d\n", n);
	467	+ if(n < 0)
	468	+ return smprint("setupstdio: %r\n");
	469	+
	470	+ if(s->remote)
	471	+ snprint(buf, 255, "%s/g%d/stdout", s->path, s->pid);
	472	+ else
	473	+ snprint(buf, 255, "%s/%d/stdout", s->path, s->pid);
	474	+ snprint(dest, 255, "%s/g%d/stdout", xgangpath, g->index);
	475	+ DPRINT(DEXE, "\tspliceto buf=(%s) dest=(%s)\n", buf, dest);
	476	+ n = spliceto(buf, dest); /* execfs initiates spliceto gangfs stdout */
	477	+ DPRINT(DEXE, "\t\tret=%d\n", n);
	478	+ if(n < 0)
	479	+ return smprint("setupstdio: %r\n");
	480	+
	481	+ if(s->remote)
	482	+ snprint(buf, 255, "%s/g%d/stderr", s->path, s->pid);
	483	+ else
	484	+ snprint(buf, 255, "%s/%d/stderr", s->path, s->pid);
	485	+ snprint(dest, 255, "%s/g%d/stderr", xgangpath, g->index);
	486	+ DPRINT(DEXE, "\tspliceto buf=(%s) dest=(%s)\n", buf, dest);
	487	+ n = spliceto(buf, dest); /* execfs initiates spliceto gangfs stderr */
	488	+ DPRINT(DEXE, "\t\tret=%d\n", n);
	489	+ if(n < 0)
	490	+ return smprint("setupstdio: %r\n");
	491	+
	492	+ return nil;
	493	+}
	494	+
	495	+
	496	+/* reserve a session using clone semantics */
	497	+/* NOTE: Should work with execfs or gangfs */
	498	+static void
	499	+cloneproc(void *arg)
	500	+{
	501	+ Session sess = (Session ) arg;
	502	+ char buf[256];
	503	+ int retries;
	504	+ int n = -1;
	505	+ char *err;
	506	+
	507	+ if(sess->remote)
	508	+ snprint(buf, 255, "%s/gclone", sess->path);
	509	+ else
	510	+ snprint(buf, 255, "%s/clone", sess->path);
	511	+
	512	+ DPRINT(DEXE, "cloneproc: **** clone proc %s (remote=%d)\n",
	513	+ buf, sess->remote);
	514	+ for(retries=0; retries <= 10; retries++) {
	515	+ sess->fd = open(buf, ORDWR);
	516	+
	517	+ if(sess->fd <= 0) {
	518	+ sleep (100);
	519	+ DPRINT(DERR, "WARNING: retry: %d opening execfs returned %d: %r -- retrying\n",
	520	+ retries, sess->fd);
	521	+ continue;
	522	+ }
	523	+
	524	+ n = pread(sess->fd, buf, 255, 0);
	525	+ if(n >= 0)
	526	+ break;
	527	+
	528	+ DPRINT(DERR, "WARNING: retry: %d execfs's pread=(%d) failed for fd=(%d): %r -- retrying\n",
	529	+ retries, n, sess->fd);
	530	+ close (sess->fd);
	531	+ }
	532	+ DPRINT(DCUR, "\tcontrol channel on sess->fd=(%d) for subsession (%d) pread=(%d)\n",
	533	+ sess->fd, sess->subindex, n);
	534	+
	535	+ if(retries > 10) {
	536	+ DPRINT(DERR, "ERROR giving up\n");
	537	+ sendp(sess->chan, smprint("execfs returning %r"));
	538	+ threadexits(Ebadpid);
	539	+ }
	540	+
	541	+ if(n < 0) {
	542	+ sendp(sess->chan,
	543	+ smprint("couldn't read from session ctl path=%s/clone fd=(%d) n=(%d): %r\n",
	544	+ sess->path, sess->fd, n));
	545	+ threadexits(Ebadctl);
	546	+ }
	547	+ buf[n] = 0; // FIXME: need to terminate the buffer or we get junk...
	548	+
	549	+ if(sess->remote)
	550	+ sess->pid = atoi(buf+1); /* skip the g */
	551	+ else
	552	+ sess->pid = atoi(buf); /* convert to local execs session number */
	553	+
	554	+ DPRINT(DEXE, "cloneproc: sess->pid=%ld buf=(%s)\n", sess->pid, buf);
	555	+
	556	+ /* if we are a remote session, setup backmount */
	557	+ if(sess->remote) {
	558	+ DPRINT(DEXE, "Attempting to remote checkmount %s on %s\n",
	559	+ mysysname, sess->path);
	560	+ n = fprint(sess->fd, "checkmount %s", mysysname);
	561	+ if(n < 0) {
	562	+ DPRINT(DEXE, "setting up backmount failed: %r\n");
	563	+ chanprint(sess->chan, "setting up backmount failed: %r\n");
	564	+ goto error;
	565	+ }
	566	+
	567	+ if(sess->g->imode == CMenum) {
	568	+ n = fprint(sess->fd, "enum");
	569	+ if(n < 0) {
	570	+ DPRINT(DEXE, "setting up enum failed: %r\n");
	571	+ chanprint(sess->chan, "setting up enum failed: %r\n");
	572	+ goto error;
	573	+ }
	574	+ }
	575	+
	576	+ n = fprint(sess->fd, "res %d", sess->remote);
	577	+ if(n < 0) {
	578	+ DPRINT(DEXE, "remote reservation failed: %r\n");
	579	+ chanprint(sess->chan, "remote reservation failed: %r\n");
	580	+ goto error;
	581	+ }
	582	+ }
	583	+
	584	+ /* finish by setting up stdio */
	585	+ if(err = setupstdio(sess)) {
	586	+ DPRINT(DEXE, "reporting failure from stdio setup session %d:%d: %s\n",
	587	+ sess->index, sess->subindex, err);
	588	+ sendp(sess->chan, err);
	589	+ } else {
	590	+ DPRINT(DEXE, "reporting success from session %d:%d\n", sess->index, sess->subindex);
	591	+ sendp(sess->chan, nil);
	592	+ }
	593	+
	594	+error:
	595	+ threadexits(nil);
	596	+}
	597	+
	598	+
	599	+static void
	600	+setupsess(Gang g, Session s, char *path, int r, int sub)
	601	+{
	602	+ if(path){
	603	+ s->path = smprint("%s/%s/proc", amprefix, path);
	604	+ } else {
	605	+ s->path = smprint(execfspath);
	606	+ }
	607	+ DPRINT(DEXE, "setupsess: path=(%s) gangpath=(%s) execfspath=(%s) \n",
	608	+ path, gangpath, execfspath);
	609	+ DPRINT(DEXE, "\tresult session path=(%s)\n", s->path);
	610	+
	611	+ s->r = nil;
	612	+ s->chan = chancreate(sizeof(char *), 0);
	613	+ s->g = g; /* back pointer */
	614	+
	615	+ // remote_override does not work here as r = the number of sessions
	616	+ s->remote = r;
	617	+
	618	+ s->subindex = sub; /* keep the subsession index to help with debugging */
	619	+}
	620	+
	621	+/* local reservation */
	622	+static char *
	623	+reslocal(Gang *g)
	624	+{
	625	+ int count;
	626	+ char *err = nil;
	627	+
	628	+ DPRINT(DEXE, "reslocal: size=(%d)\n", g->size);
	629	+
	630	+ if(mystats.next == nil)
	631	+ mystats.njobs += g->size;
	632	+
	633	+ dprint_status(&mystats, "reslocal", 0);
	634	+
	635	+ g->sess = emalloc9p(sizeof(Session)*g->size);
	636	+ /* TODO: check return value? */
	637	+ for(count = 0; count < g->size; count++) {
	638	+ setupsess(g, &g->sess[count], nil, 0, count);
	639	+ DPRINT(DEXE, "reslocal: reservation: count=%d out of %d\n", count+1, g->size);
	640	+ proccreate(cloneproc, &g->sess[count], STACK);
	641	+ }
	642	+
	643	+ for(count = 0; count < g->size; count++) {
	644	+ char *myerr;
	645	+ DPRINT(DEXE, "\twaiting on reservation: count=%d out of %d\n", count+1, g->size);
	646	+ myerr = recvp(g->sess[count].chan);
	647	+ if(myerr != nil)
	648	+ err = myerr;
	649	+ /* MAYBE: retry here? for reliability ? */
	650	+ }
	651	+
	652	+ return err;
	653	+}
	654	+
	655	+/* multinode reservation */
	656	+static char *
	657	+resgang(Gang *g)
	658	+{
	659	+ int njobs; / array of children with jobs per child */
	660	+ int remaining;
	661	+ Status *current;
	662	+ int scount = 0;
	663	+ int count;
	664	+ char *err = nil;
	665	+ int size = g->size;
	666	+
	667	+ g->size=0;
	668	+
	669	+ /* MAYBE: Lock? */
	670	+ /* TODO: implement other modes (block, time-share) */
	671	+
	672	+ /* we need to know the current status of the job market... */
	673	+ refreshstatus(&mystats);
	674	+
	675	+ DPRINT(DEXE, "resgang: requesting size=%d out of %d=(mystats.ntask=%d) - (mystats.njobs=%d)\n",
	676	+ size, mystats.ntask-mystats.njobs, mystats.ntask, mystats.njobs);
	677	+ if(size > (mystats.ntask-mystats.njobs)) {
	678	+ DPRINT(DEXE, "insufficient resources for %d out of %d\n",
	679	+ size, mystats.ntask-mystats.njobs);
	680	+ return estrdup9p(Enores);
	681	+ }
	682	+
	683	+ /* Lock down gang */
	684	+ njobs = emalloc9p(sizeof(int)*mystats.nchild);
	685	+ remaining = size;
	686	+ /* walk children tree, allocating cores and incrementing stats.njobs
	687	+ counting sessions and keeping children per subsession in an array? */
	688	+ for(current=mystats.next; current !=nil; current = current->next) {
	689	+ dprint_status(current, "resgang (child stats)", 0);
	690	+
	691	+ int avail = current->ntask - current->njobs;
	692	+ if(avail >= remaining) {
	693	+ njobs[scount] = remaining;
	694	+ current->njobs += njobs[scount];
	695	+ break;
	696	+ } else {
	697	+ njobs[scount] = current->ntask - current->njobs;
	698	+ remaining -= njobs[scount];
	699	+ }
	700	+ current->njobs += njobs[scount];
	701	+ if(njobs[scount] > 0) /* only increment when we've allocated a subnode */
	702	+ scount++;
	703	+ }
	704	+ dprint_status(&mystats, "resgang (master stats)", 0);
	705	+
	706	+ scount++; /* count instead of index now */
	707	+
	708	+ /* allocate subsessions */
	709	+ g->sess = emalloc9p(sizeof(Session)*scount);
	710	+ g->size = scount;
	711	+
	712	+ DPRINT(DEXE, "Gang opening up %d sessions\n", scount);
	713	+
	714	+ /* while more to schedule */
	715	+ current = mystats.next;
	716	+ for(count = 0; count < scount; count++) {
	717	+ checkmount(current->name);
	718	+ // the child might already be busy, so just skip it for now
	719	+ //assert(njobs[count] != 0);
	720	+ if(njobs[count] == 0)
	721	+ continue;
	722	+ DPRINT(DEXE, "resgang: Session %d Target %s Njobs: %d From %s\n",
	723	+ count, current->name, njobs[count], mysysname);
	724	+ // EBo -- FIXME -- needs to chec for children or (or leaf)
	725	+ setupsess(g, &g->sess[count], current->name, njobs[count], count);
	726	+ ///// setupsess(g, &g->sess[count], current->name, 0, count);
	727	+ proccreate(cloneproc, &g->sess[count], STACK);
	728	+ current = current->next;
	729	+ }
	730	+
	731	+ DPRINT(DEXE, "waiting on %d sessions\n", scount);
	732	+
	733	+ for(count = 0; count < scount; count++) {
	734	+ char *myerr;
	735	+ DPRINT(DEXE, "\twaiting on reservation: count=%d out of %d\n", count+1, scount);
	736	+ assert(g->sess[count].chan != 0); /* or you are stupid */
	737	+ myerr = recvp(g->sess[count].chan);
	738	+
	739	+ if(myerr != nil) {
	740	+ DPRINT(DERR, "ERROR: session %d broke for (%s) err=(%s)\n", count, g->sess[count].path, myerr); // FIXME: g->path appears to always be nil try session info
	741	+ err = myerr;
	742	+ g->status = GANG_BROKE;
	743	+ }
	744	+ /* MAYBE: retry here? for reliability ? */
	745	+
	746	+ /*
	747	+ This is extremely fragile, and should be looked at
	748	+ to retry, fail over to another session, and other
	749	+ fault tollerance behaviour. There are a couple of
	750	+ possible scenerios here 1) the node is dead, 2)
	751	+ there is a race condition, ...
	752	+
	753	+ Plan:
	754	+ * disable node (not gang)
	755	+ * keep a node fail counter
	756	+ * allow possible restart via the status update or
	757	+ retry.
	758	+
	759	+ For now, just disable and move on.
	760	+
	761	+ FIXME: we need to examine/discuss what could be
	762	+ happening with a task whose operation changes the
	763	+ behaviour of the system and if it is possible that
	764	+ failing in the middle of an operation destabalizes
	765	+ the system.
	766	+
	767	+ */
	768	+ }
	769	+
	770	+ return err;
	771	+}
	772	+
	773	+/* TODO: many of the error scenarios here probably need more cleanup */
	774	+static void
	775	+cmdres(Req *r, int num)
	776	+{
	777	+ Fid *f = r->fid;
	778	+ Gang *g = f->aux;
	779	+
	780	+ char dest[255];
	781	+ char *mntargs;
	782	+ char *err;
	783	+
	784	+ /* okay - by this point we should have gang embedded */
	785	+ if(g == nil) {
	786	+ respond(r, Enogang);
	787	+ return;
	788	+ }
	789	+
	790	+ if(g->size > 0) {
	791	+ respond(r, Ebusy);
	792	+ return;
	793	+ }
	794	+
	795	+ /* setup aggregation I/O */
	796	+ snprint(dest, 255, "%s/g%d", gangpath, g->index);
	797	+
	798	+ switch(g->imode) {
	799	+ case CMbcast:
	800	+ DPRINT(DGAN, "cmdres: broadcast mode\n");
	801	+ if (mpipe(dest, "-b stdin") < 0) {
	802	+ DPRINT(DGAN, "setting up stdin pipe failed\n");
	803	+ respond(r, Estdin);
	804	+ return;
	805	+ }
	806	+ break;
	807	+ case CMenum:
	808	+ DPRINT(DGAN, "cmdres: enumerated mode\n");
	809	+ mntargs = smprint("-e %d stdin", num);
	810	+ if (mpipe(dest, mntargs)< 0) {
	811	+ free(mntargs);
	812	+ respond(r, Estdin);
	813	+ return;
	814	+ }
	815	+ free(mntargs);
	816	+ break;
	817	+ default:
	818	+ respond(r, Ebadimode);
	819	+ return;
	820	+ };
	821	+ if (mpipe(dest, "stdout") < 0) {
	822	+ respond(r, Estdout);
	823	+ return;
	824	+ }
	825	+ if (mpipe(dest, "stderr") < 0) {
	826	+ respond(r, Estderr);
	827	+ return;
	828	+ }
	829	+
	830	+ g->size = num;
	831	+
	832	+ /* this assumes top down reservation model */
	833	+ if(mystats.nchild)
	834	+ err = resgang(g);
	835	+ else
	836	+ err = reslocal(g);
	837	+
	838	+ if(err) {
	839	+ DPRINT(DERR, "ERROR: problem establishing gang reservation: %s\n", err);
	840	+ respond(r, err);
	841	+ /* TODO: we have to cleanup gang and sessions here */
	842	+
	843	+ free(err);
	844	+ } else {
	845	+ DPRINT(DEXE, "\t gangs all here\n");
	846	+ g->status = GANG_RESV;
	847	+ r->ofcall.count = r->ifcall.count;
	848	+ respond(r, nil);
	849	+ }
	850	+}
	851	+
	852	+static void
	853	+relaycmd(void *arg)
	854	+{
	855	+ Session *s = arg;
	856	+ int n;
	857	+
	858	+ DPRINT(DEXE, "\trelaycmd: writing %d count of data to %d\n", s->r->ifcall.count, s->fd);
	859	+ n = pwrite(s->fd, s->r->ifcall.data, s->r->ifcall.count, 0);
	860	+ if(n < 0) {
	861	+ sendp(s->chan, smprint("ERROR: relaycmd: write cmd failed: %r\n"));
	862	+ } else {
	863	+ sendp(s->chan, nil);
	864	+ }
	865	+}
	866	+
	867	+static void
	868	+cmdbcast(Req r, Gang g)
	869	+{
	870	+ int count;
	871	+ char *resp;
	872	+ char *err = nil;
	873	+
	874	+ char *cmd = estrdup9p(r->ifcall.data);
	875	+ if(cmd[r->ifcall.count-1]=='\n')
	876	+ cmd[r->ifcall.count-1] = 0;
	877	+ else
	878	+ cmd[r->ifcall.count] = 0;
	879	+
	880	+ DPRINT(DBCA, "cmdbcast: broadcasting (%s) to gang %d (size: %d)\n", cmd, g->index, g->size);
	881	+ free(cmd);
	882	+
	883	+ /* broadcast to all subsessions */
	884	+ for(count = 0; count < g->size; count++) {
	885	+ /* slightly slimy */
	886	+ g->sess[count].r = r;
	887	+ proccreate(relaycmd, &g->sess[count], STACK);
	888	+ DPRINT(DEXE, "\t**** cmdbcast[%d]: relaycmd created\n", count);
	889	+ }
	890	+
	891	+ DPRINT(DEXE, "\twaiting for responses\n");
	892	+ /* wait for responses, checking for errors */
	893	+ for(count = 0; count < g->size; count++) {
	894	+ DPRINT(DEXE, "\t\twaiting on relay[%d]\n", count);
	895	+ /* wait for response */
	896	+ resp = recvp(g->sess[count].chan);
	897	+ // FIXME: do we need to free anything up?
	898	+ DPRINT(DEXE, "\t\treturning err=(%s)\n", resp);
	899	+ g->sess[count].r = nil;
	900	+ if(resp)
	901	+ err = resp;
	902	+ }
	903	+
	904	+ if(err) {
	905	+ DPRINT(DEXE, "cmdbcast: returning err=(%s)\n", err);
	906	+ respond(r, err);
	907	+ } else {
	908	+ /* success */
	909	+ DPRINT(DEXE, "cmdbcast: returning success\n");
	910	+ r->ofcall.count = r->ifcall.count;
	911	+ respond(r, nil);
	912	+ }
	913	+}

sys/src/cmd/uem/gang2fs.c+8−1370

@@ -36,6 +36,7 @@ int remote_override = 0; · 35 unmodified lines
36	36	int remote = 0;
37	37	int ntask_override = 0;
38	38
	39	+
39	40	static char defaultpath[] = "/proc";
40	41	static char defaultsrvpath[] = "gangfs";
41	42	static char defaultsrvmpipe[] = "mpipe";
@@ -210,61 +211,7 @@ Cmdtab ctlcmdtab[]={ · 168 unmodified lines
210	211	CMmount, "checkmount", 2,
211	212	};
212	213
213		−#define MAXNUM 10 /* maximum number of numbers on data line */
214		−
215		−enum
216		−{
217		− /* old /dev/swap */
218		− Mem = 0,
219		− Maxmem,
220		− Swap,
221		− Maxswap,
222		−
223		− /* /dev/sysstats */
224		− Procno = 0,
225		− Context,
226		− Interrupt,
227		− Syscall,
228		− Fault,
229		− TLBfault,
230		− TLBpurge,
231		− Load,
232		− Idle,
233		− InIntr,
234		−};
235		−
236		−/* BUG: we'll need a lock to protect the linked list */
237		−typedef struct Status Status;
238		−struct Status
239		−{
240		− char name[16];
241		− long lastup; /* last time this was updated */
242		−
243		− int ntask; /* the number of tasks (on a remote
244		− gange, it is the sum of all its children) */
245		− int nchild; /* number of children */
246		− int njobs; /* numer of tasks being run */
247		−
248		− int statsfd;
249		− int swapfd;
250		−
251		− uvlong devswap[4];
252		− uvlong devsysstat[10];
253		−
254		− /* big enough to hold /dev/sysstat even with many processors */
255		− char buf[81024]; // FIXME: check to make sure this is still true /
256		− char *bufp;
257		− char *ebufp;
258		−
259		− Status next; / linked list of children */
260		−};
261		−
262		−static Status avgstats; /* composite statistics */
263		−static char canonpath; / canonical path */
264		−static Status mystats; /* my statistics */
265		−RWLock statslock; /* protect stat updates and additions */
266		−
267		−/* FIXME: debugging... */
	214	+/* FIXME: debugging... (keep it here? */
268	215	static void
269	216	lsproc(void *arg)
270	217	{
@@ -277,478 +224,6 @@ lsproc(void *arg) · 6 unmodified lines
277	224
278	225	}
279	226
280		−static void
281		−dprint_status(Status s, char who, int dev)
282		−{
283		− DPRINT(DSTS, "%s: name=(%s)\n", who, s->name);
284		− DPRINT(DSTS, "\tntask=(%d)\n", s->ntask);
285		− DPRINT(DSTS, "\tnchild=(%d)\n", s->nchild);
286		− DPRINT(DSTS, "\tnjobs=(%d)\n", s->njobs);
287		−
288		− if(dev){
289		− DPRINT(DSTS, "\tdevswap[Mem]=(%d)\n", s->devswap[Mem]);
290		− DPRINT(DSTS, "\tdevswap[Maxmem]=(%d)\n", s->devswap[Maxmem]);
291		− DPRINT(DSTS, "\tdevsysstat[Load]=(%d)\n", s->devsysstat[Load]);
292		− DPRINT(DSTS, "\tdevsysstat[Idle]=(%d)\n", s->devsysstat[Idle]);
293		− }
294		−}
295		−
296		−static int
297		−loadbuf(Status m, int fd)
298		−{
299		− int n;
300		−
301		− if(*fd < 0)
302		− return 0;
303		− seek(*fd, 0, 0);
304		− n = read(*fd, m->buf, sizeof m->buf-1);
305		− if(n <= 0){
306		− close(*fd);
307		− *fd = -1;
308		− return 0;
309		− }
310		− m->bufp = m->buf;
311		− m->ebufp = m->buf+n;
312		− m->buf[n] = 0;
313		− return 1;
314		−}
315		−
316		−/* read one line of text from buffer and process integers */
317		−static int
318		−readnums(Status m, int n, uvlong a, int spanlines)
319		−{
320		− int i;
321		− char p, ep;
322		−
323		− if(spanlines)
324		− ep = m->ebufp;
325		− else
326		− for(ep=m->bufp; ep<m->ebufp; ep++)
327		− if(*ep == '\n')
328		− break;
329		− p = m->bufp;
330		− for(i=0; i<n && p<ep; i++){
331		− while(p<ep && (!isascii(p) \|\| !isdigit(p)) && *p!='-')
332		− p++;
333		− if(p == ep)
334		− break;
335		− a[i] = strtoull(p, &p, 10);
336		− }
337		− if(ep < m->ebufp)
338		− ep++;
339		− m->bufp = ep;
340		− return i == n;
341		−}
342		−
343		−/* read one line of text from buffer and process integers */
344		−static int
345		−readints(char buf, int n, int a)
346		−{
347		− int i;
348		− char p, ep;
349		−
350		− p = buf;
351		− ep = buf+strlen(buf);
352		−
353		− for(i=0; i<n && p<ep; i++){
354		− while(p<ep && (!isascii(p) \|\| !isdigit(p)) && *p!='-')
355		− p++;
356		− if(p == ep)
357		− break;
358		− a[i] = strtol(p, &p, 10);
359		− }
360		− if(i != n) {
361		− DPRINT(DERR, "i=%d, n=%d\n", i, n);
362		− }
363		− return i == n;
364		−}
365		−
366		−
367		−int
368		−readnum(ulong off, char *buf, ulong n, ulong val, int size)
369		−{
370		− char tmp[64];
371		−
372		− snprint(tmp, sizeof(tmp), "%.lud", size-1, size-1, val);
373		− tmp[size-1] = ' ';
374		− if(off >= size)
375		− return 0;
376		− if(off+n > size)
377		− n = size-off;
378		− memmove(buf, tmp+off, n);
379		− return n;
380		−}
381		−
382		−static int
383		−readswap(Status m, uvlong a)
384		−{
385		− if(strstr(m->buf, "memory\n")){
386		− /* new /dev/swap - skip first 3 numbers */
387		− if(!readnums(m, 7, a, 1))
388		− return 0;
389		− a[0] = a[3];
390		− a[1] = a[4];
391		− a[2] = a[5];
392		− a[3] = a[6];
393		− return 1;
394		− }
395		− return readnums(m, nelem(m->devswap), a, 0);
396		−}
397		−
398		−static int
399		−refreshstatus(Status *m)
400		−{
401		− int n;
402		− int i;
403		− uvlong a[nelem(m->devsysstat)];
404		−
405		− wlock(&statslock);
406		− // FIXME: this might not work when we are dealing with gangs
407		− // of gangs because it is both a child and a master.
408		−
409		− if(m->next) { /* aggregation node */
410		− Status *c;
411		−
412		− /* go through all the children and add the important bits up */
413		− m->ntask = 0;
414		− m->nchild = 0;
415		− m->njobs = 0; /* MAYBE: do we really want to do this? */
416		− m->devswap[Mem] = 0;
417		− m->devswap[Maxmem] = 0;
418		− m->devsysstat[Load] = 0;
419		− m->devsysstat[Idle] = 0;
420		− c = m->next;
421		− while(c != nil) {
422		− dprint_status(c, "refreshstatus (child)", 0);
423		−
424		− m->ntask += c->ntask;
425		− /* if child is an aggregation node only count its children */
426		− if(c->nchild) {
427		− m->nchild += c->nchild;
428		− } else {
429		− m->nchild++;
430		− }
431		−
432		− m->njobs += c->njobs;
433		− m->devswap[Mem] += c->devswap[Mem];
434		− m->devswap[Maxmem] += c->devswap[Maxmem];
435		− m->devsysstat[Load] += c->devsysstat[Load];
436		− m->devsysstat[Idle] =+ c->devsysstat[Idle];
437		− c = c->next;
438		− }
439		−
440		− /* MAYBE: average load and idle when done? */
441		− } else { /* child node */
442		− if(loadbuf(m, &m->swapfd) && readswap(m, a))
443		− memmove(m->devswap, a, sizeof m->devswap);
444		−
445		− if(loadbuf(m, &m->statsfd)){
446		− memset(m->devsysstat, 0, sizeof m->devsysstat);
447		− for(n=0; n<m->ntask && readnums(m, nelem(m->devsysstat), a, 0); n++)
448		− for(i=0; i<nelem(m->devsysstat); i++)
449		− m->devsysstat[i] += a[i];
450		− }
451		− }
452		−
453		− m->lastup = time(0);
454		−
455		− wunlock(&statslock);
456		−
457		− // what's going on...
458		− dprint_status(m, "refreshstatus (master after agregarion)", 1);
459		−
460		− return 1;
461		−}
462		−
463		−static void
464		−fillstatbuf(Status m, char statbuf)
465		−{
466		− char *bp;
467		− ulong t;
468		−
469		− DPRINT(DSTS, "fillstatbuf:\n");
470		− bp = statbuf;
471		− t = time(0);
472		−
473		− if(t-m->lastup > updateinterval)
474		− refreshstatus(m);
475		−
476		− // what's going on...
477		− dprint_status(m, "fillstatbuf (after refres)", 1);
478		−
479		− rlock(&statslock);
480		− bp+=snprint(bp, 17, "%-15.15s ", mystats.name);
481		−
482		− readnum(0, bp, NUMSIZE, m->ntask, NUMSIZE);
483		− bp+=NUMSIZE;
484		− readnum(0, bp, NUMSIZE, m->nchild, NUMSIZE);
485		− bp+=NUMSIZE;
486		− readnum(0, bp, NUMSIZE, m->njobs, NUMSIZE);
487		− bp+=NUMSIZE;
488		− readnum(0, bp, NUMSIZE, m->devsysstat[Load], NUMSIZE);
489		− bp+=NUMSIZE;
490		− readnum(0, bp, NUMSIZE, m->devsysstat[Idle], NUMSIZE);
491		− bp+=NUMSIZE;
492		− readnum(0, bp, NUMSIZE, m->devswap[Mem], NUMSIZE);
493		− bp+=NUMSIZE;
494		− readnum(0, bp, NUMSIZE, m->devswap[Maxmem], NUMSIZE);
495		− runlock(&statslock);
496		−
497		− bp+=NUMSIZE;
498		− bp = ' '; / space at the end to help parse things */
499		− bp++;
500		− *bp = 0;
501		−}
502		−
503		−static Status *
504		−findchild(Status m, char name)
505		−{
506		− Status *current = m;
507		−
508		− rlock(&statslock);
509		− while(current != nil) {
510		− if(strcmp(name, current->name) == 0)
511		− break;
512		− current = current->next;
513		− }
514		− runlock(&statslock);
515		− return current;
516		−}
517		−
518		−/* find lowest loaded child for next workload */
519		−/* NOTE: right now we just base this on number of jobs */
520		−static Status *
521		−idlechild(Status *m)
522		−{
523		− Status *current = m->next;
524		− Status *lowest = current;
525		−
526		− rlock(&statslock);
527		− while(current != nil) {
528		− if(current->njobs < lowest->njobs)
529		− lowest = current;
530		− current = current->next;
531		− }
532		− runlock(&statslock);
533		−
534		− return lowest;
535		−}
536		−
537		−static char *
538		−statuswrite(char *buf)
539		−{
540		− char name[16];
541		− Status *m;
542		− int ret;
543		− char *bp;
544		− int a[7];
545		−
546		− /* grab name and match it to our children */
547		− strecpy(name, name+sizeof name, buf);
548		− /* trunk name to first space */
549		− bp = strchr(name, ' ');
550		− if(bp != 0)
551		− *bp = 0;
552		−
553		− DPRINT(DSTS, "statuswrite: name=(%s)\n", name);
554		−
555		− m = findchild(&mystats, name);
556		− if(m == nil) {
557		− DPRINT(DSTS, "\tcreating new status for (%s)\n", name);
558		− m = emalloc9p(sizeof(Status));
559		− // FIXME: this was moved from below to make more sense. Probably not the correct error...
560		− if(m == nil)
561		− return Estatfind;
562		− memset(m, 0, sizeof(Status));
563		− strncpy(m->name, name, 16);
564		− wlock(&statslock);
565		− m->next = mystats.next;
566		− mystats.next = m;
567		− mystats.nchild++;
568		− wunlock(&statslock);
569		− }
570		−
571		−
572		− ret = readints(buf+16, 7, a);
573		− if(ret == 0) {
574		− DPRINT(DERR, "statuswrite: readints returned %d\n", ret);
575		− return Estatparse;
576		− } else {
577		− m->ntask = a[0];
578		− m->nchild = a[1];
579		− m->njobs = a[2];
580		− m->devsysstat[Load] = a[3];
581		− m->devsysstat[Idle] = a[4];
582		− m->devswap[Mem] = a[5];
583		− m->devswap[Maxmem] = a[6];
584		− }
585		−
586		− m->lastup = time(0);
587		−/* MAYBE: pull timestamp from child and include it in data */
588		−/* FIXME: see if this causes problems with the BG runs
589		− if(xgangpath == procpath){
590		− xgangpath = smprint("%s/%s%s", amprefix, mysysname, procpath);
591		− DPRINT(DEXE, "statuswrite: reseting xgangpath=(%s)\n", xgangpath);
592		− }
593		− DPRINT(DSTS, "statuswrite: xgangpath=(%s)\n", xgangpath);
594		−*/
595		− return nil;
596		−}
597		−
598		−static void
599		−updatestatus(void *arg)
600		−{
601		− int n;
602		− char parent = (char ) arg;
603		− char statbuf = emalloc9p((NUMSIZE11+1) + 1);
604		− char *parentpath = smprint("%s/%s/proc/status", amprefix, parent);
605		−
606		− int parentfd = open(parentpath, OWRITE);
607		−
608		− DPRINT(DCUR, "updatestatus: parent=(%s) parentpath=(%s)\n", parent, parentpath);
609		− DPRINT(DCUR, "\tparentfd=(%d)\n", parentfd);
610		− if(parentfd < 0) {
611		− DPRINT(DERR, "ERROR: couldn't open parent %s on path %s\n", parent, parentpath);
612		− return;
613		− }
614		−
615		− /* TODO: set my priority low */
616		−
617		− for(;;) {
618		− fillstatbuf(&mystats, statbuf);
619		− n = fprint(parentfd, "%s\n", statbuf);
620		− if(n < 0)
621		− break;
622		− sleep(updateinterval*1000);
623		− }
624		−
625		− // FIXME: shouldn't parentfd be closed at the end?
626		− close(parentfd);
627		−
628		− free(parentpath);
629		− free(statbuf);
630		−}
631		−
632		−static void
633		−initstatus(Status *m)
634		−{
635		− int n;
636		− uvlong a[nelem(m->devsysstat)];
637		−
638		− DPRINT(DEXE, "initstatus:\n");
639		− memset(m, 0, sizeof(Status));
640		− strncpy(m->name, mysysname, 15);
641		− m->name[15] = 0;
642		− m->swapfd = open("/dev/swap", OREAD);
643		− DPRINT(DCUR, "\tm->swapfd=(%d)\n", m->swapfd);
644		− assert(m->swapfd > 0);
645		− m->statsfd = open("/dev/sysstat", OREAD);
646		− DPRINT(DCUR, "\tm->statsfd=(%d)\n", m->statsfd);
647		− assert(m->swapfd > 0);
648		− if(loadbuf(m, &m->statsfd)){
649		− for(n=0; readnums(m, nelem(m->devsysstat), a, 0); n++)
650		− ;
651		− if(ntask_override)
652		− m->ntask = ntask_override;
653		− else
654		− m->ntask = n;
655		− } else
656		− m->ntask = 1;
657		− m->lastup = time(0);
658		−
659		− dprint_status(m, "initstatus:", 0);
660		−}
661		−
662		−static void
663		−closestatus(Status *m)
664		−{
665		− close(m->statsfd);
666		− close(m->swapfd);
667		−}
668		−
669		−/* remote support */
670		−static void
671		−startimport(void *arg)
672		−{
673		− char addr = (char ) arg;
674		− char *name = smprint("%s/%s", amprefix, addr);
675		−
676		− threadsetname("gangfs-startimport");
677		−
678		− // FIXME: should arg be replaced with addr
679		− DPRINT(DCUR, "startimport: importing name=%s\n");
680		− procexecl(nil, "/bin/import", "import", "-A", arg, "/",
681		− name, nil);
682		−
683		− DPRINT(DERR, "ERROR: startimport: procexecl returned: %r\n");
684		− threadexits(nil);
685		−}
686		−
687		−/* checkmount: check to see if a remote system is already, if not mount it */
688		−static int
689		−checkmount(char *addr)
690		−{
691		− char *dest = estrdup9p(addr);
692		− int retries = 0;
693		− int err = 0;
694		− int fd;
695		− char *srvpt = smprint("/srv/%s", addr);
696		− char *srvtg = smprint("%s/%s", amprefix, addr);
697		− char *mtpt = smprint("%s/proc", srvtg);
698		− Dir *tmp = dirstat(mtpt);
699		−
700		− DPRINT(DCUR, "checkmount: checking mount point (%s)\n", srvtg);
701		−
702		− if(tmp != nil) {
703		− DPRINT(DEXE, "\tsrv already mounted on (%s)\n", mtpt);
704		− goto out;
705		− } else
706		−
707		− DPRINT(DFID, "\tmount point (%s) does not exist... mounting\n", srvtg);
708		− DPRINT(DCUR, "\tattempting to mount %s on %s\n", srvpt, srvtg);
709		− fd = open(srvpt, ORDWR);
710		− if(fd > 0) {
711		− DPRINT(DCUR, "\tfd=(%d)\n", fd);
712		− if(mount(fd, -1, srvtg, MREPL, "") < 0) {
713		− // FIXME: what should we do here?
714		− DPRINT(DERR, "\tERROR: srv mount failed: %r\n");
715		− close(fd);
716		− } else {
717		− DPRINT(DFID, "\tsrv mount succeeded");
718		− DPRINT(DFID, "\t\tMOUNT");
719		− close(fd);
720		− goto out;
721		− }
722		− }
723		−
724		− /* no dir, spawn off an import */
725		− DPRINT(DFID, "\tno dir, spawn off an import +++\n");
726		− proccreate(startimport, dest, STACK);
727		−
728		− /* we need to wait for this to be done */
729		− while((tmp = dirstat(mtpt)) == nil) {
730		− // start a mount with a timeout...
731		− // use wait....
732		− if(retries++ > 100) {
733		− DPRINT(DERR, "ERROR: checkmount: import not complete after 10 seconds, giving up\n");
734		− sleep(100);
735		− err = -1;
736		− break;
737		− }
738		− }
739		−
740		−out:
741		− if(srvpt)
742		− free(srvpt);
743		− if(srvtg)
744		− free(srvtg);
745		− if(tmp)
746		− free(tmp);
747		− free(dest);
748		− free(mtpt);
749		− return err;
750		−}
751		−
752	227	static void
753	228	usage(void)
754	229	{
@@ -757,356 +232,6 @@ usage(void) · 2 unmodified lines
757	232	exits("usage");
758	233	}
759	234
760		−static void
761		−gangref(Gang g, char where)
762		−{
763		− g->refcount++;
764		− DPRINT(DREF, "\t\tGang(%p).refcount = %d [%s]\n", g, g->refcount, where);
765		−}
766		−
767		−static void
768		−gangrefinc(Gang g, char where)
769		−{
770		− g->refcount++;
771		− DPRINT(DREF, "\t\tGang(%p).refcount++ = %d -> %d[%s]\n", g, g->refcount-1, g->refcount, where);
772		−}
773		−
774		−static void
775		−gangrefdec(Gang g, char where)
776		−{
777		− g->refcount--;
778		− // reality check
779		− if(g->refcount < 0) {
780		− DPRINT(DERR, "%s: WARNING gangrefdec is less than 0. Resetting to 0\n", where);
781		− g->refcount = 0;
782		− }
783		− DPRINT(DREF, "\t\tGang(%p).refcount-- = %d -> %d[%s]\n", g, g->refcount+1, g->refcount, where);
784		−}
785		−
786		−static int
787		−mpipe(char path, char name)
788		−{
789		− int fd, ret;
790		− // FIXME: is the cost/savings benifit of creating srvpt
791		− // each time necessary? Look at changing srvmpipe to
792		− // /srv/mpipe, etc. and use it.
793		− char *srvpt = smprint("/srv/%s", srvmpipe);
794		− fd = open(srvpt, ORDWR);
795		− if(fd<0) {
796		− DPRINT(DERR, "ERROR: couldn't open %s: %r\n", srvpt);
797		− free(srvpt);
798		− return -1;
799		− }
800		− free(srvpt);
801		−
802		− //wunlock(&glock);
803		− DPRINT(DFID, "mpipe: mounting name=(%s) on path=(%s) pid=(%d) fd=%d\n",
804		− name, path, getpid(), fd);
805		− DPRINT(DCUR, "\tfrom mysysname=(%s)\n", mysysname);
806		− ret = mount(fd, -1, path, MBEFORE, name);
807		− //wunlock(&glock);
808		− if(ret < 0) {
809		− DPRINT(DERR, "mount of multipipe failed: %r\n");
810		− } else
811		− DPRINT(DFID, "\t\tMOUNT");
812		−
813		− close(fd);
814		− return ret;
815		−}
816		−
817		−static int
818		−flushmp(char *src)
819		−{
820		− int fd = open(src, OWRITE);
821		− int n;
822		− char hdr[255];
823		− ulong tag = ~0; /* header byte is at offset ~0 */
824		− char pkttype='.';
825		−
826		− if(fd < 0) {
827		− DPRINT(DPIP, "flushmp: open failed: %r\n");
828		− return fd;
829		− }
830		−
831		− n = snprint(hdr, 255, "%c\n%lud\n%lud\n%s\n",
832		− pkttype, (ulong)0, (ulong)0, "");
833		− DPRINT(DCUR, "flushmp: src=(%s) pid=(%d)\n", src, getpid());
834		− DPRINT(DCUR, "\tfd=(%d)\n", fd);
835		−
836		− n = pwrite(fd, hdr, n, tag);
837		− close(fd);
838		− return n;
839		−}
840		−
841		−static int
842		−spliceto(char src, char dest)
843		−{
844		− int fd = open(src, OWRITE);
845		− int n;
846		− char hdr[255];
847		− ulong tag = ~0; /* header byte is at offset ~0 */
848		− char pkttype='>';
849		−
850		− if(fd < 0) {
851		− DPRINT(DPIP, "spliceto: open %s failed: %r\n", src);
852		− return fd;
853		− }
854		−
855		− DPRINT(DCUR, "spliceto: fd=(%d)\n", fd);
856		− n = snprint(hdr, 255, "%c\n%lud\n%lud\n%s\n", pkttype, (ulong)0, (ulong)0, dest);
857		− n = pwrite(fd, hdr, n, tag);
858		− close(fd);
859		− return n;
860		−}
861		−
862		−static int
863		−splicefrom(char dest, char src)
864		−{
865		− int fd = open(dest, OWRITE);
866		− int n;
867		− char hdr[255];
868		− ulong tag = ~0; /* header byte is at offset ~0 */
869		− char pkttype='<';
870		−
871		− if(fd < 0) {
872		− DPRINT(DERR, "ERROR: splicefrom: open %s failed: %r\n", dest);
873		− return fd;
874		− }
875		−
876		− DPRINT(DCUR, "splicefrom: fd=(%d)\n", fd);
877		− DPRINT(DEXE, "just so we know where we are splicing: [%s]\n", src);
878		−
879		− n = snprint(hdr, 255, "%c\n%lud\n%lud\n%s\n", pkttype, (ulong)0, (ulong)0, src);
880		− n = pwrite(fd, hdr, n, tag);
881		− close(fd);
882		− return n;
883		−}
884		−
885		−static Gang *
886		−newgang(void)
887		−{
888		− Gang *mygang = emalloc9p(sizeof(Gang));
889		−
890		− if(mygang == nil)
891		− return nil;
892		− memset(mygang, 0, sizeof(Gang));
893		− mygang->chan = chancreate(sizeof(void *), 0);
894		−
895		− wlock(&glock);
896		− if(glist == nil) {
897		− glist = mygang;
898		− mygang->index = 0;
899		− } else {
900		− int last = -1;
901		− Gang *current;
902		− for(current=glist; current->next != nil; current = current->next) {
903		− if(current->index != last+1) {
904		− break;
905		− }
906		− last++;
907		− }
908		− mygang->next = current->next;
909		− current->next = mygang;
910		− mygang->index = current->index+1;
911		− }
912		−
913		− mygang->ctlref = 1;
914		− mygang->refcount = 1;
915		− mygang->imode = CMbcast; /* broadcast mode default */
916		− mygang->status = GANG_NEW;
917		− wunlock(&glock);
918		−
919		− return mygang;
920		−}
921		−
922		−static Gang *
923		−findgang(int index)
924		−{
925		− Gang *current = nil;
926		−
927		− rlock(&glock);
928		− if(glist == nil) {
929		− goto out;
930		− } else {
931		− for(current=glist; current != nil; current = current->next) {
932		− if((current->index == index)&&(current->status != GANG_CLOSED)) {
933		− gangrefinc(current, "findgang");
934		− goto out;
935		− }
936		− }
937		− }
938		−out:
939		− runlock(&glock);
940		− return current;
941		−}
942		−
943		−static Gang *
944		−findgangnum(int which)
945		−{
946		− Gang *current = nil;
947		− rlock(&glock);
948		− if(glist == nil) {
949		− goto out;
950		− } else {
951		− int count = 0;
952		− for(current=glist; current != nil; current = current->next) {
953		− if(current->status == GANG_CLOSED)
954		− continue;
955		− if(count == which) {
956		− gangrefinc(current, "findgangnum");
957		− goto out;
958		− }
959		− count++;
960		− }
961		− }
962		−out:
963		− runlock(&glock);
964		− return current;
965		−}
966		−
967		−static void
968		−releasesessions(Gang *g)
969		−{
970		− int count;
971		−
972		− DPRINT(DEXE, "releasesessions:\n");
973		− for(count = 0; count < g->size; count++) {
974		− DPRINT(DEXE, "\tsess[%d].path=(%s)\n", count, g->sess[count].path);
975		− DPRINT(DEXE, "\tsess[%d].chan=(%d)\n", count, g->sess[count].chan);
976		− DPRINT(DEXE, "\tsess[%d].fd=(%d)\n", count, g->sess[count].fd);
977		− free(g->sess[count].path);
978		− chanfree(g->sess[count].chan);
979		− close(g->sess[count].fd);
980		− }
981		−
982		− free(g->sess);
983		− g->sess = nil;
984		−}
985		−
986		−static void
987		−flushgang(void *arg)
988		−{
989		− Gang *g = arg;
990		− char fname[255];
991		−
992		− DPRINT(DCLN, "flushgang: flushing stdin, stdout, and stderr\n");
993		− snprint(fname, 255, "%s/g%d/stdin", gangpath, g->index);
994		− flushmp(fname); /* flush pipe to be sure */
995		− snprint(fname, 255, "%s/g%d/stdout", gangpath, g->index);
996		− flushmp(fname);
997		− snprint(fname, 255, "%s/g%d/stderr", gangpath, g->index);
998		− flushmp(fname);
999		−}
1000		−
1001		−static void
1002		−printgang(Gang *g)
1003		−{
1004		− rlock(&glock);
1005		−
1006		− DPRINT(DCUR, "\tprintgang:\n");
1007		− DPRINT(DCUR, "\t\tpath=%s\n", g->path);
1008		− DPRINT(DCUR, "\t\trefcount=%d\n", g->refcount);
1009		− DPRINT(DCUR, "\t\tctlref=%d\n", g->ctlref);
1010		− DPRINT(DCUR, "\t\tstatus=%d\n", g->status);
1011		− DPRINT(DCUR, "\t\tindex=%d\n", g->index);
1012		− DPRINT(DCUR, "\t\tsize=%d\n", g->size);
1013		− DPRINT(DCUR, "\t\timode=%d\n", g->imode);
1014		− DPRINT(DCUR, "\t\t*** sessions not reported\n");
1015		−
1016		− runlock(&glock);
1017		−}
1018		−
1019		−static void
1020		−printganglist(void)
1021		−{
1022		− Gang *g;
1023		−
1024		− DPRINT(DCUR, "printganglist:\n");
1025		− for(g = glist; g != nil; g = g->next)
1026		− printgang(g);
1027		−}
1028		−
1029		−static void
1030		−cleanupgang(void *arg)
1031		−{
1032		− Gang *g = arg;
1033		− char fname[255];
1034		−
1035		− /* make sure that the std files are completely flushed */
1036		− flushgang(arg);
1037		−
1038		− /* unmount the std files and remove their references */
1039		− DPRINT(DCLN, "cleanupgang: unmounting stdin, stdout, and stderr\n");
1040		− snprint(fname, 255, "%s/g%d/stdin", gangpath, g->index);
1041		− if(unmount(0, fname) == -1)
1042		− DPRINT(DERR, "cleanupgang: ERROR unable to unmount %s: %r\n", fname);
1043		− gangrefdec(g, "cleanupgang");
1044		−
1045		− snprint(fname, 255, "%s/g%d/stdout", gangpath, g->index);
1046		− if(unmount(0, fname) == -1)
1047		− DPRINT(DERR, "cleanupgang: ERROR unable to unmount %s: %r\n", fname);
1048		− gangrefdec(g, "cleanupgang");
1049		−
1050		− snprint(fname, 255, "%s/g%d/stderr", gangpath, g->index);
1051		− if(unmount(0, fname) == -1)
1052		− DPRINT(DERR, "cleanupgang: ERROR unable to unmount %s: %r\n", fname);
1053		− gangrefdec(g, "cleanupgang");
1054		−
1055		− /* TODO: Clean up subsessions? */
1056		−
1057		− /* close it out */
1058		− g->status=GANG_CLOSED;
1059		−printgang(g);
1060		−}
1061		−
1062		−static void
1063		−releasegang(Gang *g)
1064		−{
1065		− wlock(&glock);
1066		− DPRINT(DEXE, "releasegang:\n");
1067		−// gangrefdec(g, "releasegang"); // new gang reserved it so we
1068		−// //need to decrement the ref counter here
1069		− if(g->refcount == 0) { /* clean up */
1070		− Gang *c;
1071		− Gang *p = glist;
1072		−
1073		− if(glist == g) {
1074		− glist = g->next;
1075		− } else {
1076		− for(c = glist; c != nil; c = c->next) {
1077		− if(c == g) {
1078		− p->next = g->next;
1079		− break;
1080		− }
1081		− p = c;
1082		− }
1083		− }
1084		−
1085		− releasesessions(g);
1086		− chanfree(g->chan);
1087		− free(g);
1088		− }
1089		−
1090		− wunlock(&glock);
1091		−}
1092		−
1093		−static void
1094		−cleanup(Srv *)
1095		−{
1096		− DPRINT(DCUR, "cleanup: pid=(%d)\n", getpid());
1097		− nbsendp(iochan, 0); /* flush any blocked readers */
1098		−
1099		− // FIXME: is this necessary? the iothread is being held open...
1100		− DPRINT(DFID, "cleanup: iothread's pid=(%d)\n", threadpid(iothread_id));
1101		− threadkill(iothread_id);
1102		−
1103		− chanclose(iochan);
1104		− closestatus(&mystats);
1105		− sleep(5);
1106		− DPRINT(DCUR, "cleanup: killing remaing threads\n");
1107		− threadexitsall("done");
1108		−}
1109		−
1110	235	static void
1111	236	fsattach(Req *r)
1112	237	{
@@ -1300,7 +425,9 @@ fsclone(Fid oldfid, Fid newfid) · 187 unmodified lines
1300	425	fprint(2, "newfid->aux is bit nil when oldfid->aux was not\n");
1301	426	return nil;
1302	427	}
	428	+
1303	429	/* this will only work if only sessions use aux */
	430	+ /* EVH: I don't think this is where we want this */
1304	431	g = oldfid->aux;
1305	432	if(g) {
1306	433	newfid->aux = oldfid->aux;
@@ -1404,425 +531,14 @@ fsread(Req *r) · 97 unmodified lines
1404	531	}
1405	532	}
1406	533
1407		−/* mypath is the canonical path to my gproc */
1408		−/* path is the path to the target proc */
1409		−static char *
1410		−setupstdio(Session *s)
1411		−{
1412		− Gang *g = s->g;
1413		− char buf[255];
1414		− char dest[255];
1415		− int n;
1416		−
1417		− DPRINT(DEXE, "setupstdio: path=(%s) remote=(%d)\n", s->path, s->remote);
1418		− DPRINT(DEXE, "\tgangpath=(%s) xgangpath=(%s)\n", gangpath, xgangpath);
1419		−
1420		− /* BUG: won't work for gang because gang pid is not an int */
1421		− if(s->remote)
1422		− snprint(buf, 255, "%s/g%d", s->path, s->pid);
1423		− else
1424		− snprint(buf, 255, "%s/%d", s->path, s->pid);
1425		−
1426		− snprint(dest, 255, "%s/g%d/%d", gangpath, g->index, s->index);
1427		− n = bind(buf, dest, MREPL);
1428		− DPRINT(DEXE, "\tbind buf=(%s) dest=(%s) ret=%d\n", buf, dest, n);
1429		− if(n < 0)
1430		− return smprint("couldn't bind %s %s: %r\n", buf, dest);
1431		−
1432		−// FIXME: the rest of the gangs are actually xgangs...
1433		− if(s->remote)
1434		− snprint(buf, 255, "%s/g%d/stdin", s->path, s->pid);
1435		− else
1436		− snprint(buf, 255, "%s/%d/stdin", s->path, s->pid);
1437		− snprint(dest, 255, "%s/g%d/stdin", xgangpath, g->index);
1438		− DPRINT(DEXE, "\tsplicefrom buf=(%s) dest=(%s)\n", buf, dest);
1439		− n = splicefrom(buf, dest); /* execfs initiates splicefrom gangfs */
1440		− DPRINT(DEXE, "\t\tret=%d\n", n);
1441		− if(n < 0)
1442		− return smprint("setupstdio: %r\n");
1443		−
1444		− if(s->remote)
1445		− snprint(buf, 255, "%s/g%d/stdout", s->path, s->pid);
1446		− else
1447		− snprint(buf, 255, "%s/%d/stdout", s->path, s->pid);
1448		− snprint(dest, 255, "%s/g%d/stdout", xgangpath, g->index);
1449		− DPRINT(DEXE, "\tspliceto buf=(%s) dest=(%s)\n", buf, dest);
1450		− n = spliceto(buf, dest); /* execfs initiates spliceto gangfs stdout */
1451		− DPRINT(DEXE, "\t\tret=%d\n", n);
1452		− if(n < 0)
1453		− return smprint("setupstdio: %r\n");
1454		−
1455		− if(s->remote)
1456		− snprint(buf, 255, "%s/g%d/stderr", s->path, s->pid);
1457		− else
1458		− snprint(buf, 255, "%s/%d/stderr", s->path, s->pid);
1459		− snprint(dest, 255, "%s/g%d/stderr", xgangpath, g->index);
1460		− DPRINT(DEXE, "\tspliceto buf=(%s) dest=(%s)\n", buf, dest);
1461		− n = spliceto(buf, dest); /* execfs initiates spliceto gangfs stderr */
1462		− DPRINT(DEXE, "\t\tret=%d\n", n);
1463		− if(n < 0)
1464		− return smprint("setupstdio: %r\n");
1465		−
1466		− return nil;
1467		−}
1468		−
1469		−/* reserve a session using clone semantics */
1470		−/* NOTE: Should work with execfs or gangfs */
1471		−static void
1472		−cloneproc(void *arg)
1473		−{
1474		− Session sess = (Session ) arg;
1475		− char buf[256];
1476		− int retries;
1477		− int n = -1;
1478		− char *err;
1479		−
1480		− if(sess->remote)
1481		− snprint(buf, 255, "%s/gclone", sess->path);
1482		− else
1483		− snprint(buf, 255, "%s/clone", sess->path);
1484		−
1485		− DPRINT(DEXE, "cloneproc: **** clone proc %s (remote=%d)\n",
1486		− buf, sess->remote);
1487		− for(retries=0; retries <= 10; retries++) {
1488		− sess->fd = open(buf, ORDWR);
1489		−
1490		− if(sess->fd <= 0) {
1491		− sleep (100);
1492		− DPRINT(DERR, "WARNING: retry: %d opening execfs returned %d: %r -- retrying\n",
1493		− retries, sess->fd);
1494		− continue;
1495		− }
1496		−
1497		− n = pread(sess->fd, buf, 255, 0);
1498		− if(n >= 0)
1499		− break;
1500		−
1501		− DPRINT(DERR, "WARNING: retry: %d execfs's pread=(%d) failed for fd=(%d): %r -- retrying\n",
1502		− retries, n, sess->fd);
1503		− close (sess->fd);
1504		− }
1505		− DPRINT(DCUR, "\tcontrol channel on sess->fd=(%d) for subsession (%d) pread=(%d)\n",
1506		− sess->fd, sess->subindex, n);
1507		−
1508		− if(retries > 10) {
1509		− DPRINT(DERR, "ERROR giving up\n");
1510		− sendp(sess->chan, smprint("execfs returning %r"));
1511		− threadexits(Ebadpid);
1512		− }
1513		−
1514		− if(n < 0) {
1515		− sendp(sess->chan,
1516		− smprint("couldn't read from session ctl path=%s/clone fd=(%d) n=(%d): %r\n",
1517		− sess->path, sess->fd, n));
1518		− threadexits(Ebadctl);
1519		− }
1520		− buf[n] = 0; // FIXME: need to terminate the buffer or we get junk...
1521		−
1522		− if(sess->remote)
1523		− sess->pid = atoi(buf+1); /* skip the g */
1524		− else
1525		− sess->pid = atoi(buf); /* convert to local execs session number */
1526		−
1527		− DPRINT(DEXE, "cloneproc: sess->pid=%ld buf=(%s)\n", sess->pid, buf);
1528		−
1529		− /* if we are a remote session, setup backmount */
1530		− if(sess->remote) {
1531		− DPRINT(DEXE, "Attempting to remote checkmount %s on %s\n",
1532		− mysysname, sess->path);
1533		− n = fprint(sess->fd, "checkmount %s", mysysname);
1534		− if(n < 0) {
1535		− DPRINT(DEXE, "setting up backmount failed from %s to %s: %r\n",
1536		− mysysname, sess->path);
1537		− chanprint(sess->chan, "setting up backmount failed from %s to %s: %r\n",
1538		− mysysname, sess->path);
1539		− goto error;
1540		− }
1541		−
1542		− if(sess->g->imode == CMenum) {
1543		− n = fprint(sess->fd, "enum");
1544		− if(n < 0) {
1545		− DPRINT(DEXE, "setting up enum failed on %s: %r\n", mysysname);
1546		− chanprint(sess->chan, "setting up enum failed on %s: %r\n",
1547		− mysysname);
1548		− goto error;
1549		− }
1550		− }
1551		−
1552		− n = fprint(sess->fd, "res %d", sess->remote);
1553		− if(n < 0) {
1554		− DPRINT(DEXE, "remote reservation failed: %r\n");
1555		− chanprint(sess->chan, "remote reservation failed: %r\n");
1556		− goto error;
1557		− }
1558		− }
1559		−
1560		− /* finish by setting up stdio */
1561		− if(err = setupstdio(sess)) {
1562		− DPRINT(DEXE, "reporting failure from stdio setup session %d:%d: %s\n",
1563		− sess->index, sess->subindex, err);
1564		− sendp(sess->chan, err);
1565		− } else {
1566		− DPRINT(DEXE, "reporting success from session %d:%d\n", sess->index, sess->subindex);
1567		− sendp(sess->chan, nil);
1568		− }
1569		−
1570		−error:
1571		− threadexits(nil);
1572		−}
1573		−
1574		−static void
1575		−setupsess(Gang g, Session s, char *path, int r, int sub)
1576		−{
1577		− if(path){
1578		− s->path = smprint("%s/%s/proc", amprefix, path);
1579		− } else {
1580		− s->path = smprint(execfspath);
1581		− }
1582		− DPRINT(DEXE, "setupsess: path=(%s) gangpath=(%s) execfspath=(%s) \n",
1583		− path, gangpath, execfspath);
1584		− DPRINT(DEXE, "\tresult session path=(%s)\n", s->path);
1585		−
1586		− s->r = nil;
1587		− s->chan = chancreate(sizeof(char *), 0);
1588		− s->g = g; /* back pointer */
1589		−
1590		− // remote_override does not work here as r = the number of sessions
1591		− s->remote = r;
1592		−
1593		− s->subindex = sub; /* keep the subsession index to help with debugging */
1594		−}
1595		−
1596		−/* local reservation */
1597		−static char *
1598		−reslocal(Gang *g)
1599		−{
1600		− int count;
1601		− char *err = nil;
1602		−
1603		− DPRINT(DEXE, "reslocal: size=(%d)\n", g->size);
1604		−
1605		− if(mystats.next == nil)
1606		− mystats.njobs += g->size;
1607		−
1608		− dprint_status(&mystats, "reslocal", 0);
1609		−
1610		− g->sess = emalloc9p(sizeof(Session)*g->size);
1611		− /* TODO: check return value? */
1612		− for(count = 0; count < g->size; count++) {
1613		− setupsess(g, &g->sess[count], nil, 0, count);
1614		− DPRINT(DEXE, "reslocal: reservation: count=%d out of %d\n", count+1, g->size);
1615		− proccreate(cloneproc, &g->sess[count], STACK);
1616		− }
1617		−
1618		− for(count = 0; count < g->size; count++) {
1619		− char *myerr;
1620		− DPRINT(DEXE, "\twaiting on reservation: count=%d out of %d\n", count+1, g->size);
1621		− myerr = recvp(g->sess[count].chan);
1622		− if(myerr != nil)
1623		− err = myerr;
1624		− /* MAYBE: retry here? for reliability ? */
1625		− }
1626		−
1627		− return err;
1628		−}
1629		−
1630		−/* multinode reservation */
1631		−static char *
1632		−resgang(Gang *g)
1633		−{
1634		− int njobs; / array of children with jobs per child */
1635		− int remaining;
1636		− Status *current;
1637		− int scount = 0;
1638		− int count;
1639		− char *err = nil;
1640		− int size = g->size;
1641		−
1642		− g->size=0;
1643		−
1644		− /* MAYBE: Lock? */
1645		− /* TODO: implement other modes (block, time-share) */
1646		−
1647		− /* we need to know the current status of the job market... */
1648		− refreshstatus(&mystats);
1649		−
1650		− DPRINT(DEXE, "resgang: requesting size=%d out of %d=(mystats.ntask=%d) - (mystats.njobs=%d)\n",
1651		− size, mystats.ntask-mystats.njobs, mystats.ntask, mystats.njobs);
1652		− if(size > (mystats.ntask-mystats.njobs)) {
1653		− DPRINT(DEXE, "insufficient resources for %d out of %d\n",
1654		− size, mystats.ntask-mystats.njobs);
1655		− return estrdup9p(Enores);
1656		− }
1657		−
1658		− /* Lock down gang */
1659		− njobs = emalloc9p(sizeof(int)*mystats.nchild);
1660		− remaining = size;
1661		− /* walk children tree, allocating cores and incrementing stats.njobs
1662		− counting sessions and keeping children per subsession in an array? */
1663		− for(current=mystats.next; current !=nil; current = current->next) {
1664		− dprint_status(current, "resgang (child stats)", 0);
1665		−
1666		− int avail = current->ntask - current->njobs;
1667		− if(avail >= remaining) {
1668		− njobs[scount] = remaining;
1669		− current->njobs += njobs[scount];
1670		− break;
1671		− } else {
1672		− njobs[scount] = current->ntask - current->njobs;
1673		− remaining -= njobs[scount];
1674		− }
1675		− current->njobs += njobs[scount];
1676		− if(njobs[scount] > 0) /* only increment when we've allocated a subnode */
1677		− scount++;
1678		− }
1679		− dprint_status(&mystats, "resgang (master stats)", 0);
1680		−
1681		− scount++; /* count instead of index now */
1682		−
1683		− /* allocate subsessions */
1684		− g->sess = emalloc9p(sizeof(Session)*scount);
1685		− g->size = scount;
1686		−
1687		− DPRINT(DEXE, "Gang opening up %d sessions\n", scount);
1688		−
1689		− /* while more to schedule */
1690		− current = mystats.next;
1691		− for(count = 0; count < scount; count++) {
1692		− checkmount(current->name);
1693		− // the child might already be busy, so just skip it for now
1694		− //assert(njobs[count] != 0);
1695		− if(njobs[count] == 0)
1696		− continue;
1697		− DPRINT(DEXE, "resgang: Session %d Target %s Njobs: %d From %s\n",
1698		− count, current->name, njobs[count], mysysname);
1699		− // EBo -- FIXME -- needs to chec for children or (or leaf)
1700		− setupsess(g, &g->sess[count], current->name, njobs[count], count);
1701		− ///// setupsess(g, &g->sess[count], current->name, 0, count);
1702		− proccreate(cloneproc, &g->sess[count], STACK);
1703		− current = current->next;
1704		− }
1705		−
1706		− DPRINT(DEXE, "waiting on %d sessions\n", scount);
1707		−
1708		− for(count = 0; count < scount; count++) {
1709		− char *myerr;
1710		− DPRINT(DEXE, "\twaiting on reservation: count=%d out of %d\n", count+1, scount);
1711		− assert(g->sess[count].chan != 0); /* or you are stupid */
1712		− myerr = recvp(g->sess[count].chan);
1713		−
1714		− if(myerr != nil) {
1715		− DPRINT(DERR, "ERROR: session %d broke for (%s) err=(%s)\n", count, g->sess[count].path, myerr); // FIXME: g->path appears to always be nil try session info
1716		− err = myerr;
1717		− g->status = GANG_BROKE;
1718		− }
1719		− /* MAYBE: retry here? for reliability ? */
1720		−
1721		− /*
1722		− This is extremely fragile, and should be looked at
1723		− to retry, fail over to another session, and other
1724		− fault tollerance behaviour. There are a couple of
1725		− possible scenerios here 1) the node is dead, 2)
1726		− there is a race condition, ...
1727		−
1728		− Plan:
1729		− * disable node (not gang)
1730		− * keep a node fail counter
1731		− * allow possible restart via the status update or
1732		− retry.
1733		−
1734		− For now, just disable and move on.
1735		−
1736		− FIXME: we need to examine/discuss what could be
1737		− happening with a task whose operation changes the
1738		− behaviour of the system and if it is possible that
1739		− failing in the middle of an operation destabalizes
1740		− the system.
1741		−
1742		− */
1743		− }
1744		−
1745		− return err;
1746		−}
1747		−
1748		−/* TODO: many of the error scenarios here probably need more cleanup */
1749		−static void
1750		−cmdres(Req *r, int num)
1751		−{
1752		− Fid *f = r->fid;
1753		− Gang *g = f->aux;
1754	534
1755		− char dest[255];
1756		− char *mntargs;
1757		− char *err;
1758	535
1759		− /* okay - by this point we should have gang embedded */
1760		− if(g == nil) {
1761		− respond(r, Enogang);
1762		− return;
1763		− }
1764	536
1765		− if(g->size > 0) {
1766		− respond(r, Ebusy);
1767		− return;
1768		− }
1769	537
1770		− /* setup aggregation I/O */
1771		− snprint(dest, 255, "%s/g%d", gangpath, g->index);
1772	538
1773		− switch(g->imode) {
1774		− case CMbcast:
1775		− DPRINT(DGAN, "cmdres: broadcast mode\n");
1776		− if (mpipe(dest, "-b stdin") < 0) {
1777		− DPRINT(DGAN, "setting up stdin pipe failed\n");
1778		− respond(r, Estdin);
1779		− return;
1780		− }
1781		− break;
1782		− case CMenum:
1783		− DPRINT(DGAN, "cmdres: enumerated mode\n");
1784		− mntargs = smprint("-e %d stdin", num);
1785		− if (mpipe(dest, mntargs)< 0) {
1786		− free(mntargs);
1787		− respond(r, Estdin);
1788		− return;
1789		− }
1790		− free(mntargs);
1791		− break;
1792		− default:
1793		− respond(r, Ebadimode);
1794		− return;
1795		− };
1796		− if (mpipe(dest, "stdout") < 0) {
1797		− respond(r, Estdout);
1798		− return;
1799		− }
1800		− if (mpipe(dest, "stderr") < 0) {
1801		− respond(r, Estderr);
1802		− return;
1803		− }
1804	539
1805		− g->size = num;
1806	540
1807		− /* this assumes top down reservation model */
1808		− if(mystats.nchild)
1809		− err = resgang(g);
1810		− else
1811		− err = reslocal(g);
1812	541
1813		− if(err) {
1814		− DPRINT(DERR, "ERROR: problem establishing gang reservation: %s\n", err);
1815		− respond(r, err);
1816		− /* TODO: we have to cleanup gang and sessions here */
1817		−
1818		− free(err);
1819		− } else {
1820		− DPRINT(DEXE, "\t gangs all here\n");
1821		− g->status = GANG_RESV;
1822		− r->ofcall.count = r->ifcall.count;
1823		− respond(r, nil);
1824		− }
1825		−}
1826	542
1827	543	/* this should probably be a generic function somewhere */
1828	544	static char *
@@ -1842,68 +558,9 @@ cleanupcb(char *buf, int count) · 13 unmodified lines
1842	558	return nbuf;
1843	559	}
1844	560
1845		−static void
1846		−relaycmd(void *arg)
1847		−{
1848		− Session *s = arg;
1849		− int n;
1850		−
1851		− DPRINT(DEXE, "\trelaycmd: writing %d count of data to %d\n", s->r->ifcall.count, s->fd);
1852		− n = pwrite(s->fd, s->r->ifcall.data, s->r->ifcall.count, 0);
1853		− if(n < 0) {
1854		− sendp(s->chan, smprint("ERROR: relaycmd: write cmd failed: %r\n"));
1855		− } else {
1856		− sendp(s->chan, nil);
1857		− }
1858		−}
1859		−
1860		−static void
1861		−cmdbcast(Req r, Gang g)
1862		−{
1863		− int count;
1864		− char *resp;
1865		− char *err = nil;
1866	561
1867		− char *cmd = estrdup9p(r->ifcall.data);
1868		− if(cmd[r->ifcall.count-1]=='\n')
1869		− cmd[r->ifcall.count-1] = 0;
1870		− else
1871		− cmd[r->ifcall.count] = 0;
1872	562
1873		− DPRINT(DBCA, "cmdbcast: broadcasting (%s) to gang %d (size: %d)\n", cmd, g->index, g->size);
1874		− free(cmd);
1875	563
1876		− /* broadcast to all subsessions */
1877		− for(count = 0; count < g->size; count++) {
1878		− /* slightly slimy */
1879		− g->sess[count].r = r;
1880		− proccreate(relaycmd, &g->sess[count], STACK);
1881		− DPRINT(DEXE, "\t**** cmdbcast[%d]: relaycmd created\n", count);
1882		− }
1883		−
1884		− DPRINT(DEXE, "\twaiting for responses\n");
1885		− /* wait for responses, checking for errors */
1886		− for(count = 0; count < g->size; count++) {
1887		− DPRINT(DEXE, "\t\twaiting on relay[%d]\n", count);
1888		− /* wait for response */
1889		− resp = recvp(g->sess[count].chan);
1890		− // FIXME: do we need to free anything up?
1891		− DPRINT(DEXE, "\t\treturning err=(%s)\n", resp);
1892		− g->sess[count].r = nil;
1893		− if(resp)
1894		− err = resp;
1895		− }
1896		−
1897		− if(err) {
1898		− DPRINT(DEXE, "cmdbcast: returning err=(%s)\n", err);
1899		− respond(r, err);
1900		− } else {
1901		− /* success */
1902		− DPRINT(DEXE, "cmdbcast: returning success\n");
1903		− r->ofcall.count = r->ifcall.count;
1904		− respond(r, nil);
1905		− }
1906		−}
1907	564
1908	565	static void
1909	566	fswrite(void *arg)
@@ -1966,28 +623,6 @@ fswrite(void *arg) · 56 unmodified lines
1966	623	} else {
1967	624	DPRINT(DEXE, "Unknown command, broadcasting to children\n");
1968	625	cmdbcast(r, g);
1969		−
1970		−//gangrefdec(g, "testing cmdbcast:");
1971		− // FIXME: on large numbers of tasks,
1972		− // some of the stdouts appear to be
1973		− // removed before they are fully
1974		− // processed.
1975		−
1976		− //flushgang(g);
1977		− //sleep(5000);
1978		−
1979		− // FIXME: I am not sure this is the
1980		− // most appropriate place to put this...
1981		− //DPRINT(DCUR, "******** releasegang\n");
1982		− //releasegang(g);
1983		−// DPRINT(DCUR, "******** cleanupgang\n");
1984		−// cleanupgang(g);
1985		−// DPRINT(DCUR, "******** done\n");
1986		−//DPRINT(DCUR, "hacking the refcount\n");
1987		−//gangrefdec(g, "cmdbcast");
1988		−//gangrefdec(g, "cmdbcast");
1989		−//gangrefdec(g, "cmdbcast");
1990		−//printganglist();
1991	626	}
1992	627	} else {
1993	628	switch(cmd->index) {
@@ -2066,8 +701,11 @@ fsclunk(Fid *fid) · 72 unmodified lines
2066	701	g->ctlref--;
2067	702	DPRINT(DREF, "\tctlref=%d\n", g->ctlref);
2068	703	if(g->ctlref == 0) {
2069		− if(mystats.next == nil)
	704	+ if(mystats.next == nil) { /* if we are on a child node */
	705	+ DPRINT(DFID, "* WARNING *: njobs not decrementing njobs=%d size=%d\n", mystats.njobs, g->size);
2070	706	mystats.njobs -= g->size;
	707	+ /* TODO: have to push this back? */
	708	+ }
2071	709	g->status=GANG_CLOSING;
2072	710	wunlock(&glock);
2073	711	DPRINT(DFID, "fsclunk: g->status=GANG_CLOSING for gang %p path=%s\n", g, g->path);

sys/src/cmd/uem/gangmon.c+444−0

@@ -0,0 +1,444 @@
	1	+#define MAXNUM 10 /* maximum number of numbers on data line */
	2	+
	3	+
	4	+enum
	5	+{
	6	+ /* old /dev/swap */
	7	+ Mem = 0,
	8	+ Maxmem,
	9	+ Swap,
	10	+ Maxswap,
	11	+
	12	+ /* /dev/sysstats */
	13	+ Procno = 0,
	14	+ Context,
	15	+ Interrupt,
	16	+ Syscall,
	17	+ Fault,
	18	+ TLBfault,
	19	+ TLBpurge,
	20	+ Load,
	21	+ Idle,
	22	+ InIntr,
	23	+};
	24	+
	25	+/* BUG: we'll need a lock to protect the linked list */
	26	+typedef struct Status Status;
	27	+struct Status
	28	+{
	29	+ char name[16];
	30	+ long lastup; /* last time this was updated */
	31	+
	32	+ int ntask; /* the number of tasks (on a remote
	33	+ gange, it is the sum of all its children) */
	34	+ int nchild; /* number of children */
	35	+ int njobs; /* numer of tasks being run */
	36	+
	37	+ int statsfd;
	38	+ int swapfd;
	39	+
	40	+ uvlong devswap[4];
	41	+ uvlong devsysstat[10];
	42	+
	43	+ /* big enough to hold /dev/sysstat even with many processors */
	44	+ char buf[81024]; // FIXME: check to make sure this is still true /
	45	+ char *bufp;
	46	+ char *ebufp;
	47	+
	48	+ Status next; / linked list of children */
	49	+};
	50	+
	51	+static Status avgstats; /* composite statistics */
	52	+static char canonpath; / canonical path */
	53	+static Status mystats; /* my statistics */
	54	+RWLock statslock; /* protect stat updates and additions */
	55	+
	56	+static void
	57	+dprint_status(Status s, char who, int dev)
	58	+{
	59	+ DPRINT(DSTS, "%s: name=(%s)\n", who, s->name);
	60	+ DPRINT(DSTS, "\tntask=(%d)\n", s->ntask);
	61	+ DPRINT(DSTS, "\tnchild=(%d)\n", s->nchild);
	62	+ DPRINT(DSTS, "\tnjobs=(%d)\n", s->njobs);
	63	+
	64	+ if(dev){
	65	+ DPRINT(DSTS, "\tdevswap[Mem]=(%d)\n", s->devswap[Mem]);
	66	+ DPRINT(DSTS, "\tdevswap[Maxmem]=(%d)\n", s->devswap[Maxmem]);
	67	+ DPRINT(DSTS, "\tdevsysstat[Load]=(%d)\n", s->devsysstat[Load]);
	68	+ DPRINT(DSTS, "\tdevsysstat[Idle]=(%d)\n", s->devsysstat[Idle]);
	69	+ }
	70	+}
	71	+
	72	+static int
	73	+loadbuf(Status m, int fd)
	74	+{
	75	+ int n;
	76	+
	77	+ if(*fd < 0)
	78	+ return 0;
	79	+ seek(*fd, 0, 0);
	80	+ n = read(*fd, m->buf, sizeof m->buf-1);
	81	+ if(n <= 0){
	82	+ close(*fd);
	83	+ *fd = -1;
	84	+ return 0;
	85	+ }
	86	+ m->bufp = m->buf;
	87	+ m->ebufp = m->buf+n;
	88	+ m->buf[n] = 0;
	89	+ return 1;
	90	+}
	91	+
	92	+/* read one line of text from buffer and process integers */
	93	+static int
	94	+readnums(Status m, int n, uvlong a, int spanlines)
	95	+{
	96	+ int i;
	97	+ char p, ep;
	98	+
	99	+ if(spanlines)
	100	+ ep = m->ebufp;
	101	+ else
	102	+ for(ep=m->bufp; ep<m->ebufp; ep++)
	103	+ if(*ep == '\n')
	104	+ break;
	105	+ p = m->bufp;
	106	+ for(i=0; i<n && p<ep; i++){
	107	+ while(p<ep && (!isascii(p) \|\| !isdigit(p)) && *p!='-')
	108	+ p++;
	109	+ if(p == ep)
	110	+ break;
	111	+ a[i] = strtoull(p, &p, 10);
	112	+ }
	113	+ if(ep < m->ebufp)
	114	+ ep++;
	115	+ m->bufp = ep;
	116	+ return i == n;
	117	+}
	118	+
	119	+/* read one line of text from buffer and process integers */
	120	+static int
	121	+readints(char buf, int n, int a)
	122	+{
	123	+ int i;
	124	+ char p, ep;
	125	+
	126	+ p = buf;
	127	+ ep = buf+strlen(buf);
	128	+
	129	+ for(i=0; i<n && p<ep; i++){
	130	+ while(p<ep && (!isascii(p) \|\| !isdigit(p)) && *p!='-')
	131	+ p++;
	132	+ if(p == ep)
	133	+ break;
	134	+ a[i] = strtol(p, &p, 10);
	135	+ }
	136	+ if(i != n) {
	137	+ DPRINT(DERR, "i=%d, n=%d\n", i, n);
	138	+ }
	139	+ return i == n;
	140	+}
	141	+
	142	+
	143	+int
	144	+readnum(ulong off, char *buf, ulong n, ulong val, int size)
	145	+{
	146	+ char tmp[64];
	147	+
	148	+ snprint(tmp, sizeof(tmp), "%.lud", size-1, size-1, val);
	149	+ tmp[size-1] = ' ';
	150	+ if(off >= size)
	151	+ return 0;
	152	+ if(off+n > size)
	153	+ n = size-off;
	154	+ memmove(buf, tmp+off, n);
	155	+ return n;
	156	+}
	157	+
	158	+static int
	159	+readswap(Status m, uvlong a)
	160	+{
	161	+ if(strstr(m->buf, "memory\n")){
	162	+ /* new /dev/swap - skip first 3 numbers */
	163	+ if(!readnums(m, 7, a, 1))
	164	+ return 0;
	165	+ a[0] = a[3];
	166	+ a[1] = a[4];
	167	+ a[2] = a[5];
	168	+ a[3] = a[6];
	169	+ return 1;
	170	+ }
	171	+ return readnums(m, nelem(m->devswap), a, 0);
	172	+}
	173	+
	174	+static int
	175	+refreshstatus(Status *m)
	176	+{
	177	+ int n;
	178	+ int i;
	179	+ uvlong a[nelem(m->devsysstat)];
	180	+
	181	+ wlock(&statslock);
	182	+ // FIXME: this might not work when we are dealing with gangs
	183	+ // of gangs because it is both a child and a master.
	184	+
	185	+ if(m->next) { /* aggregation node */
	186	+ Status *c;
	187	+
	188	+ /* go through all the children and add the important bits up */
	189	+ m->ntask = 0;
	190	+ m->nchild = 0;
	191	+ m->njobs = 0; /* MAYBE: do we really want to do this? */
	192	+ m->devswap[Mem] = 0;
	193	+ m->devswap[Maxmem] = 0;
	194	+ m->devsysstat[Load] = 0;
	195	+ m->devsysstat[Idle] = 0;
	196	+ c = m->next;
	197	+ while(c != nil) {
	198	+ dprint_status(c, "refreshstatus (child)", 0);
	199	+
	200	+ m->ntask += c->ntask;
	201	+ /* if child is an aggregation node only count its children */
	202	+ if(c->nchild) {
	203	+ m->nchild += c->nchild;
	204	+ } else {
	205	+ m->nchild++;
	206	+ }
	207	+
	208	+ m->njobs += c->njobs;
	209	+ m->devswap[Mem] += c->devswap[Mem];
	210	+ m->devswap[Maxmem] += c->devswap[Maxmem];
	211	+ m->devsysstat[Load] += c->devsysstat[Load];
	212	+ m->devsysstat[Idle] =+ c->devsysstat[Idle];
	213	+ c = c->next;
	214	+ }
	215	+
	216	+ /* MAYBE: average load and idle when done? */
	217	+ } else { /* child node */
	218	+ if(loadbuf(m, &m->swapfd) && readswap(m, a))
	219	+ memmove(m->devswap, a, sizeof m->devswap);
	220	+
	221	+ if(loadbuf(m, &m->statsfd)){
	222	+ memset(m->devsysstat, 0, sizeof m->devsysstat);
	223	+ for(n=0; n<m->ntask && readnums(m, nelem(m->devsysstat), a, 0); n++)
	224	+ for(i=0; i<nelem(m->devsysstat); i++)
	225	+ m->devsysstat[i] += a[i];
	226	+ }
	227	+ }
	228	+
	229	+ m->lastup = time(0);
	230	+
	231	+ wunlock(&statslock);
	232	+
	233	+ // what's going on...
	234	+ dprint_status(m, "refreshstatus (master after agregarion)", 1);
	235	+
	236	+ return 1;
	237	+}
	238	+
	239	+static void
	240	+fillstatbuf(Status m, char statbuf)
	241	+{
	242	+ char *bp;
	243	+ ulong t;
	244	+
	245	+ DPRINT(DSTS, "fillstatbuf:\n");
	246	+ bp = statbuf;
	247	+ t = time(0);
	248	+
	249	+ if(t-m->lastup > updateinterval)
	250	+ refreshstatus(m);
	251	+
	252	+ // what's going on...
	253	+ dprint_status(m, "fillstatbuf (after refres)", 1);
	254	+
	255	+ rlock(&statslock);
	256	+ bp+=snprint(bp, 17, "%-15.15s ", mystats.name);
	257	+
	258	+ readnum(0, bp, NUMSIZE, m->ntask, NUMSIZE);
	259	+ bp+=NUMSIZE;
	260	+ readnum(0, bp, NUMSIZE, m->nchild, NUMSIZE);
	261	+ bp+=NUMSIZE;
	262	+ readnum(0, bp, NUMSIZE, m->njobs, NUMSIZE);
	263	+ bp+=NUMSIZE;
	264	+ readnum(0, bp, NUMSIZE, m->devsysstat[Load], NUMSIZE);
	265	+ bp+=NUMSIZE;
	266	+ readnum(0, bp, NUMSIZE, m->devsysstat[Idle], NUMSIZE);
	267	+ bp+=NUMSIZE;
	268	+ readnum(0, bp, NUMSIZE, m->devswap[Mem], NUMSIZE);
	269	+ bp+=NUMSIZE;
	270	+ readnum(0, bp, NUMSIZE, m->devswap[Maxmem], NUMSIZE);
	271	+ runlock(&statslock);
	272	+
	273	+ bp+=NUMSIZE;
	274	+ bp = ' '; / space at the end to help parse things */
	275	+ bp++;
	276	+ *bp = 0;
	277	+}
	278	+
	279	+static Status *
	280	+findchild(Status m, char name)
	281	+{
	282	+ Status *current = m;
	283	+
	284	+ rlock(&statslock);
	285	+ while(current != nil) {
	286	+ if(strcmp(name, current->name) == 0)
	287	+ break;
	288	+ current = current->next;
	289	+ }
	290	+ runlock(&statslock);
	291	+ return current;
	292	+}
	293	+
	294	+/* find lowest loaded child for next workload */
	295	+/* NOTE: right now we just base this on number of jobs */
	296	+static Status *
	297	+idlechild(Status *m)
	298	+{
	299	+ Status *current = m->next;
	300	+ Status *lowest = current;
	301	+
	302	+ rlock(&statslock);
	303	+ while(current != nil) {
	304	+ if(current->njobs < lowest->njobs)
	305	+ lowest = current;
	306	+ current = current->next;
	307	+ }
	308	+ runlock(&statslock);
	309	+
	310	+ return lowest;
	311	+}
	312	+
	313	+static char *
	314	+statuswrite(char *buf)
	315	+{
	316	+ char name[16];
	317	+ Status *m;
	318	+ int ret;
	319	+ char *bp;
	320	+ int a[7];
	321	+
	322	+ /* grab name and match it to our children */
	323	+ strecpy(name, name+sizeof name, buf);
	324	+ /* trunk name to first space */
	325	+ bp = strchr(name, ' ');
	326	+ if(bp != 0)
	327	+ *bp = 0;
	328	+
	329	+ DPRINT(DSTS, "statuswrite: name=(%s)\n", name);
	330	+
	331	+ m = findchild(&mystats, name);
	332	+ if(m == nil) {
	333	+ DPRINT(DSTS, "\tcreating new status for (%s)\n", name);
	334	+ m = emalloc9p(sizeof(Status));
	335	+ // FIXME: this was moved from below to make more sense. Probably not the correct error...
	336	+ if(m == nil)
	337	+ return Estatfind;
	338	+ memset(m, 0, sizeof(Status));
	339	+ strncpy(m->name, name, 16);
	340	+ wlock(&statslock);
	341	+ m->next = mystats.next;
	342	+ mystats.next = m;
	343	+ mystats.nchild++;
	344	+ wunlock(&statslock);
	345	+ }
	346	+
	347	+
	348	+ ret = readints(buf+16, 7, a);
	349	+ if(ret == 0) {
	350	+ DPRINT(DERR, "statuswrite: readints returned %d\n", ret);
	351	+ return Estatparse;
	352	+ } else {
	353	+ m->ntask = a[0];
	354	+ m->nchild = a[1];
	355	+ m->njobs = a[2];
	356	+ m->devsysstat[Load] = a[3];
	357	+ m->devsysstat[Idle] = a[4];
	358	+ m->devswap[Mem] = a[5];
	359	+ m->devswap[Maxmem] = a[6];
	360	+ }
	361	+
	362	+ m->lastup = time(0);
	363	+/* MAYBE: pull timestamp from child and include it in data */
	364	+/* FIXME: see if this causes problems with the BG runs
	365	+ if(xgangpath == procpath){
	366	+ xgangpath = smprint("%s/%s%s", amprefix, mysysname, procpath);
	367	+ DPRINT(DEXE, "statuswrite: reseting xgangpath=(%s)\n", xgangpath);
	368	+ }
	369	+ DPRINT(DSTS, "statuswrite: xgangpath=(%s)\n", xgangpath);
	370	+*/
	371	+ return nil;
	372	+}
	373	+
	374	+static void
	375	+updatestatus(void *arg)
	376	+{
	377	+ int n;
	378	+ char parent = (char ) arg;
	379	+ char statbuf = emalloc9p((NUMSIZE11+1) + 1);
	380	+ char *parentpath = smprint("%s/%s/proc/status", amprefix, parent);
	381	+
	382	+ int parentfd = open(parentpath, OWRITE);
	383	+
	384	+ DPRINT(DCUR, "updatestatus: parent=(%s) parentpath=(%s)\n", parent, parentpath);
	385	+ DPRINT(DCUR, "\tparentfd=(%d)\n", parentfd);
	386	+ if(parentfd < 0) {
	387	+ DPRINT(DERR, "ERROR: couldn't open parent %s on path %s\n", parent, parentpath);
	388	+ return;
	389	+ }
	390	+
	391	+ /* TODO: set my priority low */
	392	+
	393	+ for(;;) {
	394	+ fillstatbuf(&mystats, statbuf);
	395	+ n = fprint(parentfd, "%s\n", statbuf);
	396	+ if(n < 0)
	397	+ break;
	398	+ DPRINT(DCUR, "updatestatus: parent=(%s) parentpath=(%s)\n", parent, parentpath);
	399	+ sleep(updateinterval*1000);
	400	+ }
	401	+
	402	+ // FIXME: shouldn't parentfd be closed at the end?
	403	+ close(parentfd);
	404	+
	405	+ free(parentpath);
	406	+ free(statbuf);
	407	+}
	408	+
	409	+static void
	410	+initstatus(Status *m)
	411	+{
	412	+ int n;
	413	+ uvlong a[nelem(m->devsysstat)];
	414	+
	415	+ DPRINT(DEXE, "initstatus:\n");
	416	+ memset(m, 0, sizeof(Status));
	417	+ strncpy(m->name, mysysname, 15);
	418	+ m->name[15] = 0;
	419	+ m->swapfd = open("/dev/swap", OREAD);
	420	+ DPRINT(DCUR, "\tm->swapfd=(%d)\n", m->swapfd);
	421	+ assert(m->swapfd > 0);
	422	+ m->statsfd = open("/dev/sysstat", OREAD);
	423	+ DPRINT(DCUR, "\tm->statsfd=(%d)\n", m->statsfd);
	424	+ assert(m->swapfd > 0);
	425	+ if(loadbuf(m, &m->statsfd)){
	426	+ for(n=0; readnums(m, nelem(m->devsysstat), a, 0); n++)
	427	+ ;
	428	+ if(ntask_override)
	429	+ m->ntask = ntask_override;
	430	+ else
	431	+ m->ntask = n;
	432	+ } else
	433	+ m->ntask = 1;
	434	+ m->lastup = time(0);
	435	+
	436	+ dprint_status(m, "initstatus:", 0);
	437	+}
	438	+
	439	+static void
	440	+closestatus(Status *m)
	441	+{
	442	+ close(m->statsfd);
	443	+ close(m->swapfd);
	444	+}