op public repos

Blob

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