From f00660d2feac2b97bfd2dfe25a0d25da3ad730c7 Mon Sep 17 00:00:00 2001 From: David du Colombier <0intro@gmail.com> Date: Fri, 29 Jun 1990 00:00:00 +0000 Subject: [PATCH] Plan 9 from Bell Labs 1990-06-29 --- gnot/devcons.c | 100 +++++++++++++++++++++++++++++--------- gnot/devincon.c | 12 +++++ gnot/fns.h | 3 ++ gnot/proc.c | 4 +- gnot/screen.c | 38 ++++++++++++--- gnot/sturp.c | 75 +++++++++++++++++++++++------ port/devbit.c | 24 ++++++++-- port/devcons.c | 2 +- port/devmnt.c | 4 ++ port/devpipe.c | 62 +++++++++++------------- port/proc.c | 4 +- port/stream.c | 115 ++++++++++++++++++++++++++++++++++---------- port/sturp.c | 75 +++++++++++++++++++++++------ power/conf.h | 2 + power/dat.h | 124 +++++++++++++++++++++++++++++++++++++++++++++++- power/fns.h | 6 +++ 16 files changed, 525 insertions(+), 125 deletions(-) diff --git a/gnot/devcons.c b/gnot/devcons.c index 157e4cbc02378b4b7a7017f4deaf1dfc9c7ba056..a9135ecb760cbbd773a2adf551cdb183b8133b32 100644 --- a/gnot/devcons.c +++ b/gnot/devcons.c @@ -17,7 +17,7 @@ static struct typedef struct IOQ IOQ; -#define NQ 4096 +#define NQ 1024 struct IOQ{ union{ Lock; @@ -31,6 +31,8 @@ struct IOQ{ }; IOQ lineq; +IOQ rs232iq; +IOQ rs232oq; struct{ IOQ; /* qlock to getc; interrupt putc's */ @@ -50,12 +52,20 @@ printinit(void) kbdq.in = kbdq.buf; kbdq.out = kbdq.buf; + rs232iq.in = rs232iq.buf; + rs232iq.out = rs232iq.buf; + rs232oq.in = rs232oq.buf; + rs232oq.out = rs232oq.buf; lineq.in = lineq.buf; lineq.out = lineq.buf; qlock(&kbdq); /* allocate qlock */ qunlock(&kbdq); lock(&lineq); /* allocate lock */ unlock(&lineq); + lock(&rs232iq); /* allocate lock */ + unlock(&rs232iq); + lock(&rs232oq); /* allocate lock */ + unlock(&rs232oq); screeninit(); } @@ -114,6 +124,14 @@ getc(IOQ *q) return c; } +void +putc(IOQ *q, int c) +{ + *q->in++ = c; + if(q->in == q->buf+sizeof(q->buf)) + q->in = q->buf; +} + int sprint(char *s, char *fmt, ...) { @@ -184,12 +202,15 @@ echo(int c) */ if(c == 0x14) DEBUG(); + if(c == 0x16){ + incontoggle(); + urpdump(); + dumpqueues(); + } if(raw.ref) return; if(c == 0x15) putstrn("^U\n", 3); - if(c == 0x16) - dumpqueues(); else{ ch = c; putstrn(&ch, 1); @@ -235,6 +256,26 @@ kbdclock(void) kbdchar(kbdq.c); } +void +rs232ichar(int c) +{ + *rs232iq.in++ = c; + if(rs232iq.in == rs232iq.buf+sizeof(rs232iq.buf)) + rs232iq.in = rs232iq.buf; + wakeup(&rs232iq.r); +} + +int +getrs232o(void) +{ + int c; + + c = getc(&rs232oq); + if(c == -1) + rs232oq.state = 0; + return c; +} + int consactive(void) { @@ -253,6 +294,7 @@ enum{ Qpid, Qppid, Qrcons, + Qrs232, Qtime, Quser, }; @@ -265,6 +307,7 @@ Dirtab consdir[]={ "pid", Qpid, 12, 0600, "ppid", Qppid, 12, 0600, "rcons", Qrcons, 0, 0600, + "rs232", Qrs232, 0, 0600, "time", Qtime, 12, 0600, "user", Quser, 0, 0600, }; @@ -356,7 +399,7 @@ consopen(Chan *c, int omode) error(0, Eperm); } if(c->qid == Qrcons) - if(incref(&raw) == 0){ + if(incref(&raw) == 1){ lock(&lineq); while((ch=getc(&kbdq)) != -1){ *lineq.in++ = ch; @@ -446,6 +489,22 @@ consread(Chan *c, void *buf, long n) qunlock(&kbdq); return i; + case Qrs232: + qlock(&rs232iq); + if(waserror()){ + qunlock(&rs232iq); + nexterror(); + } + while(!cangetc(&rs232iq)) + sleep(&rs232iq.r, (int(*)(void*))cangetc, &rs232iq); + for(i=0; ioffset; if(k >= sizeof tmp) @@ -513,6 +572,20 @@ conswrite(Chan *c, void *va, long n) } break; + case Qrs232: + qlock(&rs232oq); + l = n; + while(--l >= 0) + putc(&rs232oq, *a++); + splhi(); + if(rs232oq.state == 0){ + rs232oq.state = 1; + duartstartrs232o(); + } + spllo(); + qunlock(&rs232oq); + break; + case Qtime: if(n<=0 || boottime!=0) /* only one write please */ return 0; @@ -571,22 +644,3 @@ consuserstr(Error *e, char *buf) { strcpy(buf, u->p->pgrp->user); } - -typedef struct Incon{ - unsigned char cdata; unsigned char u0; - unsigned char cstatus; unsigned char u1; - unsigned char creset; unsigned char u2; - unsigned char csend; unsigned char u3; - unsigned short data_cntl; /* data is high byte, cntl is low byte */ - unsigned char status; unsigned char u5; - unsigned char reset; unsigned char u6; - unsigned char send; unsigned char u7; -}Incon; - -/* -inconintr(Ureg *ur) -{ - int x; - x = ((Incon*)0x40700000)->status; -} -*/ diff --git a/gnot/devincon.c b/gnot/devincon.c index e911a9259b2f2a09991bf3dea49f7057d19cef81..bbfcf1161015480820a5ecb8beb6233dd9eea3eb 100644 --- a/gnot/devincon.c +++ b/gnot/devincon.c @@ -133,6 +133,8 @@ static void inconstopen(Queue*, Stream*); static void inconstclose(Queue*); Qinfo inconinfo = { nullput, inconoput, inconstopen, inconstclose, "incon" }; +int incondebug; + /* * set the incon parameters */ @@ -492,6 +494,9 @@ inconoput(Queue *q, Block *bp) ctl = bp->rptr[2]; bp->rptr += 3; + if(incondebug) + print("->(%d)%uo %d\n", chan, ctl, bp->wptr - bp->rptr); + /* * make sure there's an incon out there */ @@ -677,6 +682,8 @@ nextin(Incon *ip, unsigned int c) bp->base[0] = ip->chan; bp->base[1] = ip->chan>>8; bp->base[2] = c; + if(incondebug) + print("<-(%d)%uo %d\n", ip->chan, c, bp->wptr-bp->rptr)-3; next = (ip->wi+3)%Nin; if(next == ip->ri){ @@ -808,3 +815,8 @@ inconintr(Ureg *ur) ip->state = Dead; } } + +incontoggle() +{ + incondebug ^= 1; +} diff --git a/gnot/fns.h b/gnot/fns.h index aa530dcaa06c06be00b7fb505fbf7331e905e156..7fa826dd6081a01f2769ed7ea8570f0042893e9d 100644 --- a/gnot/fns.h +++ b/gnot/fns.h @@ -31,6 +31,7 @@ int devno(int, int); Chan* devopen(Chan*, int, Dirtab*, int, Devgen*); void devstat(Chan*, char*, Dirtab*, int, Devgen*); int devwalk(Chan*, char*, Dirtab*, int, Devgen*); +void duartstartrs232o(void); void duartstarttimer(void); void duartstoptimer(void); void dumpregs(Ureg*); @@ -54,6 +55,7 @@ void freealarm(Alarm*); Block *getb(Blist*); int getfields(char*, char**, int, char); Block *getq(Queue*); +int getrs232o(void); void gotolabel(Label*); void growpte(Orig*, ulong); void *ialloc(ulong, int); @@ -114,6 +116,7 @@ int readnum(ulong, char*, ulong, ulong, int); void ready(Proc*); int return0(void*); void rooterrstr(Error*, char*); +void rs232ichar(int); void qlock(QLock*); void qunlock(QLock*); void restartprint(Alarm*); diff --git a/gnot/proc.c b/gnot/proc.c index 845b0ea59a50adf9ba9366d4d43557a069a42448..50e4b66b912d202a3c2b03d3e13f5b12fe87f011 100644 --- a/gnot/proc.c +++ b/gnot/proc.c @@ -559,9 +559,9 @@ DEBUG() for(i=0; istate != Dead) - print("%d:%s upc %lux %s ut %ld st %ld %lux\n", + print("%d:%s upc %lux %s ut %ld st %ld q %lux r %lux\n", p->pid, p->text, p->pc, statename[p->state], - p->time[0], p->time[1], p->qlock); + p->time[0], p->time[1], p->qlock, p->r); } } diff --git a/gnot/screen.c b/gnot/screen.c index e2960b809609b60e6027f9d8ac117522e69f0516..8f097cfd0607cc2a7d7ea8d9da6d60733b58402f 100644 --- a/gnot/screen.c +++ b/gnot/screen.c @@ -183,7 +183,7 @@ duartinit(void) duart[0].sr_csr = BD4800; /* - * Pen + * RS232 */ duart[1].cmnd = RESET_RCV|DIS_TX|DIS_RX; duart[1].cmnd = RESET_TRANS; @@ -191,7 +191,7 @@ duartinit(void) duart[1].cmnd = RESET_MR; duart[1].mr1_2 = CHAR_ERR|NO_PAR|CBITS8; duart[1].mr1_2 = NORM_OP|ONESTOPB; - duart[1].sr_csr = BD2400; + duart[1].sr_csr = BD9600; /* * Output port @@ -239,6 +239,27 @@ duartstoptimer(void) duart[0].is_imr = IM_IPC|IM_RRDYB|IM_XRDYB|IM_RRDYA; } +void +duartrs232intr(void) +{ + int c; + Duart *duart; + + duart = DUARTREG; + c = getrs232o(); + if(c == -1) + duart[1].cmnd = DIS_TX; + else + duart[1].data = c; +} + +void +duartstartrs232o(void) +{ + DUARTREG[1].cmnd = ENB_TX; + duartrs232intr(); +} + void duartintr(Ureg *ur) { @@ -282,14 +303,19 @@ duartintr(Ureg *ur) /* * Is it 2? */ - if(cause & IM_RRDYB) /* pen input */ + if(cause & IM_RRDYB){ /* rs232 input */ + status = duart[1].sr_csr; c = duart[1].data; + if(status & (FRM_ERR|OVR_ERR|PAR_ERR)) + duart[1].cmnd = RESET_ERR; + else + rs232ichar(c); + } /* * Is it 3? */ - if(cause & IM_XRDYB){ - duart[1].cmnd = DIS_TX; - } + if(cause & IM_XRDYB) /* rs232 output */ + duartrs232intr(); /* * Is it 4? */ diff --git a/gnot/sturp.c b/gnot/sturp.c index 709889dd9801be2d87633525f50b489c84841f94..7d6ba5833e704e9c03cfcc3b89aa2c6a9a200383 100644 --- a/gnot/sturp.c +++ b/gnot/sturp.c @@ -65,7 +65,7 @@ struct Urp { }; #define WINDOW(u) ((u)->unechoed>(u)->next ? (u)->unechoed+(u)->maxout-(u)->next-8 :\ (u)->unechoed+(u)->maxout-(u)->next) -#define IN(x, f, n) (f<=n ? x>=f && x=f) +#define IN(x, f, n) (f<=n ? (x>=f && x=f)) #define NEXT(x) (((x)+1)&Nmask) /* @@ -117,6 +117,7 @@ static void sendrej(Urp*); static void initoutput(Urp*, int); static void initinput(Urp*, int); static void urpkproc(void *arg); +static void urpvomit(char*, Urp*); Qinfo urpinfo = { urpciput, urpoput, urpopen, urpclose, "urp" }; @@ -584,11 +585,14 @@ output(Urp *up) } /* - * fill the transmit buffers + * fill the transmit buffers, `nxb' can never overtake `unechoed' */ q = up->wq; - if(up->xb[up->nxb]==0) { - for(bp=getq(q); bp && up->xb[up->nxb]==0; up->nxb=NEXT(up->nxb)){ + i = NEXT(up->nxb); + if(i != up->unechoed) { + for(bp = getq(q); bp && i!=up->unechoed; i = NEXT(i)){ + if(up->xb[up->nxb] != 0) + urpvomit("output", up); if(BLEN(bp) > up->maxblock){ nbp = up->xb[up->nxb] = allocb(0); nbp->rptr = bp->rptr; @@ -597,17 +601,27 @@ output(Urp *up) up->xb[up->nxb] = bp; bp = getq(q); } + up->nxb = i; } if(bp) putbq(q, bp); } -/* print("output w(%d) up->xb[%d](%ux) up->nxb(%d) up->state(%ux)\n", - WINDOW(up), up->next, up->xb[up->next], up->nxb, up->state); -/**/ + /* - * if a retransmit time has elapsed since a transmit, send an ENQ + * retransmit cruft */ - if(up->unechoed!=up->next && NOW>up->timer){ + if(up->rexmit){ + /* + * if a retransmit is requested, move next back to + * the unacked blocks + */ + up->rexmit = 0; + up->next = up->unacked; + } else if(up->unacked!=up->next && NOW>up->timer){ + /* + * if a retransmit time has elapsed since a transmit, + * send an ENQ + */ up->timer = NOW + MSrexmit; up->state &= ~REJECTING; sendctl(up, ENQ); @@ -618,12 +632,11 @@ output(Urp *up) /* * if there's a window open, push some blocks out + * + * the lock is to synchronize with acknowledges that free + * blocks. */ - if(up->rexmit){ - up->rexmit = 0; - up->next = up->unechoed; - } - while(WINDOW(up)>0 && up->xb[up->next]!=0){ + while(WINDOW(up)>0 && up->next!=up->nxb){ i = up->next; qlock(&up->xl[i]); if(waserror()){ @@ -722,6 +735,10 @@ sendblock(Urp *up, int bn) * message 1, the BOT and the data */ bp = up->xb[bn]; + if(bp == 0){ + urpvomit("sendblock", up); + return; + } m = allocb(1); m->rptr = m->lim - 1; m->wptr = m->lim; @@ -768,6 +785,8 @@ rcvack(Urp *up, int msg) qlock(&up->xl[i]); if(up->xb[i]) freeb(up->xb[i]); + else + urpvomit("rcvack", up); up->xb[i] = 0; qunlock(&up->xl[i]); } @@ -924,3 +943,31 @@ urpkproc(void *arg) up->kstarted = 0; DPRINT("urpkproc %ux\n", up); } + +/* + * urp got very confused, complain + */ +static void +urpvomit(char *msg, Urp* up) +{ + print("urpvomit: %s %ux next %d unechoed %d unacked %d nxb %d\n", + msg, up, up->next, up->unechoed, up->unacked, up->nxb); + print("\txb: %ux %ux %ux %ux %ux %ux %ux %ux\n", + up->xb[0], up->xb[1], up->xb[2], up->xb[3], up->xb[4], + up->xb[5], up->xb[6], up->xb[7]); + print("\tiseq: %uo lastecho: %uo trx: %d trbuf: %uo %uo %uo\n", + up->iseq, up->lastecho, up->trx, up->trbuf[0], up->trbuf[1], + up->trbuf[2]); + print("\tupq: %ux %d %d\n", up->rq->next->first, up->rq->next->nb, + up->rq->next->len); +} + +int +urpdump(void) +{ + Urp *up; + + for(up = urp; up < &urp[Nurp]; up++) + if(up->rq) + urpvomit("", up); +} diff --git a/port/devbit.c b/port/devbit.c index e18ed9a42c5870b4db6196075ea9bb8111ad975e..0a1b4fc1a056bcd1c474d7ed60c4139647eb6f83 100644 --- a/port/devbit.c +++ b/port/devbit.c @@ -128,7 +128,7 @@ Dirtab bitdir[]={ }; #define NBIT (sizeof bitdir/sizeof(Dirtab)) -#define NINFO 150 +#define NINFO 257 void bitreset(void) @@ -277,7 +277,8 @@ bitread(Chan *c, void *va, long n) uchar *p, *q; long miny, maxy, t, x, y; ulong l, nw, ws; - int off; + int off, j; + Fontchar *i; Bitmap *src; if(c->qid & CHDIR) @@ -322,6 +323,9 @@ bitread(Chan *c, void *va, long n) * 'I' 1 * ldepth 1 * rectangle 16 + * if count great enough, also + * font info 3*12 + * fontchars 6*(defont->n+1) */ if(n < 18) error(0, Ebadblt); @@ -331,8 +335,22 @@ bitread(Chan *c, void *va, long n) PLONG(p+6, screen.r.min.y); PLONG(p+10, screen.r.max.x); PLONG(p+14, screen.r.max.y); + if(n >= 18+3*12+6*(defont->n+1)){ + p += 18; + sprint((char*)p, "%11d %11d %11d ", defont->n, + defont->height, defont->ascent); + p += 3*12; + for(i=defont->info,j=0; j<=defont->n; j++,i++,p+=6){ + PSHORT(p, i->x); + p[2] = i->top; + p[3] = i->bottom; + p[4] = i->left; + p[5] = i->width; + } + n = 18+3*12+6*(defont->n+1); + }else + n = 18; bit.init = 0; - n = 18; break; } if(bit.lastid > 0){ diff --git a/port/devcons.c b/port/devcons.c index 9f8673b48cec576916d7874ce813150e3543c64c..93b095159e8a2e7a3785ff4d247435392af788e1 100644 --- a/port/devcons.c +++ b/port/devcons.c @@ -240,7 +240,7 @@ echo(int c) if(c == 0x15) putstrn("^U\n", 3); if(c == 0x16) - dumpqueues(); + urpdump(); else{ ch = c; putstrn(&ch, 1); diff --git a/port/devmnt.c b/port/devmnt.c index 4d9b2c6db0f23f838ce65c1cf9ff478b06d82ee6..04cc2a93b4878bd9ec3a77504a1a97d18fcc1b9b 100644 --- a/port/devmnt.c +++ b/port/devmnt.c @@ -810,6 +810,10 @@ mntxmit(Mnt *m, Mnthdr *mh) wakeup(&w->r); goto Read; } +print("devmnt: undelivered response fid %d type %d\n", mh->rhdr.fid, mh->rhdr.type); +print("reader pid %d fid %d type %d\n", u->p->pid, mh->thdr.fid, mh->thdr.type); +for(w=q->writer; w; w=w->next)print("writer pid %d fid %d type %d\n",w->p->pid,w->thdr.fid,w->thdr.type); + goto Read; }else{ mh->p = u->p; diff --git a/port/devpipe.c b/port/devpipe.c index 2b960dedca066ece5ed18494ff9ec038a4a79cf2..9df787e1136966acee0ec6085c347c853074a742 100644 --- a/port/devpipe.c +++ b/port/devpipe.c @@ -32,6 +32,8 @@ Chan* pipeattach(char *spec) { Chan *c; + int i; + /* * make the first stream */ @@ -59,8 +61,19 @@ pipeclone(Chan *c, Chan *nc) /* * attach it to the first */ + c->stream->devq->ptr = (Stream *)nc->stream; + nc->stream->devq->ptr = (Stream *)c->stream; c->stream->devq->other->next = nc->stream->devq; nc->stream->devq->other->next = c->stream->devq; + + /* + * up the inuse count of each stream to reflect the + * pointer from the other stream. + */ + if(streamenter(c->stream)<0) + panic("pipeattach"); + if(streamenter(nc->stream)<0) + panic("pipeattach"); return nc; } @@ -110,7 +123,17 @@ pipewstat(Chan *c, char *db) void pipeclose(Chan *c) { - streamclose(c); + Stream *other; + + other = (Stream *)c->stream->devq->ptr; + + if(waserror()){ + streamexit(other, 0); + nexterror(); + } + streamclose(c); /* close this stream */ + streamexit(other, 0); /* release stream for other half of pipe */ + poperror(); } long @@ -164,14 +187,7 @@ pipeiput(Queue *q, Block *bp) static void pipeoput(Queue *q, Block *bp) { - lock(q); - if(q->next) - pipeiput(q->next, bp); - else{ - print("pipeoput losing block\n"); - freeb(bp); - } - unlock(q); + PUTNEXT(q, bp); } /* @@ -193,29 +209,7 @@ pipestclose(Queue *q) * send a hangup */ q = q->other; - lock(q); - if(q->next){ - bp = allocb(0); - bp->type = M_HANGUP; - pipeiput(q->next, bp); - } - unlock(q); - - /* - * disconnect (possible livelock?) - */ - for(;;){ - lock(q); - if(q->next){ - if(!canlock(q->next->other)){ - unlock(q); - continue; - } - q->next->other->next = 0; - unlock(q->next->other); - q->next = 0; - } - unlock(q); - break; - } + bp = allocb(0); + bp->type = M_HANGUP; + PUTNEXT(q, bp); } diff --git a/port/proc.c b/port/proc.c index d7e2d4c7685c8ee5fef8597b8f3c0ce118da09d1..cea5e339e6f7a283d898ae22d55b0fff3764745b 100644 --- a/port/proc.c +++ b/port/proc.c @@ -561,9 +561,9 @@ DEBUG() for(i=0; istate != Dead){ - print("%d:%s upc %lux %s ut %ld st %ld\n", + print("%d:%s upc %lux %s ut %ld st %ld r %lux\n", p->pid, p->text, p->pc, statename[p->state], - p->time[0], p->time[1]); + p->time[0], p->time[1], p->r); } } } diff --git a/port/stream.c b/port/stream.c index 16c8e83a0df887775e84c9bbceb5874854f9052d..16ddce2349703f3d938511bb5ec88292e632f8ea 100644 --- a/port/stream.c +++ b/port/stream.c @@ -257,6 +257,22 @@ allocq(Qinfo *qi) return q; } +/* + * flush a queue + */ +static void +flushq(Queue *q) +{ + Block *bp; + + q = RD(q); + while(bp = getq(q)) + freeb(bp); + q = WR(q); + while(bp = getq(q)) + freeb(bp); +} + /* * free a queue */ @@ -515,8 +531,12 @@ prepend(Block *bp, int n) void nullput(Queue *q, Block *bp) { - freeb(bp); - error(0, Ehungup); + if(bp->type == M_HANGUP) + freeb(bp); + else { + freeb(bp); + error(0, Ehungup); + } } /* @@ -658,8 +678,8 @@ streamnew(Chan *c, Qinfo *qi) * hang a device and process q off the stream */ s->inuse = 1; + s->opens = 1; s->hread = 0; - s->tag[0] = 0; q = allocq(&procinfo); s->procq = WR(q); q = allocq(qi); @@ -697,6 +717,7 @@ streamopen(Chan *c, Qinfo *qi) && s->dev == c->dev && s->id == STREAMID(c->qid)){ s->inuse++; + s->opens++; c->stream = s; unlock(s); return; @@ -711,6 +732,50 @@ streamopen(Chan *c, Qinfo *qi) streamnew(c, qi); } +/* + * Enter a stream. Increment the reference count so it can't disappear + * under foot. + */ +int +streamenter(Stream *s) +{ + lock(s); + if(s->opens == 0){ + unlock(s); + return -1; + } + s->inuse++; + unlock(s); + return 0; +} + +/* + * Decrement the reference count on a stream. If the count is + * zero, free the stream. + */ +void +streamexit(Stream *s, int locked) +{ + Queue *q; + Queue *nq; + + if(!locked) + lock(s); + if(s->inuse == 1){ + /* + * ascend the stream freeing the queues + */ + for(q = s->devq; q; q = nq){ + nq = q->next; + freeq(q); + } + s->id = s->dev = s->type = 0; + } + s->inuse--; + if(!locked) + unlock(s); +} + /* * On the last close of a stream, for each queue on the * stream release its blocks and call its close routine. @@ -729,33 +794,35 @@ streamclose(Chan *c) return; /* - * decrement the reference cound + * decrement the reference count */ lock(s); - if(s->inuse != 1){ - s->inuse--; - unlock(c->stream); - return; + if(s->opens == 1){ + /* + * descend the stream closing the queues + */ + for(q = s->procq; q; q = q->next){ + if(q->info->close) + (*q->info->close)(q->other); + /* this may be 2 streams joined device end to device end */ + if(q == s->devq->other) + break; + } + + /* + * ascend the stream flushing the queues + */ + for(q = s->devq; q; q = nq){ + nq = q->next; + flushq(q); + } } + s->opens--; /* - * descend the stream closing the queues + * leave it and free it */ - for(q = s->procq; q; q = q->next){ - if(q->info->close) - (*q->info->close)(q->other); - if(q == s->devq->other) - break; - } - /* - * ascend the stream freeing the queues - */ - for(q = s->devq; q; q = nq){ - nq = q->next; - freeq(q); - } - s->id = s->dev = s->type = 0; - s->inuse--; + streamexit(s, 1); unlock(s); } diff --git a/port/sturp.c b/port/sturp.c index 709889dd9801be2d87633525f50b489c84841f94..7d6ba5833e704e9c03cfcc3b89aa2c6a9a200383 100644 --- a/port/sturp.c +++ b/port/sturp.c @@ -65,7 +65,7 @@ struct Urp { }; #define WINDOW(u) ((u)->unechoed>(u)->next ? (u)->unechoed+(u)->maxout-(u)->next-8 :\ (u)->unechoed+(u)->maxout-(u)->next) -#define IN(x, f, n) (f<=n ? x>=f && x=f) +#define IN(x, f, n) (f<=n ? (x>=f && x=f)) #define NEXT(x) (((x)+1)&Nmask) /* @@ -117,6 +117,7 @@ static void sendrej(Urp*); static void initoutput(Urp*, int); static void initinput(Urp*, int); static void urpkproc(void *arg); +static void urpvomit(char*, Urp*); Qinfo urpinfo = { urpciput, urpoput, urpopen, urpclose, "urp" }; @@ -584,11 +585,14 @@ output(Urp *up) } /* - * fill the transmit buffers + * fill the transmit buffers, `nxb' can never overtake `unechoed' */ q = up->wq; - if(up->xb[up->nxb]==0) { - for(bp=getq(q); bp && up->xb[up->nxb]==0; up->nxb=NEXT(up->nxb)){ + i = NEXT(up->nxb); + if(i != up->unechoed) { + for(bp = getq(q); bp && i!=up->unechoed; i = NEXT(i)){ + if(up->xb[up->nxb] != 0) + urpvomit("output", up); if(BLEN(bp) > up->maxblock){ nbp = up->xb[up->nxb] = allocb(0); nbp->rptr = bp->rptr; @@ -597,17 +601,27 @@ output(Urp *up) up->xb[up->nxb] = bp; bp = getq(q); } + up->nxb = i; } if(bp) putbq(q, bp); } -/* print("output w(%d) up->xb[%d](%ux) up->nxb(%d) up->state(%ux)\n", - WINDOW(up), up->next, up->xb[up->next], up->nxb, up->state); -/**/ + /* - * if a retransmit time has elapsed since a transmit, send an ENQ + * retransmit cruft */ - if(up->unechoed!=up->next && NOW>up->timer){ + if(up->rexmit){ + /* + * if a retransmit is requested, move next back to + * the unacked blocks + */ + up->rexmit = 0; + up->next = up->unacked; + } else if(up->unacked!=up->next && NOW>up->timer){ + /* + * if a retransmit time has elapsed since a transmit, + * send an ENQ + */ up->timer = NOW + MSrexmit; up->state &= ~REJECTING; sendctl(up, ENQ); @@ -618,12 +632,11 @@ output(Urp *up) /* * if there's a window open, push some blocks out + * + * the lock is to synchronize with acknowledges that free + * blocks. */ - if(up->rexmit){ - up->rexmit = 0; - up->next = up->unechoed; - } - while(WINDOW(up)>0 && up->xb[up->next]!=0){ + while(WINDOW(up)>0 && up->next!=up->nxb){ i = up->next; qlock(&up->xl[i]); if(waserror()){ @@ -722,6 +735,10 @@ sendblock(Urp *up, int bn) * message 1, the BOT and the data */ bp = up->xb[bn]; + if(bp == 0){ + urpvomit("sendblock", up); + return; + } m = allocb(1); m->rptr = m->lim - 1; m->wptr = m->lim; @@ -768,6 +785,8 @@ rcvack(Urp *up, int msg) qlock(&up->xl[i]); if(up->xb[i]) freeb(up->xb[i]); + else + urpvomit("rcvack", up); up->xb[i] = 0; qunlock(&up->xl[i]); } @@ -924,3 +943,31 @@ urpkproc(void *arg) up->kstarted = 0; DPRINT("urpkproc %ux\n", up); } + +/* + * urp got very confused, complain + */ +static void +urpvomit(char *msg, Urp* up) +{ + print("urpvomit: %s %ux next %d unechoed %d unacked %d nxb %d\n", + msg, up, up->next, up->unechoed, up->unacked, up->nxb); + print("\txb: %ux %ux %ux %ux %ux %ux %ux %ux\n", + up->xb[0], up->xb[1], up->xb[2], up->xb[3], up->xb[4], + up->xb[5], up->xb[6], up->xb[7]); + print("\tiseq: %uo lastecho: %uo trx: %d trbuf: %uo %uo %uo\n", + up->iseq, up->lastecho, up->trx, up->trbuf[0], up->trbuf[1], + up->trbuf[2]); + print("\tupq: %ux %d %d\n", up->rq->next->first, up->rq->next->nb, + up->rq->next->len); +} + +int +urpdump(void) +{ + Urp *up; + + for(up = urp; up < &urp[Nurp]; up++) + if(up->rq) + urpvomit("", up); +} diff --git a/power/conf.h b/power/conf.h index 89d3acea340d8490b7fdc347b6e5b1e3fe110f03..45f1fae33f39d9e426692a486b2ff955f6f2bbe6 100644 --- a/power/conf.h +++ b/power/conf.h @@ -21,6 +21,8 @@ Conftab conftab[] = { {"nqueue", &conf.nqueue }, {"nblock", &conf.nblock }, {"nsrv", &conf.nsrv }, + {"nnoifc", &conf.nnoifc }, + {"nnoconv", &conf.nnoconv }, { 0, 0 }, }; diff --git a/power/dat.h b/power/dat.h index bcfe16b58d518561e20b5ca7cb6ef978066ba345..fe48f81301061969c9ec460998dad4b0d13ea5b9 100644 --- a/power/dat.h +++ b/power/dat.h @@ -17,7 +17,12 @@ typedef struct Lock Lock; typedef struct Mach Mach; typedef struct Mount Mount; typedef struct Mtab Mtab; +typedef struct Noconv Noconv; +typedef struct Nohdr Nohdr; +typedef struct Noifc Noifc; typedef struct Note Note; +typedef struct Nomsg Nomsg; +typedef struct Nocall Nocall; typedef struct Orig Orig; typedef struct PTE PTE; typedef struct Page Page; @@ -137,6 +142,8 @@ struct Conf ulong nqueue; /* stream queues */ ulong nblock; /* stream blocks */ ulong nsrv; /* public servers (devsrv.c) */ + ulong nnoifc; /* number of nonet interfaces */ + ulong nnoconv; /* number of nonet conversations/ifc */ }; struct Dev @@ -435,7 +442,8 @@ struct Queue { */ struct Stream { Lock; /* structure lock */ - int inuse; /* use count */ + int inuse; /* number of processes in stream */ + int opens; /* number of processes with stream open */ int hread; /* number of reads after hangup */ int type; /* correclation with Chan */ int dev; /* ... */ @@ -444,7 +452,6 @@ struct Stream { QLock wrlock; /* write lock */ Queue *procq; /* write queue at process end */ Queue *devq; /* read queue at device end */ - char tag[32]; /* when reading the tag qid */ }; #define RD(q) ((q)->other < (q) ? (q->other) : q) #define WR(q) ((q)->other > (q) ? (q->other) : q) @@ -469,6 +476,119 @@ enum { Streambhi= 16, /* block count high water mark */ }; +/* + * nonet constants + */ +enum { + Nnomsg = 128, /* max number of outstanding messages */ + Nnocalls = 5, /* maximum queued incoming calls */ +}; + +/* + * generic nonet header + */ +struct Nohdr { + uchar circuit[3]; /* circuit number */ + uchar flag; + uchar mid; /* message id */ + uchar ack; /* piggy back ack */ + uchar remain[2]; /* count of remaing bytes of data */ + uchar sum[2]; /* checksum (0 means none) */ +}; +#define HDRSIZE 10 +#define NEWCALL 0x1 /* flag bit marking a new circuit */ +#define HANGUP 0x2 /* flag bit requesting hangup */ +#define ACKME 0x4 /* acknowledge this message */ + +/* + * a buffer describing a nonet message + */ +struct Nomsg { + QLock; + Blist; + int mid; /* sequence number */ + int rem; /* remaining */ + long time; + int acked; +}; + +/* + * one exists for each Nonet conversation. + */ +struct Noconv { + QLock; + + Queue *rq; /* input queue */ + int version; /* incremented each time struct is changed */ + int state; /* true if listening */ + + Nomsg in[Nnomsg]; /* messages being received */ + int rcvcircuit; /* circuit number of incoming packets */ + + uchar ack[Nnomsg]; /* acknowledgements waiting to be sent */ + long atime[Nnomsg]; + int afirst; + int anext; + + QLock xlock; /* one trasmitter at a time */ + Rendez r; /* process waiting for an output mid */ + Nomsg ctl; /* for control messages */ + Nomsg out[Nnomsg]; /* messages being sent */ + int first; /* first unacknowledged message */ + int next; /* next message buffer to use */ + int lastacked; /* last message acked */ + Block *media; /* prototype media output header */ + Nohdr *hdr; /* nonet header inside of media header */ + + Noifc *ifc; + int kstarted; + char raddr[64]; /* remote address */ + int rexmit; /* statistics */ + int retry; + int bad; + int sent; + int rcvd; +}; + +/* + * an incoming call + */ +struct Nocall { + Block *msg; + char raddr[64]; + long circuit; +}; + +/* + * a nonet interface. one exists for every stream that a + * nonet multiplexor is pushed onto. + */ +struct Noifc { + Lock; + int ref; + char name[64]; /* interface name */ + Queue *wq; /* interface output queue */ + Noconv *conv; + + /* + * media dependent + */ + int maxtu; /* maximum transfer unit */ + int mintu; /* minimum transfer unit */ + int hsize; /* media header size */ + void (*connect)(Noconv *, char *); + + /* + * calls and listeners + */ + QLock listenl; + Rendez listenr; + Lock lock; + Nocall call[Nnocalls]; + int rptr; + int wptr; +}; + #define PRINTSIZE 256 struct { diff --git a/power/fns.h b/power/fns.h index dd0f8de4a7fd03db6022ae832a162d05bcbbc020..d41370dbf7ea1556b2c61aa5e7ad4f71d31e1b04 100644 --- a/power/fns.h +++ b/power/fns.h @@ -108,6 +108,10 @@ void newqinfo(Qinfo*); char *nextelem(char*, char*); void newstart(void); int newtlbpid(Proc*); +int nonetcksum(Block*, int); +void nonetfreeifc(Noifc*); +Noifc* nonetnewifc(Queue*, Stream*, int, int, int, void (*)(Noconv*, char*)); +void nonetrcvmsg(Noconv*, Block*); void novme(int); void nullput(Queue*, Block*); void online(void); @@ -158,6 +162,8 @@ int spllo(void); void splx(int); Devgen streamgen; void streamclose(Chan*); +int streamenter(Stream*); +void streamexit(Stream*, int); void streaminit(void); long streamread(Chan*, void*, long); long streamwrite(Chan*, void*, long, int);