op public repos

Blob

Date:: Mon Nov 21 20:42:52 2005 UTC
Message:: 9P2000.u fixes (Lucho Ionkov)
Actions:: History | Blame | Raw File
1 #include <u.h>
2 #include <libc.h>
3 #include <fcall.h>
4 #include <thread.h>
5 #include <errno.h>
6 
7 enum
8 {
9 	STACK = 32768,
10 	NHASH = 31,
11 	MAXMSG = 64,	/* per connection */
12 };
13 
14 typedef struct Hash Hash;
15 typedef struct Fid Fid;
16 typedef struct Msg Msg;
17 typedef struct Conn Conn;
18 typedef struct Queue Queue;
19 
20 struct Hash
21 {
22 	Hash *next;
23 	uint n;
24 	void *v;
25 };
26 
27 struct Fid
28 {
29 	int fid;
30 	int ref;
31 	int cfid;
32 	int openfd;
33 	int isdir;
34 	Fid *next;
35 };
36 
37 struct Msg
38 {
39 	Conn *c;
40 	int internal;
41 	int ref;
42 	int ctag;
43 	int tag;
44 	int isopenfd;
45 	Fcall tx;
46 	Fcall rx;
47 	Fid *fid;
48 	Fid *newfid;
49 	Fid *afid;
50 	Msg *oldm;
51 	Msg *next;
52 	uchar *tpkt;
53 	uchar *rpkt;
54 };
55 
56 struct Conn
57 {
58 	int fd;
59 	int fdmode;
60 	Fid *fdfid;
61 	int nmsg;
62 	int nfid;
63 	Channel *inc;
64 	Channel *internal;
65 	int inputstalled;
66 	char dir[40];
67 	Hash *tag[NHASH];
68 	Hash *fid[NHASH];
69 	Queue *outq;
70 	Queue *inq;
71 	int dotu;
72 };
73 
74 char *xaname;
75 char *addr;
76 int afd;
77 char adir[40];
78 int isunix;
79 Queue *outq;
80 Queue *inq;
81 int verbose = 0;
82 int logging = 0;
83 int msize = 8192;
84 u32int xafid = NOFID;
85 int attached;
86 int versioned;
87 int dotu;
88 
89 void *gethash(Hash**, uint);
90 int puthash(Hash**, uint, void*);
91 int delhash(Hash**, uint, void*);
92 Msg *mread9p(Ioproc*, int, int);
93 int mwrite9p(Ioproc*, int, uchar*);
94 uchar *read9ppkt(Ioproc*, int);
95 int write9ppkt(int, uchar*);
96 Msg *msgnew(int);
97 void msgput(Msg*);
98 void msgclear(Msg*);
99 Msg *msgget(int);
100 void msgincref(Msg*);
101 Fid *fidnew(int);
102 void fidput(Fid*);
103 void *emalloc(int);
104 void *erealloc(void*, int);
105 Queue *qalloc(void);
106 int sendq(Queue*, void*);
107 void *recvq(Queue*);
108 void connthread(void*);
109 void connoutthread(void*);
110 void listenthread(void*);
111 void outputthread(void*);
112 void inputthread(void*);
113 void rewritehdr(Fcall*, uchar*);
114 void repack(Fcall*, uchar**, int);
115 int tlisten(char*, char*);
116 int taccept(int, char*);
117 int iolisten(Ioproc*, char*, char*);
118 int ioaccept(Ioproc*, int, char*);
119 int iorecvfd(Ioproc*, int);
120 int iosendfd(Ioproc*, int, int);
121 void mainproc(void*);
122 int ignorepipe(void*, char*);
123 int timefmt(Fmt*);
124 void dorootstat(void);
125 int stripudirread(Msg*);
126 int cvtustat(Fcall*, uchar**, int);
127 
128 void
129 usage(void)
130 {
131 	fprint(2, "usage: 9pserve [-lv] [-A aname afid] [-M msize] address\n");
132 	fprint(2, "\treads/writes 9P messages on stdin/stdout\n");
133 	threadexitsall("usage");
134 }
135 
136 uchar vbuf[128];
137 extern int _threaddebuglevel;
138 void
139 threadmain(int argc, char **argv)
140 {
141 	char *file, *x;
142 	int fd;
143 
144 	x = getenv("verbose9pserve");
145 	if(x){
146 		verbose = atoi(x);
147 		fprint(2, "verbose9pserve %s => %d\n", x, verbose);
148 	}
149 	ARGBEGIN{
150 	default:
151 		usage();
152 	case 'A':
153 		attached = 1;
154 		xaname = EARGF(usage());
155 		xafid = atoi(EARGF(usage()));
156 		break;
157 	case 'M':
158 		versioned = 1;
159 		msize = atoi(EARGF(usage()));
160 		break;
161 	case 'v':
162 		verbose++;
163 		break;
164 	case 'u':
165 		isunix++;
166 		break;
167 	case 'l':
168 		logging++;
169 		break;
170 	}ARGEND
171 	
172 	if(attached && !versioned){
173 		fprint(2, "-A must be used with -M\n");
174 		usage();
175 	}
176 
177 	if(argc != 1)
178 		usage();
179 	addr = argv[0];
180 
181 	fmtinstall('T', timefmt);
182 
183 	if((afd = announce(addr, adir)) < 0)
184 		sysfatal("announce %s: %r", addr);
185 	if(logging){
186 		if(strncmp(addr, "unix!", 5) == 0)
187 			addr += 5;
188 		file = smprint("%s.log", addr);
189 		if(file == nil)
190 			sysfatal("smprint log: %r");
191 		if((fd = create(file, OWRITE, 0666)) < 0)
192 			sysfatal("create %s: %r", file);
193 		dup(fd, 2);
194 		if(fd > 2)
195 			close(fd);
196 	}
197 	if(verbose) fprint(2, "%T 9pserve running\n");
198 	proccreate(mainproc, nil, STACK);
199 }
200 
201 void
202 mainproc(void *v)
203 {
204 	int n, nn;
205 	Fcall f;
206 	USED(v);
207 
208 	atnotify(ignorepipe, 1);
209 	fmtinstall('D', dirfmt);
210 	fmtinstall('M', dirmodefmt);
211 	fmtinstall('F', fcallfmt);
212 	fmtinstall('H', encodefmt);
213 
214 	outq = qalloc();
215 	inq = qalloc();
216 
217 	if(!versioned){
218 		f.type = Tversion;
219 		f.version = "9P2000.u";
220 		f.msize = msize;
221 		f.tag = NOTAG;
222 		n = convS2M(&f, vbuf, sizeof vbuf);
223 		if(n <= BIT16SZ)
224 			sysfatal("convS2M conversion error");
225 		if(verbose > 1) fprint(2, "%T * <- %F\n", &f);
226 		nn = write(1, vbuf, n);
227 		if(n != nn)
228 			sysfatal("error writing Tversion: %r\n");
229 		n = read9pmsg(0, vbuf, sizeof vbuf);
230 		if(convM2S(vbuf, n, &f) != n)
231 			sysfatal("convM2S failure");
232 		if(f.msize < msize)
233 			msize = f.msize;
234 		if(verbose > 1) fprint(2, "%T * -> %F\n", &f);
235 		dotu = strncmp(f.version, "9P2000.u", 8) == 0;
236 	}
237 
238 	threadcreate(inputthread, nil, STACK);
239 	threadcreate(outputthread, nil, STACK);
240 
241 //	if(rootfid)
242 //		dorootstat();
243 	
244 	threadcreate(listenthread, nil, STACK);
245 	threadexits(0);
246 }
247 
248 int
249 ignorepipe(void *v, char *s)
250 {
251 	USED(v);
252 	if(strcmp(s, "sys: write on closed pipe") == 0)
253 		return 1;
254 	if(strcmp(s, "sys: tstp") == 0)
255 		return 1;
256 	fprint(2, "9pserve %s: %T note: %s\n", addr, s);
257 	return 0;
258 }
259 
260 void
261 listenthread(void *arg)
262 {
263 	Conn *c;
264 	Ioproc *io;
265 
266 	io = ioproc();
267 	USED(arg);
268 	threadsetname("listen %s", adir);
269 	for(;;){
270 		c = emalloc(sizeof(Conn));
271 		c->fd = iolisten(io, adir, c->dir);
272 		if(c->fd < 0){
273 			if(verbose) fprint(2, "%T listen: %r\n");
274 			close(afd);
275 			free(c);
276 			return;
277 		}
278 		c->inc = chancreate(sizeof(void*), 0);
279 		c->internal = chancreate(sizeof(void*), 0);
280 		c->inq = qalloc();
281 		c->outq = qalloc();
282 		if(verbose) fprint(2, "%T incoming call on %s\n", c->dir);
283 		threadcreate(connthread, c, STACK);
284 	}	
285 }
286 
287 void
288 send9pmsg(Msg *m)
289 {
290 	int n, nn;
291 
292 	n = sizeS2Mu(&m->rx, m->c->dotu);
293 	m->rpkt = emalloc(n);
294 	nn = convS2Mu(&m->rx, m->rpkt, n, m->c->dotu);
295 	if(nn <= BIT16SZ)
296 		sysfatal("convS2Mu conversion error");
297 	if(nn != n)
298 		sysfatal("sizeS2Mu and convS2Mu disagree");
299 	sendq(m->c->outq, m);
300 }
301 
302 void
303 sendomsg(Msg *m)
304 {
305 	int n, nn;
306 
307 	n = sizeS2Mu(&m->tx, m->c->dotu);
308 	m->tpkt = emalloc(n);
309 	nn = convS2Mu(&m->tx, m->tpkt, n, m->c->dotu);
310 	if(nn <= BIT16SZ)
311 		sysfatal("convS2Mu conversion error");
312 	if(nn != n)
313 		sysfatal("sizeS2Mu and convS2Mu disagree");
314 	sendq(outq, m);
315 }
316 
317 void
318 err(Msg *m, char *ename)
319 {
320 	m->rx.type = Rerror;
321 	m->rx.ename = ename;
322 	m->rx.tag = m->tx.tag;
323 	send9pmsg(m);
324 }
325 
326 char*
327 estrdup(char *s)
328 {
329 	char *t;
330 	
331 	t = emalloc(strlen(s)+1);
332 	strcpy(t, s);
333 	return t;
334 }
335 
336 void
337 connthread(void *arg)
338 {
339 	int i, fd;
340 	Conn *c;
341 	Hash *h, *hnext;
342 	Msg *m, *om, *mm;
343 	Fid *f;
344 	Ioproc *io;
345 
346 	c = arg;
347 	threadsetname("conn %s", c->dir);
348 	io = ioproc();
349 	fd = ioaccept(io, c->fd, c->dir);
350 	if(fd < 0){
351 		if(verbose) fprint(2, "%T accept %s: %r\n", c->dir);
352 		goto out;
353 	}
354 	close(c->fd);
355 	c->fd = fd;
356 	threadcreate(connoutthread, c, STACK);
357 	while((m = mread9p(io, c->fd, c->dotu)) != nil){
358 		if(verbose > 1) fprint(2, "%T fd#%d -> %F\n", c->fd, &m->tx);
359 		m->c = c;
360 		m->ctag = m->tx.tag;
361 		c->nmsg++;
362 		if(verbose > 1) fprint(2, "%T fd#%d: new msg %p\n", c->fd, m);
363 		if(puthash(c->tag, m->tx.tag, m) < 0){
364 			err(m, "duplicate tag");
365 			continue;
366 		}
367 		msgincref(m);
368 		switch(m->tx.type){
369 		case Tversion:
370 			m->rx.tag = m->tx.tag;
371 			m->rx.msize = m->tx.msize;
372 			if(m->rx.msize > msize)
373 				m->rx.msize = msize;
374 			m->rx.version = "9P2000";
375 			c->dotu = 0;
376 			if(dotu && strncmp(m->tx.version, "9P2000.u", 8) == 0){
377 				m->rx.version = "9P2000.u";
378 				c->dotu = 1;
379 			}
380 			m->rx.type = Rversion;
381 			send9pmsg(m);
382 			continue;
383 		case Tflush:
384 			if((m->oldm = gethash(c->tag, m->tx.oldtag)) == nil){
385 				m->rx.tag = m->tx.tag;
386 				m->rx.type = Rflush;
387 				send9pmsg(m);
388 				continue;
389 			}
390 			msgincref(m->oldm);
391 			break;
392 		case Tattach:
393 			m->afid = nil;
394 			if(m->tx.afid != NOFID
395 			&& (m->afid = gethash(c->fid, m->tx.afid)) == nil){
396 				err(m, "unknown fid");
397 				continue;
398 			}
399 			if(m->afid)
400 				m->afid->ref++;
401 			m->fid = fidnew(m->tx.fid);
402 			if(puthash(c->fid, m->tx.fid, m->fid) < 0){
403 				err(m, "duplicate fid");
404 				continue;
405 			}
406 			m->fid->ref++;
407 			if(attached && m->afid==nil){
408 				if(m->tx.aname[0] && strcmp(xaname, m->tx.aname) != 0){
409 					err(m, "invalid attach name");
410 					continue;
411 				}
412 				m->tx.afid = xafid;
413 				m->tx.aname = xaname;
414 				m->tx.uname = estrdup(m->tx.uname);
415 				repack(&m->tx, &m->tpkt, c->dotu);
416 				free(m->tx.uname);
417 				m->tx.uname = "XXX";
418 			}
419 			break;
420 		case Twalk:
421 			if((m->fid = gethash(c->fid, m->tx.fid)) == nil){
422 				err(m, "unknown fid");
423 				continue;
424 			}
425 			m->fid->ref++;
426 			if(m->tx.newfid == m->tx.fid){
427 				m->fid->ref++;
428 				m->newfid = m->fid;
429 			}else{
430 				m->newfid = fidnew(m->tx.newfid);
431 				if(puthash(c->fid, m->tx.newfid, m->newfid) < 0){
432 					err(m, "duplicate fid");
433 					continue;
434 				}
435 				m->newfid->ref++;
436 			}
437 			break;
438 		case Tauth:
439 			if(attached){
440 				err(m, "authentication not required");
441 				continue;
442 			}
443 			m->afid = fidnew(m->tx.afid);
444 			if(puthash(c->fid, m->tx.afid, m->afid) < 0){
445 				err(m, "duplicate fid");
446 				continue;
447 			}
448 			m->afid->ref++;
449 			break;
450 		case Topenfd:
451 			if(m->tx.mode&~(OTRUNC|3)){
452 				err(m, "bad openfd mode");
453 				continue;
454 			}
455 			m->isopenfd = 1;
456 			m->tx.type = Topen;
457 			m->tpkt[4] = Topen;
458 			/* fall through */
459 		case Tcreate:
460 		case Topen:
461 		case Tclunk:
462 		case Tread:
463 		case Twrite:
464 		case Tremove:
465 		case Tstat:
466 		case Twstat:
467 			if((m->fid = gethash(c->fid, m->tx.fid)) == nil){
468 				err(m, "unknown fid");
469 				continue;
470 			}
471 			m->fid->ref++;
472 			if(m->tx.type==Twstat && dotu && !c->dotu){
473 				if(cvtustat(&m->tx, &m->tpkt, 1) < 0){
474 					err(m, "cannot convert stat buffer");
475 					continue;
476 				}
477 			}
478 			break;
479 		}
480 
481 		/* have everything - translate and send */
482 		m->c = c;
483 		m->ctag = m->tx.tag;
484 		m->tx.tag = m->tag;
485 		if(m->fid)
486 			m->tx.fid = m->fid->fid;
487 		if(m->newfid)
488 			m->tx.newfid = m->newfid->fid;
489 		if(m->afid)
490 			m->tx.afid = m->afid->fid;
491 		if(m->oldm)
492 			m->tx.oldtag = m->oldm->tag;
493 		/* reference passes to outq */
494 		sendq(outq, m);
495 		while(c->nmsg >= MAXMSG){
496 			c->inputstalled = 1;
497 			recvp(c->inc);
498 		}
499 	}
500 
501 	if(verbose) fprint(2, "%T fd#%d eof; flushing conn\n", c->fd);
502 
503 	/* flush the output queue */
504 	sendq(c->outq, nil);
505 	while(c->outq != nil)
506 		yield();
507 
508 	/* flush all outstanding messages */
509 	for(i=0; i<NHASH; i++){
510 		for(h=c->tag[i]; h; h=hnext){
511 			om = h->v;
512 			m = msgnew(0);
513 			m->internal = 1;
514 			m->c = c;
515 			c->nmsg++;
516 			m->tx.type = Tflush;
517 			m->tx.tag = m->tag;
518 			m->tx.oldtag = om->tag;
519 			m->oldm = om;
520 			msgincref(om);
521 			msgincref(m);	/* for outq */
522 			sendomsg(m);
523 			mm = recvp(c->internal);
524 			assert(mm == m);
525 			msgput(m);	/* got from recvp */
526 			msgput(m);	/* got from msgnew */
527 			msgput(om);	/* got from hash table */
528 			hnext = h->next;
529 			free(h);
530 		}
531 	}
532 
533 	/* clunk all outstanding fids */
534 	for(i=0; i<NHASH; i++){
535 		for(h=c->fid[i]; h; h=hnext){
536 			f = h->v;
537 			m = msgnew(0);
538 			m->internal = 1;
539 			m->c = c;
540 			c->nmsg++;
541 			m->tx.type = Tclunk;
542 			m->tx.tag = m->tag;
543 			m->tx.fid = f->fid;
544 			m->fid = f;
545 			f->ref++;
546 			msgincref(m);
547 			sendomsg(m);
548 			mm = recvp(c->internal);
549 			assert(mm == m);
550 			msgclear(m);
551 			msgput(m);	/* got from recvp */
552 			msgput(m);	/* got from msgnew */
553 			fidput(f);	/* got from hash table */
554 			hnext = h->next;
555 			free(h);
556 		}
557 	}
558 
559 out:
560 	closeioproc(io);
561 	assert(c->nmsg == 0);
562 	assert(c->nfid == 0);
563 	close(c->fd);
564 	chanfree(c->internal);
565 	c->internal = 0;
566 	chanfree(c->inc);
567 	c->inc = 0;
568 	free(c->inq);
569 	c->inq = 0;
570 	free(c);
571 }
572 
573 static void
574 openfdthread(void *v)
575 {
576 	Conn *c;
577 	Fid *fid;
578 	Msg *m;
579 	int n;
580 	vlong tot;
581 	Ioproc *io;
582 	char buf[1024];
583 
584 	c = v;
585 	fid = c->fdfid;
586 	io = ioproc();
587 	threadsetname("openfd %s", c->fdfid);
588 	tot = 0;
589 	m = nil;
590 	if(c->fdmode == OREAD){
591 		for(;;){
592 			if(verbose) fprint(2, "%T tread...");
593 			m = msgnew(0);
594 			m->internal = 1;
595 			m->c = c;
596 			m->tx.type = Tread;
597 			m->tx.count = msize - IOHDRSZ;
598 			m->tx.fid = fid->fid;
599 			m->tx.tag = m->tag;
600 			m->tx.offset = tot;
601 			m->fid = fid;
602 			fid->ref++;
603 			msgincref(m);
604 			sendomsg(m);
605 			recvp(c->internal);
606 			if(m->rx.type == Rerror){
607 			//	fprint(2, "%T read error: %s\n", m->rx.ename);
608 				break;
609 			}
610 			if(m->rx.count == 0)
611 				break;
612 			tot += m->rx.count;
613 			if(iowrite(io, c->fd, m->rx.data, m->rx.count) != m->rx.count){
614 				// fprint(2, "%T pipe write error: %r\n");
615 				break;
616 			}
617 			msgput(m);
618 			msgput(m);
619 			m = nil;
620 		}
621 	}else{
622 		for(;;){
623 			if(verbose) fprint(2, "%T twrite...");
624 			n = sizeof buf;
625 			if(n > msize)
626 				n = msize;
627 			if((n=ioread(io, c->fd, buf, n)) <= 0){
628 				if(n < 0)
629 					fprint(2, "%T pipe read error: %r\n");
630 				break;
631 			}
632 			m = msgnew(0);
633 			m->internal = 1;
634 			m->c = c;
635 			m->tx.type = Twrite;
636 			m->tx.fid = fid->fid;
637 			m->tx.data = buf;
638 			m->tx.count = n;
639 			m->tx.tag = m->tag;
640 			m->tx.offset = tot;
641 			m->fid = fid;
642 			fid->ref++;
643 			msgincref(m);
644 			sendomsg(m);
645 			recvp(c->internal);
646 			if(m->rx.type == Rerror){
647 			//	fprint(2, "%T write error: %s\n", m->rx.ename);
648 			}
649 			tot += n;
650 			msgput(m);
651 			msgput(m);
652 			m = nil;
653 		}
654 	}
655 	if(verbose) fprint(2, "%T eof on %d fid %d\n", c->fd, fid->fid);
656 	close(c->fd);
657 	closeioproc(io);
658 	if(m){
659 		msgput(m);
660 		msgput(m);
661 	}
662 	if(verbose) fprint(2, "%T eof on %d fid %d ref %d\n", c->fd, fid->fid, fid->ref);
663 	if(--fid->openfd == 0){
664 		m = msgnew(0);
665 		m->internal = 1;
666 		m->c = c;
667 		m->tx.type = Tclunk;
668 		m->tx.tag = m->tag;
669 		m->tx.fid = fid->fid;
670 		m->fid = fid;
671 		fid->ref++;
672 		msgincref(m);
673 		sendomsg(m);
674 		recvp(c->internal);
675 		msgput(m);
676 		msgput(m);
677 	}
678 	fidput(fid);
679 	c->fdfid = nil;
680 	chanfree(c->internal);
681 	c->internal = 0;
682 	free(c);
683 }			
684 
685 int
686 xopenfd(Msg *m)
687 {
688 	char errs[ERRMAX];
689 	int n, p[2];
690 	Conn *nc;
691 
692 	if(pipe(p) < 0){
693 		rerrstr(errs, sizeof errs);
694 		err(m, errs);
695 		/* XXX return here? */
696 	}
697 	if(verbose) fprint(2, "%T xopen pipe %d %d...", p[0], p[1]);
698 
699 	/* now we're committed. */
700 
701 	/* a new connection for this fid */
702 	nc = emalloc(sizeof(Conn));
703 	nc->internal = chancreate(sizeof(void*), 0);
704 
705 	/* a ref for us */
706 	nc->fdfid = m->fid;
707 	m->fid->ref++;
708 	nc->fdfid->openfd++;
709 	nc->fdmode = m->tx.mode;
710 	nc->fd = p[0];
711 
712 	/* a thread to tend the pipe */
713 	threadcreate(openfdthread, nc, STACK);
714 
715 	/* if mode is ORDWR, that openfdthread will write; start a reader */
716 	if((m->tx.mode&3) == ORDWR){
717 		nc = emalloc(sizeof(Conn));
718 		nc->internal = chancreate(sizeof(void*), 0);
719 		nc->fdfid = m->fid;
720 		m->fid->ref++;
721 		nc->fdfid->openfd++;
722 		nc->fdmode = OREAD;
723 		nc->fd = dup(p[0], -1);
724 		threadcreate(openfdthread, nc, STACK);
725 	}
726 
727 	/* steal fid from other connection */
728 	if(delhash(m->c->fid, m->fid->cfid, m->fid) == 0)
729 		fidput(m->fid);
730 
731 	/* rewrite as Ropenfd */
732 	m->rx.type = Ropenfd;
733 	n = GBIT32(m->rpkt);
734 	m->rpkt = erealloc(m->rpkt, n+4);
735 	PBIT32(m->rpkt+n, p[1]);
736 	n += 4;
737 	PBIT32(m->rpkt, n);
738 	m->rpkt[4] = Ropenfd;
739 	m->rx.unixfd = p[1];
740 	return 0;
741 }
742 
743 void
744 connoutthread(void *arg)
745 {
746 	char *ename;
747 	int err;
748 	Conn *c;
749 	Queue *outq;
750 	Msg *m, *om;
751 	Ioproc *io;
752 
753 	c = arg;
754 	outq = c->outq;
755 	io = ioproc();
756 	threadsetname("connout %s", c->dir);
757 	while((m = recvq(outq)) != nil){
758 		err = m->tx.type+1 != m->rx.type;
759 		if(!err && m->isopenfd)
760 			if(xopenfd(m) < 0)
761 				continue;
762 		switch(m->tx.type){
763 		case Tflush:
764 			om = m->oldm;
765 			if(om)
766 				if(delhash(om->c->tag, om->ctag, om) == 0)
767 					msgput(om);
768 			break;
769 		case Tclunk:
770 		case Tremove:
771 			if(m->fid)
772 				if(delhash(m->c->fid, m->fid->cfid, m->fid) == 0)
773 					fidput(m->fid);
774 			break;
775 		case Tauth:
776 			if(err && m->afid){
777 				if(verbose) fprint(2, "%T auth error\n");
778 				if(delhash(m->c->fid, m->afid->cfid, m->afid) == 0)
779 					fidput(m->afid);
780 			}
781 			break;
782 		case Tattach:
783 			if(err && m->fid)
784 				if(delhash(m->c->fid, m->fid->cfid, m->fid) == 0)
785 					fidput(m->fid);
786 			break;
787 		case Twalk:
788 			if(err || m->rx.nwqid < m->tx.nwname)
789 			if(m->tx.fid != m->tx.newfid && m->newfid)
790 				if(delhash(m->c->fid, m->newfid->cfid, m->newfid) == 0)
791 					fidput(m->newfid);
792 			break;
793 		case Tread:
794 			if(!err && m->fid->isdir && dotu && !m->c->dotu)
795 				stripudirread(m);
796 			break;
797 		case Tstat:
798 			if(!err && dotu && !m->c->dotu)
799 				cvtustat(&m->rx, &m->rpkt, 0);
800 			break;
801 		case Topen:
802 		case Tcreate:
803 			m->fid->isdir = (m->rx.qid.type & QTDIR);
804 			break;
805 		}
806 		if(m->rx.type==Rerror && dotu && !c->dotu){
807 			ename = estrdup(m->rx.ename);
808 			m->rx.ename = ename;
809 			repack(&m->rx, &m->rpkt, c->dotu);
810 			free(ename);
811 			m->rx.ename = "XXX";
812 		}
813 		if(delhash(m->c->tag, m->ctag, m) == 0)
814 			msgput(m);
815 		if(verbose > 1) fprint(2, "%T fd#%d <- %F\n", c->fd, &m->rx);
816 		rewritehdr(&m->rx, m->rpkt);
817 		if(mwrite9p(io, c->fd, m->rpkt) < 0)
818 			if(verbose) fprint(2, "%T write error: %r\n");
819 		msgput(m);
820 		if(c->inputstalled && c->nmsg < MAXMSG)
821 			nbsendp(c->inc, 0);
822 	}
823 	closeioproc(io);
824 	free(outq);
825 	c->outq = nil;
826 }
827 
828 void
829 outputthread(void *arg)
830 {
831 	Msg *m;
832 	Ioproc *io;
833 
834 	USED(arg);
835 	io = ioproc();
836 	threadsetname("output");
837 	while((m = recvq(outq)) != nil){
838 		if(verbose > 1) fprint(2, "%T * <- %F\n", &m->tx);
839 		rewritehdr(&m->tx, m->tpkt);
840 		if(mwrite9p(io, 1, m->tpkt) < 0)
841 			sysfatal("output error: %r");
842 		msgput(m);
843 	}
844 	closeioproc(io);
845 	fprint(2, "%T output eof\n");
846 	threadexitsall(0);
847 }	
848 
849 void
850 inputthread(void *arg)
851 {
852 	uchar *pkt;
853 	int n, nn, tag;
854 	Msg *m;
855 	Ioproc *io;
856 
857 	threadsetname("input");
858 	if(verbose) fprint(2, "%T input thread\n");
859 	io = ioproc();
860 	USED(arg);
861 	while((pkt = read9ppkt(io, 0)) != nil){
862 		n = GBIT32(pkt);
863 		if(n < 7){
864 			fprint(2, "%T short 9P packet from server\n");
865 			free(pkt);
866 			continue;
867 		}
868 		if(verbose > 2) fprint(2, "%T read %.*H\n", n, pkt);
869 		tag = GBIT16(pkt+5);
870 		if((m = msgget(tag)) == nil){
871 			fprint(2, "%T unexpected 9P response tag %d\n", tag);
872 			free(pkt);
873 			continue;
874 		}
875 		if((nn = convM2Su(pkt, n, &m->rx, dotu)) != n){
876 			fprint(2, "%T bad packet - convM2S %d but %d\n", nn, n);
877 			free(pkt);
878 			msgput(m);
879 			continue;
880 		}
881 		if(verbose > 1) fprint(2, "%T * -> %F%s\n", &m->rx,
882 			m->internal ? " (internal)" : "");
883 		m->rpkt = pkt;
884 		m->rx.tag = m->ctag;
885 		if(m->internal)
886 			sendp(m->c->internal, m);
887 		else if(m->c->outq)
888 			sendq(m->c->outq, m);
889 		else
890 			msgput(m);
891 	}
892 	closeioproc(io);
893 	//fprint(2, "%T input eof\n");
894 	threadexitsall(0);
895 }
896 
897 void*
898 gethash(Hash **ht, uint n)
899 {
900 	Hash *h;
901 
902 	for(h=ht[n%NHASH]; h; h=h->next)
903 		if(h->n == n)
904 			return h->v;
905 	return nil;
906 }
907 
908 int
909 delhash(Hash **ht, uint n, void *v)
910 {
911 	Hash *h, **l;
912 
913 	for(l=&ht[n%NHASH]; h=*l; l=&h->next)
914 		if(h->n == n){
915 			if(h->v != v){
916 				if(verbose) fprint(2, "%T delhash %d got %p want %p\n", n, h->v, v);
917 				return -1;
918 			}
919 			*l = h->next;
920 			free(h);
921 			return 0;
922 		}
923 	return -1;
924 }
925 
926 int
927 puthash(Hash **ht, uint n, void *v)
928 {
929 	Hash *h;
930 
931 	if(gethash(ht, n))
932 		return -1;
933 	h = emalloc(sizeof(Hash));
934 	h->next = ht[n%NHASH];
935 	h->n = n;
936 	h->v = v;
937 	ht[n%NHASH] = h;
938 	return 0;
939 }
940 
941 Fid **fidtab;
942 int nfidtab;
943 Fid *freefid;
944 
945 Fid*
946 fidnew(int cfid)
947 {
948 	Fid *f;
949 
950 	if(freefid == nil){
951 		fidtab = erealloc(fidtab, (nfidtab+1)*sizeof(fidtab[0]));
952 		if(nfidtab == xafid){
953 			fidtab[nfidtab++] = nil;
954 			fidtab = erealloc(fidtab, (nfidtab+1)*sizeof(fidtab[0]));
955 		}
956 		fidtab[nfidtab] = emalloc(sizeof(Fid));
957 		freefid = fidtab[nfidtab];
958 		freefid->fid = nfidtab++;
959 	}
960 	f = freefid;
961 	freefid = f->next;
962 	f->cfid = cfid;
963 	f->ref = 1;
964 	f->isdir = -1;
965 	return f;
966 }
967 
968 void
969 fidput(Fid *f)
970 {
971 	if(f == nil)
972 		return;
973 	assert(f->ref > 0);
974 	if(--f->ref > 0)
975 		return;
976 	f->next = freefid;
977 	f->cfid = -1;
978 	freefid = f;
979 }
980 
981 Msg **msgtab;
982 int nmsgtab;
983 int nmsg;
984 Msg *freemsg;
985 
986 void
987 msgincref(Msg *m)
988 {
989 	if(verbose > 1) fprint(2, "%T msgincref @0x%lux %p tag %d/%d ref %d=>%d\n",
990 		getcallerpc(&m), m, m->tag, m->ctag, m->ref, m->ref+1);
991 	m->ref++;
992 }
993 
994 Msg*
995 msgnew(int x)
996 {
997 	Msg *m;
998 
999 	if(freemsg == nil){
1000 		msgtab = erealloc(msgtab, (nmsgtab+1)*sizeof(msgtab[0]));
1001 		msgtab[nmsgtab] = emalloc(sizeof(Msg));
1002 		freemsg = msgtab[nmsgtab];
1003 		freemsg->tag = nmsgtab++;
1004 	}
1005 	m = freemsg;
1006 	freemsg = m->next;
1007 	m->ref = 1;
1008 	if(verbose > 1) fprint(2, "%T msgnew @0x%lux %p tag %d ref %d\n",
1009 		getcallerpc(&x), m, m->tag, m->ref);
1010 	nmsg++;
1011 	return m;
1012 }
1013 
1014 /*
1015  * Clear data associated with connections, so that
1016  * if all msgs have been msgcleared, the connection
1017  * can be freed.  Note that this does *not* free the tpkt
1018  * and rpkt; they are freed in msgput with the msg itself. 
1019  * The io write thread might still be holding a ref to msg
1020  * even once the connection has finished with it.
1021  */
1022 void
1023 msgclear(Msg *m)
1024 {
1025 	if(m->c){
1026 		m->c->nmsg--;
1027 		m->c = nil;
1028 	}
1029 	if(m->oldm){
1030 		msgput(m->oldm);
1031 		m->oldm = nil;
1032 	}
1033 	if(m->fid){
1034 		fidput(m->fid);
1035 		m->fid = nil;
1036 	}
1037 	if(m->afid){
1038 		fidput(m->afid);
1039 		m->afid = nil;
1040 	}
1041 	if(m->newfid){
1042 		fidput(m->newfid);
1043 		m->newfid = nil;
1044 	}
1045 	if(m->rx.type == Ropenfd && m->rx.unixfd >= 0){
1046 		close(m->rx.unixfd);
1047 		m->rx.unixfd = -1;
1048 	}
1049 }
1050 
1051 void
1052 msgput(Msg *m)
1053 {
1054 	if(m == nil)
1055 		return;
1056 
1057 	if(verbose > 1) fprint(2, "%T msgput 0x%lux %p tag %d/%d ref %d\n", 
1058 		getcallerpc(&m), m, m->tag, m->ctag, m->ref);
1059 	assert(m->ref > 0);
1060 	if(--m->ref > 0)
1061 		return;
1062 	nmsg--;
1063 	msgclear(m);
1064 	if(m->tpkt){
1065 		free(m->tpkt);
1066 		m->tpkt = nil;
1067 	}
1068 	if(m->rpkt){
1069 		free(m->rpkt);
1070 		m->rpkt = nil;
1071 	}
1072 	m->isopenfd = 0;
1073 	m->internal = 0;
1074 	m->next = freemsg;
1075 	freemsg = m;
1076 }
1077 
1078 Msg*
1079 msgget(int n)
1080 {
1081 	Msg *m;
1082 
1083 	if(n < 0 || n >= nmsgtab)
1084 		return nil;
1085 	m = msgtab[n];
1086 	if(m->ref == 0)
1087 		return nil;
1088 	if(verbose) fprint(2, "%T msgget %d = %p\n", n, m);
1089 	msgincref(m);
1090 	return m;
1091 }
1092 
1093 
1094 void*
1095 emalloc(int n)
1096 {
1097 	void *v;
1098 
1099 	v = mallocz(n, 1);
1100 	if(v == nil){
1101 		abort();
1102 		sysfatal("out of memory allocating %d", n);
1103 	}
1104 	return v;
1105 }
1106 
1107 void*
1108 erealloc(void *v, int n)
1109 {
1110 	v = realloc(v, n);
1111 	if(v == nil){
1112 		abort();
1113 		sysfatal("out of memory reallocating %d", n);
1114 	}
1115 	return v;
1116 }
1117 
1118 typedef struct Qel Qel;
1119 struct Qel
1120 {
1121 	Qel *next;
1122 	void *p;
1123 };
1124 
1125 struct Queue
1126 {
1127 	int hungup;
1128 	QLock lk;
1129 	Rendez r;
1130 	Qel *head;
1131 	Qel *tail;
1132 };
1133 
1134 Queue*
1135 qalloc(void)
1136 {
1137 	Queue *q;
1138 
1139 	q = mallocz(sizeof(Queue), 1);
1140 	if(q == nil)
1141 		return nil;
1142 	q->r.l = &q->lk;
1143 	return q;
1144 }
1145 
1146 int
1147 sendq(Queue *q, void *p)
1148 {
1149 	Qel *e;
1150 
1151 	e = emalloc(sizeof(Qel));
1152 	qlock(&q->lk);
1153 	if(q->hungup){
1154 		free(e);
1155 		werrstr("hungup queue");
1156 		qunlock(&q->lk);
1157 		return -1;
1158 	}
1159 	e->p = p;
1160 	e->next = nil;
1161 	if(q->head == nil)
1162 		q->head = e;
1163 	else
1164 		q->tail->next = e;
1165 	q->tail = e;
1166 	rwakeup(&q->r);
1167 	qunlock(&q->lk);
1168 	return 0;
1169 }
1170 
1171 void*
1172 recvq(Queue *q)
1173 {
1174 	void *p;
1175 	Qel *e;
1176 
1177 	qlock(&q->lk);
1178 	while(q->head == nil && !q->hungup)
1179 		rsleep(&q->r);
1180 	if(q->hungup){
1181 		qunlock(&q->lk);
1182 		return nil;
1183 	}
1184 	e = q->head;
1185 	q->head = e->next;
1186 	qunlock(&q->lk);
1187 	p = e->p;
1188 	free(e);
1189 	return p;
1190 }
1191 
1192 uchar*
1193 read9ppkt(Ioproc *io, int fd)
1194 {
1195 	uchar buf[4], *pkt;
1196 	int n, nn;
1197 
1198 	n = ioreadn(io, fd, buf, 4);
1199 	if(n != 4)
1200 		return nil;
1201 	n = GBIT32(buf);
1202 	pkt = emalloc(n);
1203 	PBIT32(pkt, n);
1204 	nn = ioreadn(io, fd, pkt+4, n-4);
1205 	if(nn != n-4){
1206 		free(pkt);
1207 		return nil;
1208 	}
1209 /* would do this if we ever got one of these, but we only generate them
1210 	if(pkt[4] == Ropenfd){
1211 		newfd = iorecvfd(io, fd);
1212 		PBIT32(pkt+n-4, newfd);
1213 	}
1214 */
1215 	return pkt;
1216 }
1217 
1218 Msg*
1219 mread9p(Ioproc *io, int fd, int dotu)
1220 {
1221 	int n, nn;
1222 	uchar *pkt;
1223 	Msg *m;
1224 
1225 	if((pkt = read9ppkt(io, fd)) == nil)
1226 		return nil;
1227 
1228 	m = msgnew(0);
1229 	m->tpkt = pkt;
1230 	n = GBIT32(pkt);
1231 	nn = convM2Su(pkt, n, &m->tx, dotu);
1232 	if(nn != n){
1233 		fprint(2, "%T read bad packet from %d\n", fd);
1234 		return nil;
1235 	}
1236 	return m;
1237 }
1238 
1239 int
1240 mwrite9p(Ioproc *io, int fd, uchar *pkt)
1241 {
1242 	int n, nfd;
1243 
1244 	n = GBIT32(pkt);
1245 	if(verbose > 2) fprint(2, "%T write %d %d %.*H\n", fd, n, n, pkt);
1246 if(verbose > 1) fprint(2, "%T before iowrite\n");
1247 	if(iowrite(io, fd, pkt, n) != n){
1248 		fprint(2, "%T write error: %r\n");
1249 		return -1;
1250 	}
1251 if(verbose > 1) fprint(2, "%T after iowrite\n");
1252 	if(pkt[4] == Ropenfd){
1253 		nfd = GBIT32(pkt+n-4);
1254 		if(iosendfd(io, fd, nfd) < 0){
1255 			fprint(2, "%T send fd error: %r\n");
1256 			return -1;
1257 		}
1258 	}
1259 	return 0;
1260 }
1261 
1262 void
1263 restring(uchar *pkt, int pn, char *s)
1264 {
1265 	int n;
1266 
1267 	if(s < (char*)pkt || s >= (char*)pkt+pn)
1268 		return;
1269 
1270 	n = strlen(s);
1271 	memmove(s+1, s, n);
1272 	PBIT16((uchar*)s-1, n);
1273 }
1274 
1275 void
1276 repack(Fcall *f, uchar **ppkt, int dotu)
1277 {
1278 	uint n, nn;
1279 	uchar *pkt;
1280 	
1281 	pkt = *ppkt;
1282 	n = GBIT32(pkt);
1283 	nn = sizeS2Mu(f, dotu);
1284 	if(nn > n){
1285 		free(pkt);
1286 		pkt = emalloc(nn);
1287 		*ppkt = pkt;
1288 	}
1289 	n = convS2Mu(f, pkt, nn, dotu);	
1290 	if(n <= BIT16SZ)
1291 		sysfatal("convS2M conversion error");
1292 	if(n != nn)
1293 		sysfatal("convS2Mu and sizeS2Mu disagree");
1294 }
1295 
1296 void
1297 rewritehdr(Fcall *f, uchar *pkt)
1298 {
1299 	int i, n;
1300 
1301 	n = GBIT32(pkt);
1302 	PBIT16(pkt+5, f->tag);
1303 	switch(f->type){
1304 	case Tversion:
1305 	case Rversion:
1306 		restring(pkt, n, f->version);
1307 		break;
1308 	case Tauth:
1309 		PBIT32(pkt+7, f->afid);
1310 		restring(pkt, n, f->uname);
1311 		restring(pkt, n, f->aname);
1312 		break;
1313 	case Tflush:
1314 		PBIT16(pkt+7, f->oldtag);
1315 		break;
1316 	case Tattach:
1317 		restring(pkt, n, f->uname);
1318 		restring(pkt, n, f->aname);
1319 		PBIT32(pkt+7, f->fid);
1320 		PBIT32(pkt+11, f->afid);
1321 		break;
1322 	case Twalk:
1323 		PBIT32(pkt+7, f->fid);
1324 		PBIT32(pkt+11, f->newfid);
1325 		for(i=0; i<f->nwname; i++)
1326 			restring(pkt, n, f->wname[i]);
1327 		break;
1328 	case Tcreate:
1329 		restring(pkt, n, f->name);
1330 		/* fall through */
1331 	case Topen:
1332 	case Tread:
1333 	case Twrite:
1334 	case Tclunk:
1335 	case Tremove:
1336 	case Tstat:
1337 	case Twstat:
1338 		PBIT32(pkt+7, f->fid);
1339 		break;
1340 	case Rerror:
1341 		restring(pkt, n, f->ename);
1342 		break;
1343 	}
1344 }
1345 
1346 static long
1347 _iolisten(va_list *arg)
1348 {
1349 	char *a, *b;
1350 
1351 	a = va_arg(*arg, char*);
1352 	b = va_arg(*arg, char*);
1353 	return listen(a, b);
1354 }
1355 
1356 int
1357 iolisten(Ioproc *io, char *a, char *b)
1358 {
1359 	return iocall(io, _iolisten, a, b);
1360 }
1361 
1362 static long
1363 _ioaccept(va_list *arg)
1364 {
1365 	int fd;
1366 	char *dir;
1367 
1368 	fd = va_arg(*arg, int);
1369 	dir = va_arg(*arg, char*);
1370 	return accept(fd, dir);
1371 }
1372 
1373 int
1374 ioaccept(Ioproc *io, int fd, char *dir)
1375 {
1376 	return iocall(io, _ioaccept, fd, dir);
1377 }
1378 
1379 int
1380 timefmt(Fmt *fmt)
1381 {
1382 	static char *mon[] = { "Jan", "Feb", "Mar", "Apr", "May", "Jun", 
1383 		"Jul", "Aug", "Sep", "Oct", "Nov", "Dec" };
1384 	vlong ns;
1385 	Tm tm;
1386 	ns = nsec();
1387 	tm = *localtime(time(0));
1388 	return fmtprint(fmt, "%s %2d %02d:%02d:%02d.%03d", 
1389 		mon[tm.mon], tm.mday, tm.hour, tm.min, tm.sec,
1390 		(int)(ns%1000000000)/1000000);
1391 }
1392 
1393 int
1394 cvtustat(Fcall *f, uchar **fpkt, int tounix)
1395 {
1396 	int n;
1397 	uchar *buf;
1398 	char *str;
1399 	Dir dir;
1400 
1401 	str = emalloc(f->nstat);
1402 	n = convM2Du(f->stat, f->nstat, &dir, str, !tounix);
1403 	if(n <= BIT16SZ){
1404 		free(str);
1405 		return -1;
1406 	}
1407 
1408 	n = sizeD2Mu(&dir, tounix);
1409 	buf = emalloc(n);
1410 	if(convD2Mu(&dir, buf, n, tounix) != n)
1411 		sysfatal("convD2Mu conversion error");
1412 	f->nstat = n;
1413 	f->stat = buf;
1414 
1415 	repack(f, fpkt, dotu);
1416 	free(buf);
1417 	f->stat = nil;	/* is this okay ??? */
1418 	free(str);
1419 
1420 	return 0;
1421 }
1422 
1423 int
1424 stripudirread(Msg* msg)
1425 {
1426 	char *str;
1427 	int i, m, n, nn;
1428 	uchar *buf;
1429 	Dir d;
1430 	Fcall* rx;
1431 
1432 	buf = nil;
1433 	str = nil;
1434 	rx = &msg->rx;
1435 	n = 0;
1436 	nn = 0;
1437 	for(i = 0; i < rx->count; i += m){
1438 		m = BIT16SZ + GBIT16(&rx->data[i]);
1439 		if(statchecku((uchar*)&rx->data[i], m, 1) < 0)
1440 			return -1;
1441 		if(nn < m)
1442 			nn = m;
1443 		n++;
1444 	}
1445 
1446 	str = emalloc(nn);
1447 	buf = emalloc(rx->count);
1448 
1449 	nn = 0;
1450 	for(i = 0; i < rx->count; i += m){
1451 		m = BIT16SZ + GBIT16(&rx->data[i]);
1452 		if(convM2Du((uchar*)&rx->data[i], m, &d, str, 1) != m){
1453 			free(buf);
1454 			free(str);
1455 			return -1;
1456 		}
1457 
1458 		n = convD2M(&d, &buf[nn], rx->count - nn);
1459 		if(n <= BIT16SZ){
1460 			free(buf);
1461 			free(str);
1462 			return -1;
1463 		}
1464 
1465 		nn += n;
1466 	}
1467 
1468 	rx->count = nn;
1469 	rx->data = (char*)buf;
1470 
1471 	repack(&msg->rx, &msg->rpkt, 0);
1472 	free(str);
1473 	free(buf);
1474 	rx->data = nil;	/* is this okay ??? */
1475 
1476 	return 0;
1477 }
1478