op public repos

Blob

Date:: Fri Mar 3 16:34:49 2006 UTC
Message:: change Tcreate message in 9P2000.u
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 Tcreate:
453 			if(dotu && !c->dotu && (m->tx.perm&(DMSYMLINK|DMDEVICE|DMNAMEDPIPE|DMSOCKET))){
454 				err(m, "unsupported file type");
455 				continue;
456 			}
457 			goto caseTopen;
458 		case Topenfd:
459 			if(m->tx.mode&~(OTRUNC|3)){
460 				err(m, "bad openfd mode");
461 				continue;
462 			}
463 			m->isopenfd = 1;
464 			m->tx.type = Topen;
465 			m->tpkt[4] = Topen;
466 			/* fall through */
467 		caseTopen:
468 		case Topen:
469 		case Tclunk:
470 		case Tread:
471 		case Twrite:
472 		case Tremove:
473 		case Tstat:
474 		case Twstat:
475 			if((m->fid = gethash(c->fid, m->tx.fid)) == nil){
476 				err(m, "unknown fid");
477 				continue;
478 			}
479 			m->fid->ref++;
480 			if(m->tx.type==Twstat && dotu && !c->dotu){
481 				if(cvtustat(&m->tx, &m->tpkt, 1) < 0){
482 					err(m, "cannot convert stat buffer");
483 					continue;
484 				}
485 			}
486 			if(m->tx.type==Tread && m->fid->isdir && dotu && !c->dotu){
487 				if(m->tx.offset = m->fid->coffset)
488 					m->tx.offset = m->fid->offset;
489 				else
490 					m->fid->offset = m->fid->coffset;
491 			}
492 			break;
493 		}
494 
495 		/* have everything - translate and send */
496 		m->c = c;
497 		m->ctag = m->tx.tag;
498 		m->tx.tag = m->tag;
499 		if(m->fid)
500 			m->tx.fid = m->fid->fid;
501 		if(m->newfid)
502 			m->tx.newfid = m->newfid->fid;
503 		if(m->afid)
504 			m->tx.afid = m->afid->fid;
505 		if(m->oldm)
506 			m->tx.oldtag = m->oldm->tag;
507 		/* reference passes to outq */
508 		sendq(outq, m);
509 		while(c->nmsg >= MAXMSG){
510 			c->inputstalled = 1;
511 			recvp(c->inc);
512 		}
513 	}
514 
515 	if(verbose) fprint(2, "%T fd#%d eof; flushing conn\n", c->fd);
516 
517 	/* flush the output queue */
518 	sendq(c->outq, nil);
519 	while(c->outq != nil)
520 		yield();
521 
522 	/* flush all outstanding messages */
523 	for(i=0; i<NHASH; i++){
524 		for(h=c->tag[i]; h; h=hnext){
525 			om = h->v;
526 			m = msgnew(0);
527 			m->internal = 1;
528 			m->c = c;
529 			c->nmsg++;
530 			m->tx.type = Tflush;
531 			m->tx.tag = m->tag;
532 			m->tx.oldtag = om->tag;
533 			m->oldm = om;
534 			msgincref(om);
535 			msgincref(m);	/* for outq */
536 			sendomsg(m);
537 			mm = recvp(c->internal);
538 			assert(mm == m);
539 			msgput(m);	/* got from recvp */
540 			msgput(m);	/* got from msgnew */
541 			msgput(om);	/* got from hash table */
542 			hnext = h->next;
543 			free(h);
544 		}
545 	}
546 
547 	/* clunk all outstanding fids */
548 	for(i=0; i<NHASH; i++){
549 		for(h=c->fid[i]; h; h=hnext){
550 			f = h->v;
551 			m = msgnew(0);
552 			m->internal = 1;
553 			m->c = c;
554 			c->nmsg++;
555 			m->tx.type = Tclunk;
556 			m->tx.tag = m->tag;
557 			m->tx.fid = f->fid;
558 			m->fid = f;
559 			f->ref++;
560 			msgincref(m);
561 			sendomsg(m);
562 			mm = recvp(c->internal);
563 			assert(mm == m);
564 			msgclear(m);
565 			msgput(m);	/* got from recvp */
566 			msgput(m);	/* got from msgnew */
567 			fidput(f);	/* got from hash table */
568 			hnext = h->next;
569 			free(h);
570 		}
571 	}
572 
573 out:
574 	closeioproc(io);
575 	assert(c->nmsg == 0);
576 	assert(c->nfid == 0);
577 	close(c->fd);
578 	chanfree(c->internal);
579 	c->internal = 0;
580 	chanfree(c->inc);
581 	c->inc = 0;
582 	free(c->inq);
583 	c->inq = 0;
584 	free(c);
585 }
586 
587 static void
588 openfdthread(void *v)
589 {
590 	Conn *c;
591 	Fid *fid;
592 	Msg *m;
593 	int n;
594 	vlong tot;
595 	Ioproc *io;
596 	char buf[1024];
597 
598 	c = v;
599 	fid = c->fdfid;
600 	io = ioproc();
601 	threadsetname("openfd %s", c->fdfid);
602 	tot = 0;
603 	m = nil;
604 	if(c->fdmode == OREAD){
605 		for(;;){
606 			if(verbose) fprint(2, "%T tread...");
607 			m = msgnew(0);
608 			m->internal = 1;
609 			m->c = c;
610 			m->tx.type = Tread;
611 			m->tx.count = msize - IOHDRSZ;
612 			m->tx.fid = fid->fid;
613 			m->tx.tag = m->tag;
614 			m->tx.offset = tot;
615 			m->fid = fid;
616 			fid->ref++;
617 			msgincref(m);
618 			sendomsg(m);
619 			recvp(c->internal);
620 			if(m->rx.type == Rerror){
621 			//	fprint(2, "%T read error: %s\n", m->rx.ename);
622 				break;
623 			}
624 			if(m->rx.count == 0)
625 				break;
626 			tot += m->rx.count;
627 			if(iowrite(io, c->fd, m->rx.data, m->rx.count) != m->rx.count){
628 				// fprint(2, "%T pipe write error: %r\n");
629 				break;
630 			}
631 			msgput(m);
632 			msgput(m);
633 			m = nil;
634 		}
635 	}else{
636 		for(;;){
637 			if(verbose) fprint(2, "%T twrite...");
638 			n = sizeof buf;
639 			if(n > msize)
640 				n = msize;
641 			if((n=ioread(io, c->fd, buf, n)) <= 0){
642 				if(n < 0)
643 					fprint(2, "%T pipe read error: %r\n");
644 				break;
645 			}
646 			m = msgnew(0);
647 			m->internal = 1;
648 			m->c = c;
649 			m->tx.type = Twrite;
650 			m->tx.fid = fid->fid;
651 			m->tx.data = buf;
652 			m->tx.count = n;
653 			m->tx.tag = m->tag;
654 			m->tx.offset = tot;
655 			m->fid = fid;
656 			fid->ref++;
657 			msgincref(m);
658 			sendomsg(m);
659 			recvp(c->internal);
660 			if(m->rx.type == Rerror){
661 			//	fprint(2, "%T write error: %s\n", m->rx.ename);
662 			}
663 			tot += n;
664 			msgput(m);
665 			msgput(m);
666 			m = nil;
667 		}
668 	}
669 	if(verbose) fprint(2, "%T eof on %d fid %d\n", c->fd, fid->fid);
670 	close(c->fd);
671 	closeioproc(io);
672 	if(m){
673 		msgput(m);
674 		msgput(m);
675 	}
676 	if(verbose) fprint(2, "%T eof on %d fid %d ref %d\n", c->fd, fid->fid, fid->ref);
677 	if(--fid->openfd == 0){
678 		m = msgnew(0);
679 		m->internal = 1;
680 		m->c = c;
681 		m->tx.type = Tclunk;
682 		m->tx.tag = m->tag;
683 		m->tx.fid = fid->fid;
684 		m->fid = fid;
685 		fid->ref++;
686 		msgincref(m);
687 		sendomsg(m);
688 		recvp(c->internal);
689 		msgput(m);
690 		msgput(m);
691 	}
692 	fidput(fid);
693 	c->fdfid = nil;
694 	chanfree(c->internal);
695 	c->internal = 0;
696 	free(c);
697 }			
698 
699 int
700 xopenfd(Msg *m)
701 {
702 	char errs[ERRMAX];
703 	int n, p[2];
704 	Conn *nc;
705 
706 	if(pipe(p) < 0){
707 		rerrstr(errs, sizeof errs);
708 		err(m, errs);
709 		/* XXX return here? */
710 	}
711 	if(verbose) fprint(2, "%T xopen pipe %d %d...", p[0], p[1]);
712 
713 	/* now we're committed. */
714 
715 	/* a new connection for this fid */
716 	nc = emalloc(sizeof(Conn));
717 	nc->internal = chancreate(sizeof(void*), 0);
718 
719 	/* a ref for us */
720 	nc->fdfid = m->fid;
721 	m->fid->ref++;
722 	nc->fdfid->openfd++;
723 	nc->fdmode = m->tx.mode;
724 	nc->fd = p[0];
725 
726 	/* a thread to tend the pipe */
727 	threadcreate(openfdthread, nc, STACK);
728 
729 	/* if mode is ORDWR, that openfdthread will write; start a reader */
730 	if((m->tx.mode&3) == ORDWR){
731 		nc = emalloc(sizeof(Conn));
732 		nc->internal = chancreate(sizeof(void*), 0);
733 		nc->fdfid = m->fid;
734 		m->fid->ref++;
735 		nc->fdfid->openfd++;
736 		nc->fdmode = OREAD;
737 		nc->fd = dup(p[0], -1);
738 		threadcreate(openfdthread, nc, STACK);
739 	}
740 
741 	/* steal fid from other connection */
742 	if(delhash(m->c->fid, m->fid->cfid, m->fid) == 0)
743 		fidput(m->fid);
744 
745 	/* rewrite as Ropenfd */
746 	m->rx.type = Ropenfd;
747 	n = GBIT32(m->rpkt);
748 	m->rpkt = erealloc(m->rpkt, n+4);
749 	PBIT32(m->rpkt+n, p[1]);
750 	n += 4;
751 	PBIT32(m->rpkt, n);
752 	m->rpkt[4] = Ropenfd;
753 	m->rx.unixfd = p[1];
754 	return 0;
755 }
756 
757 void
758 connoutthread(void *arg)
759 {
760 	char *ename;
761 	int err;
762 	Conn *c;
763 	Queue *outq;
764 	Msg *m, *om;
765 	Ioproc *io;
766 
767 	c = arg;
768 	outq = c->outq;
769 	io = ioproc();
770 	threadsetname("connout %s", c->dir);
771 	while((m = recvq(outq)) != nil){
772 		err = m->tx.type+1 != m->rx.type;
773 		if(!err && m->isopenfd)
774 			if(xopenfd(m) < 0)
775 				continue;
776 		switch(m->tx.type){
777 		case Tflush:
778 			om = m->oldm;
779 			if(om)
780 				if(delhash(om->c->tag, om->ctag, om) == 0)
781 					msgput(om);
782 			break;
783 		case Tclunk:
784 		case Tremove:
785 			if(m->fid)
786 				if(delhash(m->c->fid, m->fid->cfid, m->fid) == 0)
787 					fidput(m->fid);
788 			break;
789 		case Tauth:
790 			if(err && m->afid){
791 				if(verbose) fprint(2, "%T auth error\n");
792 				if(delhash(m->c->fid, m->afid->cfid, m->afid) == 0)
793 					fidput(m->afid);
794 			}
795 			break;
796 		case Tattach:
797 			if(err && m->fid)
798 				if(delhash(m->c->fid, m->fid->cfid, m->fid) == 0)
799 					fidput(m->fid);
800 			break;
801 		case Twalk:
802 			if(err || m->rx.nwqid < m->tx.nwname)
803 			if(m->tx.fid != m->tx.newfid && m->newfid)
804 				if(delhash(m->c->fid, m->newfid->cfid, m->newfid) == 0)
805 					fidput(m->newfid);
806 			break;
807 		case Tread:
808 			if(!err && m->fid->isdir && dotu && !m->c->dotu){
809 				m->fid->offset += m->rx.count;
810 				stripudirread(m);
811 				m->fid->coffset += m->rx.count;
812 			}
813 			break;
814 		case Tstat:
815 			if(!err && dotu && !m->c->dotu)
816 				cvtustat(&m->rx, &m->rpkt, 0);
817 			break;
818 		case Topen:
819 		case Tcreate:
820 			m->fid->isdir = (m->rx.qid.type & QTDIR);
821 			break;
822 		}
823 		if(m->rx.type==Rerror && dotu && !c->dotu){
824 			ename = estrdup(m->rx.ename);
825 			m->rx.ename = ename;
826 			repack(&m->rx, &m->rpkt, c->dotu);
827 			free(ename);
828 			m->rx.ename = "XXX";
829 		}
830 		if(delhash(m->c->tag, m->ctag, m) == 0)
831 			msgput(m);
832 		if(verbose > 1) fprint(2, "%T fd#%d <- %F\n", c->fd, &m->rx);
833 		rewritehdr(&m->rx, m->rpkt);
834 		if(mwrite9p(io, c->fd, m->rpkt) < 0)
835 			if(verbose) fprint(2, "%T write error: %r\n");
836 		msgput(m);
837 		if(c->inputstalled && c->nmsg < MAXMSG)
838 			nbsendp(c->inc, 0);
839 	}
840 	closeioproc(io);
841 	free(outq);
842 	c->outq = nil;
843 }
844 
845 void
846 outputthread(void *arg)
847 {
848 	Msg *m;
849 	Ioproc *io;
850 
851 	USED(arg);
852 	io = ioproc();
853 	threadsetname("output");
854 	while((m = recvq(outq)) != nil){
855 		if(verbose > 1) fprint(2, "%T * <- %F\n", &m->tx);
856 		rewritehdr(&m->tx, m->tpkt);
857 		if(mwrite9p(io, 1, m->tpkt) < 0)
858 			sysfatal("output error: %r");
859 		msgput(m);
860 	}
861 	closeioproc(io);
862 	fprint(2, "%T output eof\n");
863 	threadexitsall(0);
864 }	
865 
866 void
867 inputthread(void *arg)
868 {
869 	uchar *pkt;
870 	int n, nn, tag;
871 	Msg *m;
872 	Ioproc *io;
873 
874 	threadsetname("input");
875 	if(verbose) fprint(2, "%T input thread\n");
876 	io = ioproc();
877 	USED(arg);
878 	while((pkt = read9ppkt(io, 0)) != nil){
879 		n = GBIT32(pkt);
880 		if(n < 7){
881 			fprint(2, "%T short 9P packet from server\n");
882 			free(pkt);
883 			continue;
884 		}
885 		if(verbose > 2) fprint(2, "%T read %.*H\n", n, pkt);
886 		tag = GBIT16(pkt+5);
887 		if((m = msgget(tag)) == nil){
888 			fprint(2, "%T unexpected 9P response tag %d\n", tag);
889 			free(pkt);
890 			continue;
891 		}
892 		if((nn = convM2Su(pkt, n, &m->rx, dotu)) != n){
893 			fprint(2, "%T bad packet - convM2S %d but %d\n", nn, n);
894 			free(pkt);
895 			msgput(m);
896 			continue;
897 		}
898 		if(verbose > 1) fprint(2, "%T * -> %F%s\n", &m->rx,
899 			m->internal ? " (internal)" : "");
900 		m->rpkt = pkt;
901 		m->rx.tag = m->ctag;
902 		if(m->internal)
903 			sendp(m->c->internal, m);
904 		else if(m->c->outq)
905 			sendq(m->c->outq, m);
906 		else
907 			msgput(m);
908 	}
909 	closeioproc(io);
910 	//fprint(2, "%T input eof\n");
911 	threadexitsall(0);
912 }
913 
914 void*
915 gethash(Hash **ht, uint n)
916 {
917 	Hash *h;
918 
919 	for(h=ht[n%NHASH]; h; h=h->next)
920 		if(h->n == n)
921 			return h->v;
922 	return nil;
923 }
924 
925 int
926 delhash(Hash **ht, uint n, void *v)
927 {
928 	Hash *h, **l;
929 
930 	for(l=&ht[n%NHASH]; h=*l; l=&h->next)
931 		if(h->n == n){
932 			if(h->v != v){
933 				if(verbose) fprint(2, "%T delhash %d got %p want %p\n", n, h->v, v);
934 				return -1;
935 			}
936 			*l = h->next;
937 			free(h);
938 			return 0;
939 		}
940 	return -1;
941 }
942 
943 int
944 puthash(Hash **ht, uint n, void *v)
945 {
946 	Hash *h;
947 
948 	if(gethash(ht, n))
949 		return -1;
950 	h = emalloc(sizeof(Hash));
951 	h->next = ht[n%NHASH];
952 	h->n = n;
953 	h->v = v;
954 	ht[n%NHASH] = h;
955 	return 0;
956 }
957 
958 Fid **fidtab;
959 int nfidtab;
960 Fid *freefid;
961 
962 Fid*
963 fidnew(int cfid)
964 {
965 	Fid *f;
966 
967 	if(freefid == nil){
968 		fidtab = erealloc(fidtab, (nfidtab+1)*sizeof(fidtab[0]));
969 		if(nfidtab == xafid){
970 			fidtab[nfidtab++] = nil;
971 			fidtab = erealloc(fidtab, (nfidtab+1)*sizeof(fidtab[0]));
972 		}
973 		fidtab[nfidtab] = emalloc(sizeof(Fid));
974 		freefid = fidtab[nfidtab];
975 		freefid->fid = nfidtab++;
976 	}
977 	f = freefid;
978 	freefid = f->next;
979 	f->cfid = cfid;
980 	f->ref = 1;
981 	f->offset = 0;
982 	f->coffset = 0;
983 	f->isdir = -1;
984 	return f;
985 }
986 
987 void
988 fidput(Fid *f)
989 {
990 	if(f == nil)
991 		return;
992 	assert(f->ref > 0);
993 	if(--f->ref > 0)
994 		return;
995 	f->next = freefid;
996 	f->cfid = -1;
997 	freefid = f;
998 }
999 
1000 Msg **msgtab;
1001 int nmsgtab;
1002 int nmsg;
1003 Msg *freemsg;
1004 
1005 void
1006 msgincref(Msg *m)
1007 {
1008 	if(verbose > 1) fprint(2, "%T msgincref @0x%lux %p tag %d/%d ref %d=>%d\n",
1009 		getcallerpc(&m), m, m->tag, m->ctag, m->ref, m->ref+1);
1010 	m->ref++;
1011 }
1012 
1013 Msg*
1014 msgnew(int x)
1015 {
1016 	Msg *m;
1017 
1018 	if(freemsg == nil){
1019 		msgtab = erealloc(msgtab, (nmsgtab+1)*sizeof(msgtab[0]));
1020 		msgtab[nmsgtab] = emalloc(sizeof(Msg));
1021 		freemsg = msgtab[nmsgtab];
1022 		freemsg->tag = nmsgtab++;
1023 	}
1024 	m = freemsg;
1025 	freemsg = m->next;
1026 	m->ref = 1;
1027 	if(verbose > 1) fprint(2, "%T msgnew @0x%lux %p tag %d ref %d\n",
1028 		getcallerpc(&x), m, m->tag, m->ref);
1029 	nmsg++;
1030 	return m;
1031 }
1032 
1033 /*
1034  * Clear data associated with connections, so that
1035  * if all msgs have been msgcleared, the connection
1036  * can be freed.  Note that this does *not* free the tpkt
1037  * and rpkt; they are freed in msgput with the msg itself. 
1038  * The io write thread might still be holding a ref to msg
1039  * even once the connection has finished with it.
1040  */
1041 void
1042 msgclear(Msg *m)
1043 {
1044 	if(m->c){
1045 		m->c->nmsg--;
1046 		m->c = nil;
1047 	}
1048 	if(m->oldm){
1049 		msgput(m->oldm);
1050 		m->oldm = nil;
1051 	}
1052 	if(m->fid){
1053 		fidput(m->fid);
1054 		m->fid = nil;
1055 	}
1056 	if(m->afid){
1057 		fidput(m->afid);
1058 		m->afid = nil;
1059 	}
1060 	if(m->newfid){
1061 		fidput(m->newfid);
1062 		m->newfid = nil;
1063 	}
1064 	if(m->rx.type == Ropenfd && m->rx.unixfd >= 0){
1065 		close(m->rx.unixfd);
1066 		m->rx.unixfd = -1;
1067 	}
1068 }
1069 
1070 void
1071 msgput(Msg *m)
1072 {
1073 	if(m == nil)
1074 		return;
1075 
1076 	if(verbose > 1) fprint(2, "%T msgput 0x%lux %p tag %d/%d ref %d\n", 
1077 		getcallerpc(&m), m, m->tag, m->ctag, m->ref);
1078 	assert(m->ref > 0);
1079 	if(--m->ref > 0)
1080 		return;
1081 	nmsg--;
1082 	msgclear(m);
1083 	if(m->tpkt){
1084 		free(m->tpkt);
1085 		m->tpkt = nil;
1086 	}
1087 	if(m->rpkt){
1088 		free(m->rpkt);
1089 		m->rpkt = nil;
1090 	}
1091 	m->isopenfd = 0;
1092 	m->internal = 0;
1093 	m->next = freemsg;
1094 	freemsg = m;
1095 }
1096 
1097 Msg*
1098 msgget(int n)
1099 {
1100 	Msg *m;
1101 
1102 	if(n < 0 || n >= nmsgtab)
1103 		return nil;
1104 	m = msgtab[n];
1105 	if(m->ref == 0)
1106 		return nil;
1107 	if(verbose) fprint(2, "%T msgget %d = %p\n", n, m);
1108 	msgincref(m);
1109 	return m;
1110 }
1111 
1112 
1113 void*
1114 emalloc(int n)
1115 {
1116 	void *v;
1117 
1118 	v = mallocz(n, 1);
1119 	if(v == nil){
1120 		abort();
1121 		sysfatal("out of memory allocating %d", n);
1122 	}
1123 	return v;
1124 }
1125 
1126 void*
1127 erealloc(void *v, int n)
1128 {
1129 	v = realloc(v, n);
1130 	if(v == nil){
1131 		abort();
1132 		sysfatal("out of memory reallocating %d", n);
1133 	}
1134 	return v;
1135 }
1136 
1137 typedef struct Qel Qel;
1138 struct Qel
1139 {
1140 	Qel *next;
1141 	void *p;
1142 };
1143 
1144 struct Queue
1145 {
1146 	int hungup;
1147 	QLock lk;
1148 	Rendez r;
1149 	Qel *head;
1150 	Qel *tail;
1151 };
1152 
1153 Queue*
1154 qalloc(void)
1155 {
1156 	Queue *q;
1157 
1158 	q = mallocz(sizeof(Queue), 1);
1159 	if(q == nil)
1160 		return nil;
1161 	q->r.l = &q->lk;
1162 	return q;
1163 }
1164 
1165 int
1166 sendq(Queue *q, void *p)
1167 {
1168 	Qel *e;
1169 
1170 	e = emalloc(sizeof(Qel));
1171 	qlock(&q->lk);
1172 	if(q->hungup){
1173 		free(e);
1174 		werrstr("hungup queue");
1175 		qunlock(&q->lk);
1176 		return -1;
1177 	}
1178 	e->p = p;
1179 	e->next = nil;
1180 	if(q->head == nil)
1181 		q->head = e;
1182 	else
1183 		q->tail->next = e;
1184 	q->tail = e;
1185 	rwakeup(&q->r);
1186 	qunlock(&q->lk);
1187 	return 0;
1188 }
1189 
1190 void*
1191 recvq(Queue *q)
1192 {
1193 	void *p;
1194 	Qel *e;
1195 
1196 	qlock(&q->lk);
1197 	while(q->head == nil && !q->hungup)
1198 		rsleep(&q->r);
1199 	if(q->hungup){
1200 		qunlock(&q->lk);
1201 		return nil;
1202 	}
1203 	e = q->head;
1204 	q->head = e->next;
1205 	qunlock(&q->lk);
1206 	p = e->p;
1207 	free(e);
1208 	return p;
1209 }
1210 
1211 uchar*
1212 read9ppkt(Ioproc *io, int fd)
1213 {
1214 	uchar buf[4], *pkt;
1215 	int n, nn;
1216 
1217 	n = ioreadn(io, fd, buf, 4);
1218 	if(n != 4)
1219 		return nil;
1220 	n = GBIT32(buf);
1221 	pkt = emalloc(n);
1222 	PBIT32(pkt, n);
1223 	nn = ioreadn(io, fd, pkt+4, n-4);
1224 	if(nn != n-4){
1225 		free(pkt);
1226 		return nil;
1227 	}
1228 /* would do this if we ever got one of these, but we only generate them
1229 	if(pkt[4] == Ropenfd){
1230 		newfd = iorecvfd(io, fd);
1231 		PBIT32(pkt+n-4, newfd);
1232 	}
1233 */
1234 	return pkt;
1235 }
1236 
1237 Msg*
1238 mread9p(Ioproc *io, int fd, int dotu)
1239 {
1240 	int n, nn;
1241 	uchar *pkt;
1242 	Msg *m;
1243 
1244 	if((pkt = read9ppkt(io, fd)) == nil)
1245 		return nil;
1246 
1247 	m = msgnew(0);
1248 	m->tpkt = pkt;
1249 	n = GBIT32(pkt);
1250 	nn = convM2Su(pkt, n, &m->tx, dotu);
1251 	if(nn != n){
1252 		fprint(2, "%T read bad packet from %d\n", fd);
1253 		return nil;
1254 	}
1255 	return m;
1256 }
1257 
1258 int
1259 mwrite9p(Ioproc *io, int fd, uchar *pkt)
1260 {
1261 	int n, nfd;
1262 
1263 	n = GBIT32(pkt);
1264 	if(verbose > 2) fprint(2, "%T write %d %d %.*H\n", fd, n, n, pkt);
1265 if(verbose > 1) fprint(2, "%T before iowrite\n");
1266 	if(iowrite(io, fd, pkt, n) != n){
1267 		fprint(2, "%T write error: %r\n");
1268 		return -1;
1269 	}
1270 if(verbose > 1) fprint(2, "%T after iowrite\n");
1271 	if(pkt[4] == Ropenfd){
1272 		nfd = GBIT32(pkt+n-4);
1273 		if(iosendfd(io, fd, nfd) < 0){
1274 			fprint(2, "%T send fd error: %r\n");
1275 			return -1;
1276 		}
1277 	}
1278 	return 0;
1279 }
1280 
1281 void
1282 restring(uchar *pkt, int pn, char *s)
1283 {
1284 	int n;
1285 
1286 	if(s < (char*)pkt || s >= (char*)pkt+pn)
1287 		return;
1288 
1289 	n = strlen(s);
1290 	memmove(s+1, s, n);
1291 	PBIT16((uchar*)s-1, n);
1292 }
1293 
1294 void
1295 repack(Fcall *f, uchar **ppkt, int dotu)
1296 {
1297 	uint n, nn;
1298 	uchar *pkt;
1299 	
1300 	pkt = *ppkt;
1301 	n = GBIT32(pkt);
1302 	nn = sizeS2Mu(f, dotu);
1303 	if(nn > n){
1304 		free(pkt);
1305 		pkt = emalloc(nn);
1306 		*ppkt = pkt;
1307 	}
1308 	n = convS2Mu(f, pkt, nn, dotu);	
1309 	if(n <= BIT16SZ)
1310 		sysfatal("convS2M conversion error");
1311 	if(n != nn)
1312 		sysfatal("convS2Mu and sizeS2Mu disagree");
1313 }
1314 
1315 void
1316 rewritehdr(Fcall *f, uchar *pkt)
1317 {
1318 	int i, n;
1319 
1320 	n = GBIT32(pkt);
1321 	PBIT16(pkt+5, f->tag);
1322 	switch(f->type){
1323 	case Tversion:
1324 	case Rversion:
1325 		restring(pkt, n, f->version);
1326 		break;
1327 	case Tauth:
1328 		PBIT32(pkt+7, f->afid);
1329 		restring(pkt, n, f->uname);
1330 		restring(pkt, n, f->aname);
1331 		break;
1332 	case Tflush:
1333 		PBIT16(pkt+7, f->oldtag);
1334 		break;
1335 	case Tattach:
1336 		restring(pkt, n, f->uname);
1337 		restring(pkt, n, f->aname);
1338 		PBIT32(pkt+7, f->fid);
1339 		PBIT32(pkt+11, f->afid);
1340 		break;
1341 	case Twalk:
1342 		PBIT32(pkt+7, f->fid);
1343 		PBIT32(pkt+11, f->newfid);
1344 		for(i=0; i<f->nwname; i++)
1345 			restring(pkt, n, f->wname[i]);
1346 		break;
1347 	case Tcreate:
1348 		restring(pkt, n, f->name);
1349 		/* fall through */
1350 	case Topen:
1351 	case Tclunk:
1352 	case Tremove:
1353 	case Tstat:
1354 	case Twstat:
1355 	case Twrite:
1356 		PBIT32(pkt+7, f->fid);
1357 		break;
1358 	case Tread:
1359 		PBIT32(pkt+7, f->fid);
1360 		PBIT64(pkt+11, f->offset);
1361 		break;
1362 	case Rerror:
1363 		restring(pkt, n, f->ename);
1364 		break;
1365 	}
1366 }
1367 
1368 static long
1369 _iolisten(va_list *arg)
1370 {
1371 	char *a, *b;
1372 
1373 	a = va_arg(*arg, char*);
1374 	b = va_arg(*arg, char*);
1375 	return listen(a, b);
1376 }
1377 
1378 int
1379 iolisten(Ioproc *io, char *a, char *b)
1380 {
1381 	return iocall(io, _iolisten, a, b);
1382 }
1383 
1384 static long
1385 _ioaccept(va_list *arg)
1386 {
1387 	int fd;
1388 	char *dir;
1389 
1390 	fd = va_arg(*arg, int);
1391 	dir = va_arg(*arg, char*);
1392 	return accept(fd, dir);
1393 }
1394 
1395 int
1396 ioaccept(Ioproc *io, int fd, char *dir)
1397 {
1398 	return iocall(io, _ioaccept, fd, dir);
1399 }
1400 
1401 int
1402 timefmt(Fmt *fmt)
1403 {
1404 	static char *mon[] = { "Jan", "Feb", "Mar", "Apr", "May", "Jun", 
1405 		"Jul", "Aug", "Sep", "Oct", "Nov", "Dec" };
1406 	vlong ns;
1407 	Tm tm;
1408 	ns = nsec();
1409 	tm = *localtime(time(0));
1410 	return fmtprint(fmt, "%s %2d %02d:%02d:%02d.%03d", 
1411 		mon[tm.mon], tm.mday, tm.hour, tm.min, tm.sec,
1412 		(int)(ns%1000000000)/1000000);
1413 }
1414 
1415 int
1416 cvtustat(Fcall *f, uchar **fpkt, int tounix)
1417 {
1418 	int n;
1419 	uchar *buf;
1420 	char *str;
1421 	Dir dir;
1422 
1423 	str = emalloc(f->nstat);
1424 	n = convM2Du(f->stat, f->nstat, &dir, str, !tounix);
1425 	if(n <= BIT16SZ){
1426 		free(str);
1427 		return -1;
1428 	}
1429 
1430 	n = sizeD2Mu(&dir, tounix);
1431 	buf = emalloc(n);
1432 	if(convD2Mu(&dir, buf, n, tounix) != n)
1433 		sysfatal("convD2Mu conversion error");
1434 	f->nstat = n;
1435 	f->stat = buf;
1436 
1437 	repack(f, fpkt, dotu);
1438 	free(buf);
1439 	f->stat = nil;	/* is this okay ??? */
1440 	free(str);
1441 
1442 	return 0;
1443 }
1444 
1445 int
1446 stripudirread(Msg* msg)
1447 {
1448 	char *str;
1449 	int i, m, n, nn;
1450 	uchar *buf;
1451 	Dir d;
1452 	Fcall* rx;
1453 
1454 	buf = nil;
1455 	str = nil;
1456 	rx = &msg->rx;
1457 	n = 0;
1458 	nn = 0;
1459 	for(i = 0; i < rx->count; i += m){
1460 		m = BIT16SZ + GBIT16(&rx->data[i]);
1461 		if(statchecku((uchar*)&rx->data[i], m, 1) < 0)
1462 			return -1;
1463 		if(nn < m)
1464 			nn = m;
1465 		n++;
1466 	}
1467 
1468 	str = emalloc(nn);
1469 	buf = emalloc(rx->count);
1470 
1471 	nn = 0;
1472 	for(i = 0; i < rx->count; i += m){
1473 		m = BIT16SZ + GBIT16(&rx->data[i]);
1474 		if(convM2Du((uchar*)&rx->data[i], m, &d, str, 1) != m){
1475 			free(buf);
1476 			free(str);
1477 			return -1;
1478 		}
1479 
1480 		n = convD2M(&d, &buf[nn], rx->count - nn);
1481 		if(n <= BIT16SZ){
1482 			free(buf);
1483 			free(str);
1484 			return -1;
1485 		}
1486 
1487 		nn += n;
1488 	}
1489 
1490 	rx->count = nn;
1491 	rx->data = (char*)buf;
1492 
1493 	repack(&msg->rx, &msg->rpkt, 0);
1494 	free(str);
1495 	free(buf);
1496 	rx->data = nil;	/* is this okay ??? */
1497 
1498 	return 0;
1499 }
1500