Blob


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