![[BACK]](/cvsweb/icons/back.gif){kind=icon}
Return to stream.c CVS log ![[TXT]](/cvsweb/icons/text.gif){kind=icon}
![[DIR]](/cvsweb/icons/dir.gif){kind=icon}
Up to [Plan 9 NeXT] / lucent / sys / src / 9 / port|
Return to stream.c CVS log | Up to [Plan 9 NeXT] / lucent / sys / src / 9 / port |
1.1 ! root 1: #include "u.h" ! 2: #include "../port/lib.h" ! 3: #include "mem.h" ! 4: #include "dat.h" ! 5: #include "fns.h" ! 6: #include "io.h" ! 7: #include "../port/error.h" ! 8: #include "devtab.h" ! 9: ! 10: /* ! 11: * Part 1) Blocks ! 12: */ ! 13: ! 14: /* ! 15: * Allocate a block. Put the data portion at the end of the smalloc'd ! 16: * chunk so that it can easily grow from the front to add protocol ! 17: * headers. Thank Larry Peterson for the suggestion. ! 18: */ ! 19: Block * ! 20: allocb(ulong size) ! 21: { ! 22: Block *bp; ! 23: uchar *base, *lim; ! 24: int n; ! 25: ! 26: bp = smalloc(sizeof(Block)+size); ! 27: ! 28: base = (uchar*)bp + sizeof(Block); ! 29: n = msize(bp); ! 30: lim = (uchar*)bp + n; ! 31: n -= size + sizeof(Block); ! 32: if(n > 0) ! 33: memset(lim - n, 0, n); ! 34: bp->wptr = bp->rptr = lim - size; ! 35: bp->base = base; ! 36: bp->lim = lim; ! 37: bp->flags = 0; ! 38: bp->next = 0; ! 39: bp->list = 0; ! 40: bp->type = M_DATA; ! 41: bp->pc = getcallerpc(((uchar*)&size) - sizeof(size)); ! 42: return bp; ! 43: } ! 44: ! 45: /* ! 46: * Free a block (or list of blocks). Poison its pointers so that ! 47: * someone trying to access it after freeing will cause a panic. ! 48: */ ! 49: void ! 50: freeb(Block *bp) ! 51: { ! 52: Block *next; ! 53: ! 54: while(bp){ ! 55: bp->rptr = 0; ! 56: bp->wptr = 0; ! 57: next = bp->next; ! 58: free(bp); ! 59: bp = next; ! 60: } ! 61: } ! 62: ! 63: /* ! 64: * Pad a block to the front with n bytes. This is used to add protocol ! 65: * headers to the front of blocks. ! 66: */ ! 67: Block * ! 68: padb(Block *bp, int n) ! 69: { ! 70: Block *nbp; ! 71: ! 72: if(bp->base && bp->rptr-bp->base>=n){ ! 73: bp->rptr -= n; ! 74: return bp; ! 75: } else { ! 76: nbp = allocb(n); ! 77: nbp->wptr = nbp->lim; ! 78: nbp->rptr = nbp->wptr - n; ! 79: nbp->next = bp; ! 80: return nbp; ! 81: } ! 82: } ! 83: ! 84: /* ! 85: * make sure the first block has n bytes ! 86: */ ! 87: Block * ! 88: pullup(Block *bp, int n) ! 89: { ! 90: Block *nbp; ! 91: int i; ! 92: ! 93: /* ! 94: * this should almost always be true, the rest it ! 95: * just for to avoid every caller checking. ! 96: */ ! 97: if(BLEN(bp) >= n) ! 98: return bp; ! 99: ! 100: /* ! 101: * if not enough room in the first block, ! 102: * add another to the front of the list. ! 103: */ ! 104: if(bp->lim - bp->rptr < n){ ! 105: nbp = allocb(n); ! 106: nbp->next = bp; ! 107: bp = nbp; ! 108: } ! 109: ! 110: /* ! 111: * copy bytes from the trailing blocks into the first ! 112: */ ! 113: n -= BLEN(bp); ! 114: while(nbp = bp->next){ ! 115: i = BLEN(nbp); ! 116: if(i >= n) { ! 117: memmove(bp->wptr, nbp->rptr, n); ! 118: bp->wptr += n; ! 119: nbp->rptr += n; ! 120: return bp; ! 121: } else { ! 122: memmove(bp->wptr, nbp->rptr, i); ! 123: bp->wptr += i; ! 124: bp->next = nbp->next; ! 125: nbp->next = 0; ! 126: freeb(nbp); ! 127: n -= i; ! 128: } ! 129: } ! 130: freeb(bp); ! 131: return 0; ! 132: } ! 133: ! 134: /* ! 135: * return the number of data bytes of a list of blocks ! 136: */ ! 137: int ! 138: blen(Block *bp) ! 139: { ! 140: int len; ! 141: ! 142: len = 0; ! 143: while(bp) { ! 144: len += BLEN(bp); ! 145: bp = bp->next; ! 146: } ! 147: ! 148: return len; ! 149: } ! 150: ! 151: /* ! 152: * round a block chain to some even number of bytes. Used ! 153: * by devip.c becuase all IP packets must have an even number ! 154: * of bytes. ! 155: * ! 156: * The last block in the returned chain will have S_DELIM set. ! 157: */ ! 158: int ! 159: bround(Block *bp, int amount) ! 160: { ! 161: Block *last; ! 162: int len, pad; ! 163: ! 164: len = 0; ! 165: SET(last); /* Ken's magic */ ! 166: ! 167: while(bp) { ! 168: len += BLEN(bp); ! 169: last = bp; ! 170: bp = bp->next; ! 171: } ! 172: ! 173: pad = ((len + amount) & ~amount) - len; ! 174: if(pad) { ! 175: if(last->lim - last->wptr >= pad){ ! 176: memset(last->wptr, 0, pad); ! 177: last->wptr += pad; ! 178: } else { ! 179: last->next = allocb(pad); ! 180: last->flags &= ~S_DELIM; ! 181: last = last->next; ! 182: last->wptr += pad; ! 183: last->flags |= S_DELIM; ! 184: } ! 185: } ! 186: ! 187: return len + pad; ! 188: } ! 189: ! 190: /* ! 191: * expand a block list to be one block, len bytes long. used by ! 192: * ethernet routines. ! 193: */ ! 194: Block* ! 195: expandb(Block *bp, int len) ! 196: { ! 197: Block *nbp, *new; ! 198: int i; ! 199: ulong delim = 0; ! 200: ! 201: new = allocb(len); ! 202: if(new == 0){ ! 203: freeb(bp); ! 204: return 0; ! 205: } ! 206: ! 207: /* ! 208: * copy bytes into new block ! 209: */ ! 210: for(nbp = bp; len>0 && nbp; nbp = nbp->next){ ! 211: delim = nbp->flags & S_DELIM; ! 212: i = BLEN(nbp); ! 213: if(i > len) { ! 214: memmove(new->wptr, nbp->rptr, len); ! 215: new->wptr += len; ! 216: break; ! 217: } else { ! 218: memmove(new->wptr, nbp->rptr, i); ! 219: new->wptr += i; ! 220: len -= i; ! 221: } ! 222: } ! 223: if(len){ ! 224: memset(new->wptr, 0, len); ! 225: new->wptr += len; ! 226: } ! 227: new->flags |= delim; ! 228: freeb(bp); ! 229: return new; ! 230: ! 231: } ! 232: ! 233: /* ! 234: * make a copy of the first 'count' bytes of a block chain. Use ! 235: * by transport protocols. ! 236: */ ! 237: Block * ! 238: copyb(Block *bp, int count) ! 239: { ! 240: Block *nb, *head, **p; ! 241: int l; ! 242: ! 243: p = &head; ! 244: while(count) { ! 245: l = BLEN(bp); ! 246: if(count < l) ! 247: l = count; ! 248: nb = allocb(l); ! 249: if(nb == 0) ! 250: panic("copyb.1"); ! 251: memmove(nb->wptr, bp->rptr, l); ! 252: nb->wptr += l; ! 253: count -= l; ! 254: if(bp->flags & S_DELIM) ! 255: nb->flags |= S_DELIM; ! 256: *p = nb; ! 257: p = &nb->next; ! 258: bp = bp->next; ! 259: if(bp == 0) ! 260: break; ! 261: } ! 262: if(count) { ! 263: nb = allocb(count); ! 264: if(nb == 0) ! 265: panic("copyb.2"); ! 266: memset(nb->wptr, 0, count); ! 267: nb->wptr += count; ! 268: nb->flags |= S_DELIM; ! 269: *p = nb; ! 270: } ! 271: if(blen(head) == 0) ! 272: print("copyb: zero length\n"); ! 273: ! 274: return head; ! 275: } ! 276: ! 277: /* ! 278: * Part 2) Queues ! 279: */ ! 280: ! 281: /* ! 282: * process end line discipline ! 283: */ ! 284: static Streamput stputq; ! 285: Qinfo procinfo = ! 286: { ! 287: stputq, ! 288: nullput, ! 289: 0, ! 290: 0, ! 291: "process" ! 292: }; ! 293: ! 294: /* ! 295: * line disciplines that can be pushed ! 296: */ ! 297: static Qinfo *lds; ! 298: ! 299: /* ! 300: * make known a stream module and call its initialization routine, if ! 301: * it has one. ! 302: */ ! 303: void ! 304: newqinfo(Qinfo *qi) ! 305: { ! 306: if(qi->next) ! 307: panic("newqinfo: already configured"); ! 308: ! 309: qi->next = lds; ! 310: lds = qi; ! 311: if(qi->reset) ! 312: (*qi->reset)(); ! 313: } ! 314: ! 315: /* ! 316: * find the info structure for line discipline 'name' ! 317: */ ! 318: Qinfo * ! 319: qinfofind(char *name) ! 320: { ! 321: Qinfo *qi; ! 322: ! 323: if(name == 0) ! 324: return 0; ! 325: for(qi = lds; qi; qi = qi->next) ! 326: if(strcmp(qi->name, name)==0) ! 327: return qi; ! 328: return 0; ! 329: } ! 330: ! 331: /* ! 332: * allocate a pair of queues. flavor them with the requested put routines. ! 333: * the `QINUSE' flag on the read side is the only one used. ! 334: */ ! 335: static Queue * ! 336: allocq(Qinfo *qi) ! 337: { ! 338: Queue *q, *wq; ! 339: ! 340: q = smalloc(2*sizeof(Queue)); ! 341: ! 342: q->flag = QINUSE; ! 343: q->r.p = 0; ! 344: q->info = qi; ! 345: q->put = qi->iput; ! 346: q->len = q->nb = 0; ! 347: q->ptr = 0; ! 348: q->rp = &q->r; ! 349: wq = q->other = q + 1; ! 350: ! 351: wq->flag = QINUSE; ! 352: wq->r.p = 0; ! 353: wq->info = qi; ! 354: wq->put = qi->oput; ! 355: wq->other = q; ! 356: wq->ptr = 0; ! 357: wq->len = wq->nb = 0; ! 358: wq->rp = &wq->r; ! 359: ! 360: return q; ! 361: } ! 362: ! 363: /* ! 364: * free a queue ! 365: */ ! 366: static void ! 367: freeq(Queue *q) ! 368: { ! 369: Block *bp; ! 370: ! 371: q = RD(q); ! 372: while(bp = getq(q)) ! 373: freeb(bp); ! 374: q = WR(q); ! 375: while(bp = getq(q)) ! 376: freeb(bp); ! 377: free(RD(q)); ! 378: } ! 379: ! 380: /* ! 381: * flush a queue ! 382: */ ! 383: static void ! 384: flushq(Queue *q) ! 385: { ! 386: Block *bp; ! 387: ! 388: q = RD(q); ! 389: while(bp = getq(q)) ! 390: freeb(bp); ! 391: q = WR(q); ! 392: while(bp = getq(q)) ! 393: freeb(bp); ! 394: } ! 395: ! 396: /* ! 397: * push a queue onto a stream referenced by the proc side write q ! 398: */ ! 399: Queue * ! 400: pushq(Stream* s, Qinfo *qi) ! 401: { ! 402: Queue *q; ! 403: Queue *nq; ! 404: ! 405: q = RD(s->procq); ! 406: ! 407: /* ! 408: * make the new queue ! 409: */ ! 410: nq = allocq(qi); ! 411: ! 412: /* ! 413: * push ! 414: */ ! 415: qlock(s); ! 416: RD(nq)->next = q; ! 417: RD(WR(q)->next)->next = RD(nq); ! 418: WR(nq)->next = WR(q)->next; ! 419: WR(q)->next = WR(nq); ! 420: qunlock(s); ! 421: ! 422: if(qi->open) ! 423: (*qi->open)(RD(nq), s); ! 424: ! 425: return WR(nq)->next; ! 426: } ! 427: ! 428: /* ! 429: * pop off the top line discipline ! 430: */ ! 431: static void ! 432: popq(Stream *s) ! 433: { ! 434: Queue *q; ! 435: ! 436: if(waserror()){ ! 437: qunlock(s); ! 438: nexterror(); ! 439: } ! 440: qlock(s); ! 441: if(s->procq->next == WR(s->devq)) ! 442: error(Ebadld); ! 443: q = s->procq->next; ! 444: if(q->info->close) ! 445: (*q->info->close)(RD(q)); ! 446: s->procq->next = q->next; ! 447: RD(q->next)->next = RD(s->procq); ! 448: qunlock(s); ! 449: freeq(q); ! 450: } ! 451: ! 452: /* ! 453: * add a block (or list of blocks) to the end of a queue. return true ! 454: * if one of the blocks contained a delimiter. ! 455: */ ! 456: int ! 457: putq(Queue *q, Block *bp) ! 458: { ! 459: int delim; ! 460: ! 461: lock(q); ! 462: if(q->first) ! 463: q->last->next = bp; ! 464: else ! 465: q->first = bp; ! 466: q->len += BLEN(bp); ! 467: q->nb++; ! 468: delim = bp->flags & S_DELIM; ! 469: while(bp->next) { ! 470: bp = bp->next; ! 471: q->len += BLEN(bp); ! 472: q->nb++; ! 473: delim |= bp->flags & S_DELIM; ! 474: } ! 475: q->last = bp; ! 476: if(q->len >= Streamhi || q->nb >= Streambhi) ! 477: q->flag |= QHIWAT; ! 478: unlock(q); ! 479: return delim; ! 480: } ! 481: ! 482: int ! 483: putb(Blist *q, Block *bp) ! 484: { ! 485: int delim; ! 486: ! 487: if(q->first) ! 488: q->last->next = bp; ! 489: else ! 490: q->first = bp; ! 491: q->len += BLEN(bp); ! 492: delim = bp->flags & S_DELIM; ! 493: while(bp->next) { ! 494: bp = bp->next; ! 495: q->len += BLEN(bp); ! 496: delim |= bp->flags & S_DELIM; ! 497: } ! 498: q->last = bp; ! 499: return delim; ! 500: } ! 501: ! 502: /* ! 503: * add a block to the start of a queue ! 504: */ ! 505: void ! 506: putbq(Blist *q, Block *bp) ! 507: { ! 508: lock(q); ! 509: if(q->first) ! 510: bp->next = q->first; ! 511: else ! 512: q->last = bp; ! 513: q->first = bp; ! 514: q->len += BLEN(bp); ! 515: q->nb++; ! 516: unlock(q); ! 517: } ! 518: ! 519: /* ! 520: * remove the first block from a queue ! 521: */ ! 522: Block * ! 523: getq(Queue *q) ! 524: { ! 525: Block *bp; ! 526: ! 527: lock(q); ! 528: bp = q->first; ! 529: if(bp) { ! 530: q->first = bp->next; ! 531: if(q->first == 0) ! 532: q->last = 0; ! 533: q->len -= BLEN(bp); ! 534: q->nb--; ! 535: if((q->flag&QHIWAT) && q->len<Streamhi/2 && q->nb<Streambhi/2 &&q->other){ ! 536: wakeup(q->other->next->other->rp); ! 537: q->flag &= ~QHIWAT; ! 538: } ! 539: bp->next = 0; ! 540: } ! 541: unlock(q); ! 542: return bp; ! 543: } ! 544: ! 545: /* ! 546: * grab all the blocks in a queue ! 547: */ ! 548: Block * ! 549: grabq(Queue *q) ! 550: { ! 551: Block *bp; ! 552: ! 553: lock(q); ! 554: bp = q->first; ! 555: if(bp){ ! 556: q->first = 0; ! 557: q->last = 0; ! 558: q->len = 0; ! 559: q->nb = 0; ! 560: if(q->flag&QHIWAT){ ! 561: wakeup(q->other->next->other->rp); ! 562: q->flag &= ~QHIWAT; ! 563: } ! 564: } ! 565: unlock(q); ! 566: return bp; ! 567: } ! 568: ! 569: /* ! 570: * remove the first block from a list of blocks ! 571: */ ! 572: Block * ! 573: getb(Blist *q) ! 574: { ! 575: Block *bp; ! 576: ! 577: bp = q->first; ! 578: if(bp) { ! 579: q->first = bp->next; ! 580: if(q->first == 0) ! 581: q->last = 0; ! 582: q->len -= BLEN(bp); ! 583: bp->next = 0; ! 584: } ! 585: return bp; ! 586: } ! 587: ! 588: ! 589: /* ! 590: * put a block into the bit bucket ! 591: */ ! 592: void ! 593: nullput(Queue *q, Block *bp) ! 594: { ! 595: USED(q); ! 596: if(bp->type == M_HANGUP) ! 597: freeb(bp); ! 598: else { ! 599: freeb(bp); ! 600: error(Ehungup); ! 601: } ! 602: } ! 603: ! 604: /* ! 605: * Part 3) Streams ! 606: */ ! 607: ! 608: /* ! 609: * the per stream directory structure ! 610: */ ! 611: Dirtab streamdir[]={ ! 612: "data", {Sdataqid}, 0, 0600, ! 613: "ctl", {Sctlqid}, 0, 0600, ! 614: }; ! 615: ! 616: /* ! 617: * hash buckets containing all streams ! 618: */ ! 619: enum ! 620: { ! 621: Nbits= 5, ! 622: Nhash= 1<<Nbits, ! 623: Nmask= Nhash-1, ! 624: }; ! 625: typedef struct Sthash Sthash; ! 626: struct Sthash ! 627: { ! 628: QLock; ! 629: Stream *s; ! 630: }; ! 631: static Sthash ht[Nhash]; ! 632: ! 633: static void hangup(Stream*); ! 634: ! 635: /* ! 636: * A stream device consists of the contents of streamdir plus ! 637: * any directory supplied by the actual device. ! 638: * ! 639: * values of s: ! 640: * 0 to ntab-1 apply to the auxiliary directory. ! 641: * ntab to ntab+Shighqid-Slowqid+1 apply to streamdir. ! 642: */ ! 643: int ! 644: streamgen(Chan *c, Dirtab *tab, int ntab, int s, Dir *dp) ! 645: { ! 646: if(s < ntab) ! 647: tab = &tab[s]; ! 648: else if(s < ntab + Shighqid - Slowqid + 1) ! 649: tab = &streamdir[s - ntab]; ! 650: else ! 651: return -1; ! 652: ! 653: devdir(c, (Qid){STREAMQID(STREAMID(c->qid.path),tab->qid.path), 0}, ! 654: tab->name, tab->length, eve, tab->perm, dp); ! 655: return 1; ! 656: } ! 657: ! 658: /* ! 659: * return a hash bucket for a stream ! 660: */ ! 661: static Sthash* ! 662: hash(int type, int dev, int id) ! 663: { ! 664: return &ht[(type*7*7 + dev*7 + id) & Nmask]; ! 665: } ! 666: ! 667: /* ! 668: * create a new stream, if noopen is non-zero, don't increment the open count ! 669: */ ! 670: Stream * ! 671: streamnew(ushort type, ushort dev, ulong id, Qinfo *qi, int noopen) ! 672: { ! 673: Stream *s; ! 674: Queue *q; ! 675: Sthash *hb; ! 676: ! 677: hb = hash(type, dev, id); ! 678: ! 679: /* ! 680: * if the stream already exists, just increment the reference counts. ! 681: */ ! 682: qlock(hb); ! 683: for(s = hb->s; s; s = s->next) { ! 684: if(s->type == type && s->dev == dev && s->id == id){ ! 685: s->inuse++; ! 686: qunlock(hb); ! 687: if(noopen == 0){ ! 688: qlock(s); ! 689: s->opens++; ! 690: qunlock(s); ! 691: } ! 692: return s; ! 693: } ! 694: } ! 695: ! 696: /* ! 697: * create and init a new stream ! 698: */ ! 699: s = smalloc(sizeof(Stream)); ! 700: s->inuse = 1; ! 701: s->type = type; ! 702: s->dev = dev; ! 703: s->id = id; ! 704: s->err = 0; ! 705: s->hread = 0; ! 706: s->next = hb->s; ! 707: hb->s = s; ! 708: ! 709: /* ! 710: * The ordering of these 2 instructions is very important. ! 711: * It makes sure we finish the stream initialization before ! 712: * anyone else can access it. ! 713: */ ! 714: qlock(s); ! 715: qunlock(hb); ! 716: ! 717: if(waserror()){ ! 718: qunlock(s); ! 719: streamclose1(s); ! 720: nexterror(); ! 721: } ! 722: ! 723: /* ! 724: * hang a device and process q off the stream ! 725: */ ! 726: if(noopen) ! 727: s->opens = 0; ! 728: else ! 729: s->opens = 1; ! 730: q = allocq(&procinfo); ! 731: WR(q)->ptr = s; ! 732: RD(q)->ptr = s; ! 733: s->procq = WR(q); ! 734: q = allocq(qi); ! 735: s->devq = RD(q); ! 736: WR(s->procq)->next = WR(s->devq); ! 737: RD(s->procq)->next = 0; ! 738: RD(s->devq)->next = RD(s->procq); ! 739: WR(s->devq)->next = 0; ! 740: ! 741: if(qi->open) ! 742: (*qi->open)(RD(s->devq), s); ! 743: ! 744: qunlock(s); ! 745: poperror(); ! 746: return s; ! 747: } ! 748: ! 749: /* ! 750: * Associate a stream with a channel ! 751: */ ! 752: void ! 753: streamopen(Chan *c, Qinfo *qi) ! 754: { ! 755: c->stream = streamnew(c->type, c->dev, STREAMID(c->qid.path), qi, 0); ! 756: } ! 757: ! 758: /* ! 759: * Enter a stream only if the stream exists and is open. Increment the ! 760: * reference count so it can't disappear under foot. ! 761: * ! 762: * Return -1 if the stream no longer exists or is not opened. ! 763: */ ! 764: int ! 765: streamenter(Stream *s) ! 766: { ! 767: Sthash *hb; ! 768: Stream *ns; ! 769: ! 770: hb = hash(s->type, s->dev, s->id); ! 771: qlock(hb); ! 772: for(ns = hb->s; ns; ns = ns->next) ! 773: if(s->type == ns->type && s->dev == ns->dev && s->id == ns->id){ ! 774: s->inuse++; ! 775: qunlock(hb); ! 776: if(s->opens == 0){ ! 777: streamexit(s); ! 778: return -1; ! 779: } ! 780: return 0; ! 781: } ! 782: qunlock(hb); ! 783: return -1; ! 784: } ! 785: ! 786: /* ! 787: * Decrement the reference count on a stream. If the count is ! 788: * zero, free the stream. ! 789: */ ! 790: void ! 791: streamexit(Stream *s) ! 792: { ! 793: Queue *q; ! 794: Queue *nq; ! 795: Sthash *hb; ! 796: Stream **l, *ns; ! 797: ! 798: hb = hash(s->type, s->dev, s->id); ! 799: qlock(hb); ! 800: if(s->inuse-- == 1){ ! 801: if(s->opens != 0) ! 802: panic("streamexit %d %s\n", s->opens, s->devq->info->name); ! 803: ! 804: /* ! 805: * ascend the stream freeing the queues ! 806: */ ! 807: for(q = s->devq; q; q = nq){ ! 808: nq = q->next; ! 809: freeq(q); ! 810: } ! 811: if(s->err) ! 812: freeb(s->err); ! 813: ! 814: /* ! 815: * unchain it from the hash bucket and free ! 816: */ ! 817: l = &hb->s; ! 818: for(ns = hb->s; ns; ns = ns->next){ ! 819: if(s == ns){ ! 820: *l = s->next; ! 821: break; ! 822: } ! 823: l = &ns->next; ! 824: } ! 825: free(s); ! 826: } ! 827: qunlock(hb); ! 828: } ! 829: ! 830: /* ! 831: * nail down a stream so that it can't be closed ! 832: */ ! 833: void ! 834: naildownstream(Stream *s) ! 835: { ! 836: s->opens++; ! 837: s->inuse++; ! 838: } ! 839: ! 840: /* ! 841: * Decrement the open count. When it goes to zero, call the close ! 842: * routines for each queue in the stream. ! 843: */ ! 844: int ! 845: streamclose1(Stream *s) ! 846: { ! 847: Queue *q, *nq; ! 848: int rv; ! 849: ! 850: /* ! 851: * decrement the open count ! 852: */ ! 853: qlock(s); ! 854: if(s->opens-- == 1){ ! 855: /* ! 856: * descend the stream closing the queues ! 857: */ ! 858: for(q = s->procq; q; q = q->next){ ! 859: if(!waserror()){ ! 860: if(q->info->close) ! 861: (*q->info->close)(q->other); ! 862: poperror(); ! 863: } ! 864: WR(q)->put = nullput; ! 865: ! 866: /* ! 867: * this may be 2 streams joined device end to device end ! 868: */ ! 869: if(q == s->devq->other) ! 870: break; ! 871: } ! 872: ! 873: /* ! 874: * ascend the stream flushing the queues ! 875: */ ! 876: for(q = s->devq; q; q = nq){ ! 877: nq = q->next; ! 878: flushq(q); ! 879: } ! 880: } ! 881: rv = s->opens; ! 882: qunlock(s); ! 883: ! 884: /* ! 885: * leave it and free it ! 886: */ ! 887: streamexit(s); ! 888: return rv; ! 889: } ! 890: int ! 891: streamclose(Chan *c) ! 892: { ! 893: /* ! 894: * if no stream, ignore it ! 895: */ ! 896: if(!c->stream) ! 897: return 0; ! 898: return streamclose1(c->stream); ! 899: } ! 900: ! 901: /* ! 902: * put a block to be read into the queue. wakeup any waiting reader ! 903: */ ! 904: void ! 905: stputq(Queue *q, Block *bp) ! 906: { ! 907: int awaken; ! 908: Stream *s; ! 909: ! 910: if(bp->type == M_HANGUP){ ! 911: s = q->ptr; ! 912: if(bp->rptr<bp->wptr && s->err==0) ! 913: s->err = bp; ! 914: else ! 915: freeb(bp); ! 916: q->flag |= QHUNGUP; ! 917: q->other->flag |= QHUNGUP; ! 918: wakeup(q->other->rp); ! 919: awaken = 1; ! 920: } else { ! 921: lock(q); ! 922: if(q->first) ! 923: q->last->next = bp; ! 924: else ! 925: q->first = bp; ! 926: q->len += BLEN(bp); ! 927: q->nb++; ! 928: awaken = bp->flags & S_DELIM; ! 929: while(bp->next) { ! 930: bp = bp->next; ! 931: q->len += BLEN(bp); ! 932: q->nb++; ! 933: awaken |= bp->flags & S_DELIM; ! 934: } ! 935: q->last = bp; ! 936: if(q->len >= Streamhi || q->nb >= Streambhi){ ! 937: q->flag |= QHIWAT; ! 938: awaken = 1; ! 939: } ! 940: unlock(q); ! 941: } ! 942: if(awaken) ! 943: wakeup(q->rp); ! 944: } ! 945: ! 946: /* ! 947: * return the stream id ! 948: */ ! 949: long ! 950: streamctlread(Chan *c, void *vbuf, long n) ! 951: { ! 952: char *buf = vbuf; ! 953: char num[32]; ! 954: Stream *s; ! 955: ! 956: s = c->stream; ! 957: if(STREAMTYPE(c->qid.path) == Sctlqid){ ! 958: sprint(num, "%d", s->id); ! 959: return readstr(c->offset, buf, n, num); ! 960: } else { ! 961: if(CHDIR & c->qid.path) ! 962: return devdirread(c, vbuf, n, 0, 0, streamgen); ! 963: else ! 964: panic("streamctlread"); ! 965: } ! 966: return 0; /* not reached */ ! 967: } ! 968: ! 969: /* ! 970: * return true if there is an output buffer available ! 971: */ ! 972: static int ! 973: isinput(void *x) ! 974: { ! 975: Queue *q; ! 976: ! 977: q = (Queue *)x; ! 978: return (q->flag&QHUNGUP) || q->first!=0; ! 979: } ! 980: ! 981: /* ! 982: * read until we fill the buffer or until a DELIM is encountered ! 983: */ ! 984: long ! 985: streamread(Chan *c, void *vbuf, long n) ! 986: { ! 987: Block *bp; ! 988: Block *tofree; ! 989: Stream *s; ! 990: Queue *q; ! 991: int left, i; ! 992: uchar *buf = vbuf; ! 993: ! 994: if(STREAMTYPE(c->qid.path) != Sdataqid) ! 995: return streamctlread(c, vbuf, n); ! 996: ! 997: /* ! 998: * one reader at a time ! 999: */ ! 1000: s = c->stream; ! 1001: left = n; ! 1002: qlock(&s->rdlock); ! 1003: tofree = 0; ! 1004: q = 0; ! 1005: if(waserror()){ ! 1006: /* ! 1007: * put any partially read message back into the ! 1008: * queue ! 1009: */ ! 1010: while(tofree){ ! 1011: bp = tofree; ! 1012: tofree = bp->next; ! 1013: bp->next = 0; ! 1014: putbq(q, bp); ! 1015: } ! 1016: qunlock(&s->rdlock); ! 1017: nexterror(); ! 1018: } ! 1019: ! 1020: /* ! 1021: * sleep till data is available ! 1022: */ ! 1023: q = RD(s->procq); ! 1024: while(left){ ! 1025: bp = getq(q); ! 1026: if(bp == 0){ ! 1027: if(q->flag & QHUNGUP){ ! 1028: if(s->err) ! 1029: error((char*)s->err->rptr); ! 1030: else if(s->hread++<3) ! 1031: break; ! 1032: else ! 1033: error(Ehungup); ! 1034: } ! 1035: q->rp = &q->r; ! 1036: sleep(q->rp, isinput, (void *)q); ! 1037: continue; ! 1038: } ! 1039: ! 1040: i = BLEN(bp); ! 1041: if(i <= left){ ! 1042: memmove(buf, bp->rptr, i); ! 1043: left -= i; ! 1044: buf += i; ! 1045: bp->next = tofree; ! 1046: tofree = bp; ! 1047: if(bp->flags & S_DELIM) ! 1048: break; ! 1049: } else { ! 1050: memmove(buf, bp->rptr, left); ! 1051: bp->rptr += left; ! 1052: putbq(q, bp); ! 1053: left = 0; ! 1054: } ! 1055: } ! 1056: ! 1057: /* ! 1058: * free completely read blocks ! 1059: */ ! 1060: if(tofree) ! 1061: freeb(tofree); ! 1062: ! 1063: qunlock(&s->rdlock); ! 1064: poperror(); ! 1065: return n - left; ! 1066: } ! 1067: ! 1068: /* ! 1069: * look for an instance of the line discipline `name' on ! 1070: * the stream `s' ! 1071: */ ! 1072: void ! 1073: qlook(Stream *s, char *name) ! 1074: { ! 1075: Queue *q; ! 1076: ! 1077: for(q = s->procq; q; q = q->next){ ! 1078: if(strcmp(q->info->name, name) == 0) ! 1079: return; ! 1080: ! 1081: /* ! 1082: * this may be 2 streams joined device end to device end ! 1083: */ ! 1084: if(q == s->devq->other) ! 1085: break; ! 1086: } ! 1087: error(Ebadarg); ! 1088: } ! 1089: ! 1090: /* ! 1091: * Handle a ctl request. Streamwide requests are: ! 1092: * ! 1093: * hangup -- send an M_HANGUP up the stream ! 1094: * push ldname -- push the line discipline named ldname ! 1095: * pop -- pop a line discipline ! 1096: * look ldname -- look for a line discipline ! 1097: * ! 1098: * This routing is entered with s->wrlock'ed and must unlock. ! 1099: */ ! 1100: static long ! 1101: streamctlwrite(Chan *c, void *a, long n) ! 1102: { ! 1103: Qinfo *qi; ! 1104: Block *bp; ! 1105: Stream *s; ! 1106: ! 1107: if(STREAMTYPE(c->qid.path) != Sctlqid) ! 1108: panic("streamctlwrite %lux", c->qid); ! 1109: s = c->stream; ! 1110: ! 1111: /* ! 1112: * package ! 1113: */ ! 1114: bp = allocb(n+1); ! 1115: memmove(bp->wptr, a, n); ! 1116: bp->wptr[n] = 0; ! 1117: bp->wptr += n + 1; ! 1118: ! 1119: /* ! 1120: * check for standard requests ! 1121: */ ! 1122: if(streamparse("hangup", bp)){ ! 1123: hangup(s); ! 1124: freeb(bp); ! 1125: } else if(streamparse("push", bp)){ ! 1126: qi = qinfofind((char *)bp->rptr); ! 1127: if(qi == 0) ! 1128: error(Ebadld); ! 1129: pushq(s, qi); ! 1130: freeb(bp); ! 1131: } else if(streamparse("pop", bp)){ ! 1132: popq(s); ! 1133: freeb(bp); ! 1134: } else if(streamparse("look", bp)){ ! 1135: qlook(s, (char *)bp->rptr); ! 1136: freeb(bp); ! 1137: } else { ! 1138: bp->type = M_CTL; ! 1139: bp->flags |= S_DELIM; ! 1140: PUTNEXT(s->procq, bp); ! 1141: } ! 1142: ! 1143: return n; ! 1144: } ! 1145: ! 1146: /* ! 1147: * wait till there's room in the next stream ! 1148: */ ! 1149: static int ! 1150: notfull(void *arg) ! 1151: { ! 1152: return !QFULL((Queue *)arg); ! 1153: } ! 1154: void ! 1155: flowctl(Queue *q, Block *bp) ! 1156: { ! 1157: if(bp->type != M_HANGUP){ ! 1158: qlock(&q->rlock); ! 1159: if(waserror()){ ! 1160: qunlock(&q->rlock); ! 1161: freeb(bp); ! 1162: nexterror(); ! 1163: } ! 1164: q->rp = &q->r; ! 1165: sleep(q->rp, notfull, q->next); ! 1166: qunlock(&q->rlock); ! 1167: poperror(); ! 1168: } ! 1169: PUTNEXT(q, bp); ! 1170: } ! 1171: ! 1172: /* ! 1173: * send the request as a single delimited block ! 1174: */ ! 1175: long ! 1176: streamwrite(Chan *c, void *a, long n, int docopy) ! 1177: { ! 1178: Stream *s; ! 1179: Queue *q; ! 1180: long rem; ! 1181: int i; ! 1182: Block *bp; ! 1183: char *va; ! 1184: ! 1185: /* ! 1186: * docopy will get used if I ever figure out when to avoid copying ! 1187: * data. -- presotto ! 1188: */ ! 1189: USED(docopy); ! 1190: ! 1191: s = c->stream; ! 1192: ! 1193: /* ! 1194: * decode the qid ! 1195: */ ! 1196: if(STREAMTYPE(c->qid.path) != Sdataqid) ! 1197: return streamctlwrite(c, a, n); ! 1198: ! 1199: /* ! 1200: * No writes allowed on hungup channels ! 1201: */ ! 1202: q = s->procq; ! 1203: if(q->other->flag & QHUNGUP){ ! 1204: if(s->err) ! 1205: error((char*)(s->err->rptr)); ! 1206: else ! 1207: error(Ehungup); ! 1208: } ! 1209: ! 1210: /* ! 1211: * Write the message using blocks <= Streamhi bytes longs ! 1212: */ ! 1213: va = a; ! 1214: rem = n; ! 1215: for(;;){ ! 1216: if(rem > Streamhi) ! 1217: i = Streamhi; ! 1218: else ! 1219: i = rem; ! 1220: bp = allocb(i); ! 1221: memmove(bp->wptr, va, i); ! 1222: bp->wptr += i; ! 1223: va += i; ! 1224: rem -= i; ! 1225: if(rem > 0){ ! 1226: FLOWCTL(q, bp); ! 1227: } else { ! 1228: bp->flags |= S_DELIM; ! 1229: FLOWCTL(q, bp); ! 1230: break; ! 1231: } ! 1232: } ! 1233: return n; ! 1234: } ! 1235: ! 1236: /* ! 1237: * stat a stream. the length is the number of bytes up to the ! 1238: * first delimiter. ! 1239: */ ! 1240: void ! 1241: streamstat(Chan *c, char *db, char *name, long perm) ! 1242: { ! 1243: Dir dir; ! 1244: Stream *s; ! 1245: Queue *q; ! 1246: Block *bp; ! 1247: long n; ! 1248: ! 1249: s = c->stream; ! 1250: n = 0; ! 1251: if(s) { ! 1252: q = RD(s->procq); ! 1253: if(q->flag & QHUNGUP) ! 1254: error(Ehungup); ! 1255: lock(q); ! 1256: for(bp=q->first; bp; bp = bp->next){ ! 1257: n += BLEN(bp); ! 1258: if(bp->flags&S_DELIM) ! 1259: break; ! 1260: } ! 1261: unlock(q); ! 1262: } ! 1263: ! 1264: devdir(c, c->qid, name, n, eve, perm, &dir); ! 1265: convD2M(&dir, db); ! 1266: } ! 1267: ! 1268: /* ! 1269: * send a hangup up a stream ! 1270: */ ! 1271: static void ! 1272: hangup(Stream *s) ! 1273: { ! 1274: Block *bp; ! 1275: ! 1276: bp = allocb(0); ! 1277: bp->type = M_HANGUP; ! 1278: if(s->devq && s->devq->put) ! 1279: (*s->devq->put)(s->devq, bp); ! 1280: else ! 1281: freeb(bp); ! 1282: } ! 1283: ! 1284: /* ! 1285: * parse a string and return a pointer to the second element if the ! 1286: * first matches name. bp->rptr will be updated to point to the ! 1287: * second element. ! 1288: * ! 1289: * return 0 if no match. ! 1290: * ! 1291: * it is assumed that the block data is null terminated. streamwrite ! 1292: * guarantees this. ! 1293: */ ! 1294: int ! 1295: streamparse(char *name, Block *bp) ! 1296: { ! 1297: int len; ! 1298: ! 1299: len = strlen(name); ! 1300: if(BLEN(bp) < len) ! 1301: return 0; ! 1302: if(strncmp(name, (char *)bp->rptr, len)==0){ ! 1303: if(bp->rptr[len] == ' ') ! 1304: bp->rptr += len+1; ! 1305: else if(bp->rptr[len]) ! 1306: return 0; ! 1307: else ! 1308: bp->rptr += len; ! 1309: while(*bp->rptr==' ' && bp->wptr>bp->rptr) ! 1310: bp->rptr++; ! 1311: return 1; ! 1312: } ! 1313: return 0; ! 1314: } ! 1315: ! 1316: /* ! 1317: * like andrew's getmfields but no hidden state ! 1318: */ ! 1319: int ! 1320: getfields(char *lp, char **fields, int n, char *sep) ! 1321: { ! 1322: int i; ! 1323: ! 1324: for(i=0; lp && *lp && i<n; i++){ ! 1325: while(*lp && strchr(sep, *lp) != 0) ! 1326: *lp++=0; ! 1327: if(*lp == 0) ! 1328: break; ! 1329: fields[i]=lp; ! 1330: while(*lp && strchr(sep, *lp) == 0) ! 1331: lp++; ! 1332: } ! 1333: return i; ! 1334: } ! 1335: ! 1336: static Streamopen permstopen; ! 1337: static Streamput permstput; ! 1338: ! 1339: /* ! 1340: * pushing this line discipline makes the stream unclosable, i.e., always there ! 1341: */ ! 1342: Qinfo perminfo = ! 1343: { ! 1344: permstput, ! 1345: permstput, ! 1346: permstopen, ! 1347: 0, ! 1348: "permanent" ! 1349: }; ! 1350: ! 1351: static void ! 1352: permstopen(Queue *q, Stream *s) ! 1353: { ! 1354: USED(q); ! 1355: s->opens++; ! 1356: s->inuse++; ! 1357: } ! 1358: ! 1359: static void ! 1360: permstput(Queue *q, Block *bp) ! 1361: { ! 1362: PUTNEXT(q, bp); ! 1363: }
This archive runs on limited infrastructure. Preserving old code on modern bandwidth. Automated agents are requested to crawl responsibly.