1 #include "dat.h"
2 #include "fns.h"
3 #include "error.h"
4
5 #define QDEBUG if(0)
6
7 /*
8 * IO queues
9 */
10
11 struct Queue
12 {
13 Lock l;
14
15 Block* bfirst; /* buffer */
16 Block* blast;
17
18 int len; /* bytes allocated to queue */
19 int dlen; /* data bytes in queue */
20 int limit; /* max bytes in queue */
21 int inilim; /* initial limit */
22 int state;
23 int noblock; /* true if writes return immediately when q full */
24 int eof; /* number of eofs read by user */
25
26 void (*kick)(void*); /* restart output */
27 void (*bypass)(void*, Block*); /* bypass queue altogether */
28 void* arg; /* argument to kick */
29
30 QLock rlock; /* mutex for reading processes */
31 Rendez rr; /* process waiting to read */
32 QLock wlock; /* mutex for writing processes */
33 Rendez wr; /* process waiting to write */
34
35 char err[ERRMAX];
36 int want;
37 };
38
39 enum
40 {
41 Maxatomic = 32*1024
42 };
43
44 uint qiomaxatomic = Maxatomic;
45
46 void
checkb(Block * b,char * msg)47 checkb(Block *b, char *msg)
48 {
49 if(b->base > b->lim)
50 panic("checkb 0 %s %lux %lux", msg, b->base, b->lim);
51 if(b->rp < b->base)
52 panic("checkb 1 %s %lux %lux", msg, b->base, b->rp);
53 if(b->wp < b->base)
54 panic("checkb 2 %s %lux %lux", msg, b->base, b->wp);
55 if(b->rp > b->lim)
56 panic("checkb 3 %s %lux %lux", msg, b->rp, b->lim);
57 if(b->wp > b->lim)
58 panic("checkb 4 %s %lux %lux", msg, b->wp, b->lim);
59 }
60
61 void
freeb(Block * b)62 freeb(Block *b)
63 {
64 if(b == nil)
65 return;
66
67 /*
68 * drivers which perform non cache coherent DMA manage their own buffer
69 * pool of uncached buffers and provide their own free routine.
70 */
71 if(b->free) {
72 b->free(b);
73 return;
74 }
75
76 /* poison the block in case someone is still holding onto it */
77 b->next = (void*)0xdeadbabe;
78 b->rp = (void*)0xdeadbabe;
79 b->wp = (void*)0xdeadbabe;
80 b->lim = (void*)0xdeadbabe;
81 b->base = (void*)0xdeadbabe;
82
83 free(b);
84 }
85
86 /*
87 * free a list of blocks
88 */
89 void
freeblist(Block * b)90 freeblist(Block *b)
91 {
92 Block *next;
93
94 for(; b != 0; b = next){
95 next = b->next;
96 b->next = 0;
97 freeb(b);
98 }
99 }
100
101 ulong padblockoverhead;
102
103 /*
104 * pad a block to the front (or the back if size is negative)
105 */
106 Block*
padblock(Block * bp,int size)107 padblock(Block *bp, int size)
108 {
109 int n;
110 Block *nbp;
111
112 if(size >= 0){
113 if(bp->rp - bp->base >= size){
114 bp->rp -= size;
115 return bp;
116 }
117
118 if(bp->next)
119 panic("padblock 0x%luX", getcallerpc(&bp));
120 n = BLEN(bp);
121 padblockoverhead += n;
122 nbp = allocb(size+n);
123 nbp->rp += size;
124 nbp->wp = nbp->rp;
125 memmove(nbp->wp, bp->rp, n);
126 nbp->wp += n;
127 freeb(bp);
128 nbp->rp -= size;
129 } else {
130 size = -size;
131
132 if(bp->next)
133 panic("padblock 0x%luX", getcallerpc(&bp));
134
135 if(bp->lim - bp->wp >= size)
136 return bp;
137
138 n = BLEN(bp);
139 padblockoverhead += n;
140 nbp = allocb(size+n);
141 memmove(nbp->wp, bp->rp, n);
142 nbp->wp += n;
143 freeb(bp);
144 }
145 return nbp;
146 }
147
148 /*
149 * return count of bytes in a string of blocks
150 */
151 int
blocklen(Block * bp)152 blocklen(Block *bp)
153 {
154 int len;
155
156 len = 0;
157 while(bp) {
158 len += BLEN(bp);
159 bp = bp->next;
160 }
161 return len;
162 }
163
164 /*
165 * return count of space in blocks
166 */
167 int
blockalloclen(Block * bp)168 blockalloclen(Block *bp)
169 {
170 int len;
171
172 len = 0;
173 while(bp) {
174 len += BALLOC(bp);
175 bp = bp->next;
176 }
177 return len;
178 }
179
180 /*
181 * copy the string of blocks into
182 * a single block and free the string
183 */
184 Block*
concatblock(Block * bp)185 concatblock(Block *bp)
186 {
187 int len;
188 Block *nb, *f;
189
190 if(bp->next == 0)
191 return bp;
192
193 nb = allocb(blocklen(bp));
194 for(f = bp; f; f = f->next) {
195 len = BLEN(f);
196 memmove(nb->wp, f->rp, len);
197 nb->wp += len;
198 }
199 freeblist(bp);
200 return nb;
201 }
202
203 /*
204 * make sure the first block has at least n bytes. If we started with
205 * less than n bytes, make sure we have exactly n bytes. devssl.c depends
206 * on this.
207 */
208 Block*
pullupblock(Block * bp,int n)209 pullupblock(Block *bp, int n)
210 {
211 int i;
212 Block *nbp;
213
214 /*
215 * this should almost always be true, the rest it
216 * just to avoid every caller checking.
217 */
218 if(BLEN(bp) >= n)
219 return bp;
220
221 /*
222 * if not enough room in the first block,
223 * add another to the front of the list.
224 */
225 if(bp->lim - bp->rp < n){
226 nbp = allocb(n);
227 nbp->next = bp;
228 bp = nbp;
229 }
230
231 /*
232 * copy bytes from the trailing blocks into the first
233 */
234 n -= BLEN(bp);
235 while(nbp = bp->next){
236 i = BLEN(nbp);
237 if(i > n) {
238 memmove(bp->wp, nbp->rp, n);
239 bp->wp += n;
240 nbp->rp += n;
241 return bp;
242 }
243 else {
244 memmove(bp->wp, nbp->rp, i);
245 bp->wp += i;
246 bp->next = nbp->next;
247 nbp->next = 0;
248 freeb(nbp);
249 n -= i;
250 if(n == 0)
251 return bp;
252 }
253 }
254 freeblist(bp);
255 return 0;
256 }
257
258 /*
259 * make sure the first block has at least n bytes
260 */
261 Block*
pullupqueue(Queue * q,int n)262 pullupqueue(Queue *q, int n)
263 {
264 Block *b;
265
266 if(BLEN(q->bfirst) >= n)
267 return q->bfirst;
268 q->bfirst = pullupblock(q->bfirst, n);
269 for(b = q->bfirst; b != nil && b->next != nil; b = b->next)
270 ;
271 q->blast = b;
272 return q->bfirst;
273 }
274
275 /*
276 * trim to len bytes starting at offset
277 */
278 Block *
trimblock(Block * bp,int offset,int len)279 trimblock(Block *bp, int offset, int len)
280 {
281 ulong l;
282 Block *nb, *startb;
283
284 if(blocklen(bp) < offset+len) {
285 freeblist(bp);
286 return nil;
287 }
288
289 while((l = BLEN(bp)) < offset) {
290 offset -= l;
291 nb = bp->next;
292 bp->next = nil;
293 freeb(bp);
294 bp = nb;
295 }
296
297 startb = bp;
298 bp->rp += offset;
299
300 while((l = BLEN(bp)) < len) {
301 len -= l;
302 bp = bp->next;
303 }
304
305 bp->wp -= (BLEN(bp) - len);
306
307 if(bp->next) {
308 freeblist(bp->next);
309 bp->next = nil;
310 }
311
312 return startb;
313 }
314
315 /*
316 * copy 'count' bytes into a new block
317 */
318 Block*
copyblock(Block * bp,int count)319 copyblock(Block *bp, int count)
320 {
321 int l;
322 Block *nbp;
323
324 nbp = allocb(count);
325 for(; count > 0 && bp != 0; bp = bp->next){
326 l = BLEN(bp);
327 if(l > count)
328 l = count;
329 memmove(nbp->wp, bp->rp, l);
330 nbp->wp += l;
331 count -= l;
332 }
333 if(count > 0){
334 memset(nbp->wp, 0, count);
335 nbp->wp += count;
336 }
337
338 return nbp;
339 }
340
341 Block*
adjustblock(Block * bp,int len)342 adjustblock(Block* bp, int len)
343 {
344 int n;
345 Block *nbp;
346
347 if(len < 0){
348 freeb(bp);
349 return nil;
350 }
351
352 if(bp->rp+len > bp->lim){
353 nbp = copyblock(bp, len);
354 freeblist(bp);
355 QDEBUG checkb(nbp, "adjustblock 1");
356
357 return nbp;
358 }
359
360 n = BLEN(bp);
361 if(len > n)
362 memset(bp->wp, 0, len-n);
363 bp->wp = bp->rp+len;
364 QDEBUG checkb(bp, "adjustblock 2");
365
366 return bp;
367 }
368
369 /*
370 * throw away up to count bytes from a
371 * list of blocks. Return count of bytes
372 * thrown away.
373 */
374 int
pullblock(Block ** bph,int count)375 pullblock(Block **bph, int count)
376 {
377 Block *bp;
378 int n, bytes;
379
380 bytes = 0;
381 if(bph == nil)
382 return 0;
383
384 while(*bph != nil && count != 0) {
385 bp = *bph;
386 n = BLEN(bp);
387 if(count < n)
388 n = count;
389 bytes += n;
390 count -= n;
391 bp->rp += n;
392 if(BLEN(bp) == 0) {
393 *bph = bp->next;
394 bp->next = nil;
395 freeb(bp);
396 }
397 }
398 return bytes;
399 }
400
401 /*
402 * allocate queues and blocks (round data base address to 64 bit boundary)
403 */
404 Block*
iallocb(int size)405 iallocb(int size)
406 {
407 Block *b;
408 ulong addr;
409
410 b = kmalloc(sizeof(Block)+size);
411 if(b == 0)
412 return 0;
413 memset(b, 0, sizeof(Block));
414
415 addr = (ulong)b + sizeof(Block);
416 b->base = (uchar*)addr;
417 b->lim = b->base + size;
418 b->rp = b->base;
419 b->wp = b->rp;
420
421 return b;
422 }
423
424
425 /*
426 * call error if iallocb fails
427 */
428 Block*
allocb(int size)429 allocb(int size)
430 {
431 Block *b;
432
433 b = iallocb(size);
434 if(b == 0)
435 exhausted("allocb");
436 return b;
437 }
438
439
440 /*
441 * get next block from a queue, return null if nothing there
442 */
443 Block*
qget(Queue * q)444 qget(Queue *q)
445 {
446 int dowakeup;
447 Block *b;
448
449 /* sync with qwrite */
450 lock(&q->l);
451
452 b = q->bfirst;
453 if(b == 0){
454 q->state |= Qstarve;
455 unlock(&q->l);
456 return 0;
457 }
458 q->bfirst = b->next;
459 b->next = 0;
460 q->len -= BALLOC(b);
461 q->dlen -= BLEN(b);
462 QDEBUG checkb(b, "qget");
463
464 /* if writer flow controlled, restart */
465 if((q->state & Qflow) && q->len < q->limit/2){
466 q->state &= ~Qflow;
467 dowakeup = 1;
468 } else
469 dowakeup = 0;
470
471 unlock(&q->l);
472
473 if(dowakeup)
474 Wakeup(&q->wr);
475
476 return b;
477 }
478
479 /*
480 * throw away the next 'len' bytes in the queue
481 */
482 int
qdiscard(Queue * q,int len)483 qdiscard(Queue *q, int len)
484 {
485 Block *b;
486 int dowakeup, n, sofar;
487
488 lock(&q->l);
489 for(sofar = 0; sofar < len; sofar += n){
490 b = q->bfirst;
491 if(b == nil)
492 break;
493 QDEBUG checkb(b, "qdiscard");
494 n = BLEN(b);
495 if(n <= len - sofar){
496 q->bfirst = b->next;
497 b->next = 0;
498 q->len -= BALLOC(b);
499 q->dlen -= BLEN(b);
500 freeb(b);
501 } else {
502 n = len - sofar;
503 b->rp += n;
504 q->dlen -= n;
505 }
506 }
507
508 /* if writer flow controlled, restart */
509 if((q->state & Qflow) && q->len < q->limit){
510 q->state &= ~Qflow;
511 dowakeup = 1;
512 } else
513 dowakeup = 0;
514
515 unlock(&q->l);
516
517 if(dowakeup)
518 Wakeup(&q->wr);
519
520 return sofar;
521 }
522
523 /*
524 * Interrupt level copy out of a queue, return # bytes copied.
525 */
526 int
qconsume(Queue * q,void * vp,int len)527 qconsume(Queue *q, void *vp, int len)
528 {
529 Block *b;
530 int n, dowakeup;
531 uchar *p = vp;
532 Block *tofree = nil;
533
534 /* sync with qwrite */
535 lock(&q->l);
536
537 for(;;) {
538 b = q->bfirst;
539 if(b == 0){
540 q->state |= Qstarve;
541 unlock(&q->l);
542 return -1;
543 }
544 QDEBUG checkb(b, "qconsume 1");
545
546 n = BLEN(b);
547 if(n > 0)
548 break;
549 q->bfirst = b->next;
550 q->len -= BALLOC(b);
551
552 /* remember to free this */
553 b->next = tofree;
554 tofree = b;
555 }
556
557 if(n < len)
558 len = n;
559 memmove(p, b->rp, len);
560 if((q->state & Qmsg) || len == n)
561 q->bfirst = b->next;
562 b->rp += len;
563 q->dlen -= len;
564
565 /* discard the block if we're done with it */
566 if((q->state & Qmsg) || len == n){
567 b->next = 0;
568 q->len -= BALLOC(b);
569 q->dlen -= BLEN(b);
570
571 /* remember to free this */
572 b->next = tofree;
573 tofree = b;
574 }
575
576 /* if writer flow controlled, restart */
577 if((q->state & Qflow) && q->len < q->limit/2){
578 q->state &= ~Qflow;
579 dowakeup = 1;
580 } else
581 dowakeup = 0;
582
583 unlock(&q->l);
584
585 if(dowakeup)
586 Wakeup(&q->wr);
587
588 if(tofree != nil)
589 freeblist(tofree);
590
591 return len;
592 }
593
594 int
qpass(Queue * q,Block * b)595 qpass(Queue *q, Block *b)
596 {
597 int dlen, len, dowakeup;
598
599 /* sync with qread */
600 dowakeup = 0;
601 lock(&q->l);
602 if(q->len >= q->limit){
603 unlock(&q->l);
604 freeblist(b);
605 return -1;
606 }
607 if(q->state & Qclosed){
608 unlock(&q->l);
609 len = blocklen(b);
610 freeblist(b);
611 return len;
612 }
613
614 /* add buffer to queue */
615 if(q->bfirst)
616 q->blast->next = b;
617 else
618 q->bfirst = b;
619 len = BALLOC(b);
620 dlen = BLEN(b);
621 QDEBUG checkb(b, "qpass");
622 while(b->next){
623 b = b->next;
624 QDEBUG checkb(b, "qpass");
625 len += BALLOC(b);
626 dlen += BLEN(b);
627 }
628 q->blast = b;
629 q->len += len;
630 q->dlen += dlen;
631
632 if(q->len >= q->limit/2)
633 q->state |= Qflow;
634
635 if(q->state & Qstarve){
636 q->state &= ~Qstarve;
637 dowakeup = 1;
638 }
639 unlock(&q->l);
640
641 if(dowakeup)
642 Wakeup(&q->rr);
643
644 return len;
645 }
646
647 int
qpassnolim(Queue * q,Block * b)648 qpassnolim(Queue *q, Block *b)
649 {
650 int dlen, len, dowakeup;
651
652 /* sync with qread */
653 dowakeup = 0;
654 lock(&q->l);
655
656 len = BALLOC(b);
657 if(q->state & Qclosed){
658 unlock(&q->l);
659 freeblist(b);
660 return len;
661 }
662
663 /* add buffer to queue */
664 if(q->bfirst)
665 q->blast->next = b;
666 else
667 q->bfirst = b;
668 dlen = BLEN(b);
669 QDEBUG checkb(b, "qpass");
670 while(b->next){
671 b = b->next;
672 QDEBUG checkb(b, "qpass");
673 len += BALLOC(b);
674 dlen += BLEN(b);
675 }
676 q->blast = b;
677 q->len += len;
678 q->dlen += dlen;
679
680 if(q->len >= q->limit/2)
681 q->state |= Qflow;
682
683 if(q->state & Qstarve){
684 q->state &= ~Qstarve;
685 dowakeup = 1;
686 }
687 unlock(&q->l);
688
689 if(dowakeup)
690 Wakeup(&q->rr);
691
692 return len;
693 }
694
695 /*
696 * if the allocated space is way out of line with the used
697 * space, reallocate to a smaller block
698 */
699 Block*
packblock(Block * bp)700 packblock(Block *bp)
701 {
702 Block **l, *nbp;
703 int n;
704
705 for(l = &bp; *l; l = &(*l)->next){
706 nbp = *l;
707 n = BLEN(nbp);
708 if((n<<2) < BALLOC(nbp)){
709 *l = allocb(n);
710 memmove((*l)->wp, nbp->rp, n);
711 (*l)->wp += n;
712 (*l)->next = nbp->next;
713 freeb(nbp);
714 }
715 }
716
717 return bp;
718 }
719
720 int
qproduce(Queue * q,void * vp,int len)721 qproduce(Queue *q, void *vp, int len)
722 {
723 Block *b;
724 int dowakeup;
725 uchar *p = vp;
726
727 /* sync with qread */
728 dowakeup = 0;
729 lock(&q->l);
730
731 if(q->state & Qclosed){
732 unlock(&q->l);
733 return -1;
734 }
735
736 /* no waiting receivers, room in buffer? */
737 if(q->len >= q->limit){
738 q->state |= Qflow;
739 unlock(&q->l);
740 return -1;
741 }
742
743 /* save in buffer */
744 b = iallocb(len);
745 if(b == 0){
746 unlock(&q->l);
747 print("qproduce: iallocb failed\n");
748 return -1;
749 }
750 memmove(b->wp, p, len);
751 b->wp += len;
752 if(q->bfirst)
753 q->blast->next = b;
754 else
755 q->bfirst = b;
756 q->blast = b;
757 /* b->next = 0; done by allocb() */
758 q->len += BALLOC(b);
759 q->dlen += BLEN(b);
760 QDEBUG checkb(b, "qproduce");
761
762 if(q->state & Qstarve){
763 q->state &= ~Qstarve;
764 dowakeup = 1;
765 }
766
767 if(q->len >= q->limit)
768 q->state |= Qflow;
769 unlock(&q->l);
770
771
772 if(dowakeup)
773 Wakeup(&q->rr);
774
775 return len;
776 }
777
778 /*
779 * copy from offset in the queue
780 */
781 Block*
qcopy(Queue * q,int len,ulong offset)782 qcopy(Queue *q, int len, ulong offset)
783 {
784 int sofar;
785 int n;
786 Block *b, *nb;
787 uchar *p;
788
789 nb = allocb(len);
790
791 lock(&q->l);
792
793 /* go to offset */
794 b = q->bfirst;
795 for(sofar = 0; ; sofar += n){
796 if(b == nil){
797 unlock(&q->l);
798 return nb;
799 }
800 n = BLEN(b);
801 if(sofar + n > offset){
802 p = b->rp + offset - sofar;
803 n -= offset - sofar;
804 break;
805 }
806 b = b->next;
807 }
808
809 /* copy bytes from there */
810 for(sofar = 0; sofar < len;){
811 if(n > len - sofar)
812 n = len - sofar;
813 memmove(nb->wp, p, n);
814 sofar += n;
815 nb->wp += n;
816 b = b->next;
817 if(b == nil)
818 break;
819 n = BLEN(b);
820 p = b->rp;
821 }
822 unlock(&q->l);
823
824 return nb;
825 }
826
827 /*
828 * called by non-interrupt code
829 */
830 Queue*
qopen(int limit,int msg,void (* kick)(void *),void * arg)831 qopen(int limit, int msg, void (*kick)(void*), void *arg)
832 {
833 Queue *q;
834
835 q = kmalloc(sizeof(Queue));
836 if(q == 0)
837 return 0;
838
839 q->limit = q->inilim = limit;
840 q->kick = kick;
841 q->arg = arg;
842 q->state = msg;
843 q->state |= Qstarve;
844 q->eof = 0;
845 q->noblock = 0;
846
847 return q;
848 }
849
850 /* open a queue to be bypassed */
851 Queue*
qbypass(void (* bypass)(void *,Block *),void * arg)852 qbypass(void (*bypass)(void*, Block*), void *arg)
853 {
854 Queue *q;
855
856 q = malloc(sizeof(Queue));
857 if(q == 0)
858 return 0;
859
860 q->limit = 0;
861 q->arg = arg;
862 q->bypass = bypass;
863 q->state = 0;
864
865 return q;
866 }
867
868 static int
notempty(void * a)869 notempty(void *a)
870 {
871 Queue *q = a;
872
873 return (q->state & Qclosed) || q->bfirst != 0;
874 }
875
876 /*
877 * wait for the queue to be non-empty or closed.
878 * called with q ilocked.
879 */
880 static int
qwait(Queue * q)881 qwait(Queue *q)
882 {
883 /* wait for data */
884 for(;;){
885 if(q->bfirst != nil)
886 break;
887
888 if(q->state & Qclosed){
889 if(++q->eof > 3)
890 return -1;
891 if(*q->err && strcmp(q->err, Ehungup) != 0)
892 return -1;
893 return 0;
894 }
895
896 q->state |= Qstarve; /* flag requesting producer to wake me */
897 unlock(&q->l);
898 Sleep(&q->rr, notempty, q);
899 lock(&q->l);
900 }
901 return 1;
902 }
903
904 /*
905 * add a block list to a queue
906 */
907 void
qaddlist(Queue * q,Block * b)908 qaddlist(Queue *q, Block *b)
909 {
910 /* queue the block */
911 if(q->bfirst)
912 q->blast->next = b;
913 else
914 q->bfirst = b;
915 q->len += blockalloclen(b);
916 q->dlen += blocklen(b);
917 while(b->next)
918 b = b->next;
919 q->blast = b;
920 }
921
922 /*
923 * called with q locked
924 */
925 Block*
qremove(Queue * q)926 qremove(Queue *q)
927 {
928 Block *b;
929
930 b = q->bfirst;
931 if(b == nil)
932 return nil;
933 q->bfirst = b->next;
934 b->next = nil;
935 q->dlen -= BLEN(b);
936 q->len -= BALLOC(b);
937 QDEBUG checkb(b, "qremove");
938 return b;
939 }
940
941 /*
942 * copy the contents of a string of blocks into
943 * memory. emptied blocks are freed. return
944 * pointer to first unconsumed block.
945 */
946 Block*
bl2mem(uchar * p,Block * b,int n)947 bl2mem(uchar *p, Block *b, int n)
948 {
949 int i;
950 Block *next;
951
952 for(; b != nil; b = next){
953 i = BLEN(b);
954 if(i > n){
955 memmove(p, b->rp, n);
956 b->rp += n;
957 return b;
958 }
959 memmove(p, b->rp, i);
960 n -= i;
961 p += i;
962 b->rp += i;
963 next = b->next;
964 freeb(b);
965 }
966 return nil;
967 }
968
969 /*
970 * copy the contents of memory into a string of blocks.
971 * return nil on error.
972 */
973 Block*
mem2bl(uchar * p,int len)974 mem2bl(uchar *p, int len)
975 {
976 int n;
977 Block *b, *first, **l;
978
979 first = nil;
980 l = &first;
981 if(waserror()){
982 freeblist(first);
983 nexterror();
984 }
985 do {
986 n = len;
987 if(n > Maxatomic)
988 n = Maxatomic;
989
990 *l = b = allocb(n);
991 setmalloctag(b, (up->text[0]<<24)|(up->text[1]<<16)|(up->text[2]<<8)|up->text[3]);
992 memmove(b->wp, p, n);
993 b->wp += n;
994 p += n;
995 len -= n;
996 l = &b->next;
997 } while(len > 0);
998 poperror();
999
1000 return first;
1001 }
1002
1003 /*
1004 * put a block back to the front of the queue
1005 * called with q ilocked
1006 */
1007 void
qputback(Queue * q,Block * b)1008 qputback(Queue *q, Block *b)
1009 {
1010 b->next = q->bfirst;
1011 if(q->bfirst == nil)
1012 q->blast = b;
1013 q->bfirst = b;
1014 q->len += BALLOC(b);
1015 q->dlen += BLEN(b);
1016 }
1017
1018 /*
1019 * flow control, get producer going again
1020 * called with q locked
1021 */
1022 static void
qwakeup_unlock(Queue * q)1023 qwakeup_unlock(Queue *q)
1024 {
1025 int dowakeup = 0;
1026
1027 /* if writer flow controlled, restart */
1028 if((q->state & Qflow) && q->len < q->limit/2){
1029 q->state &= ~Qflow;
1030 dowakeup = 1;
1031 }
1032
1033 unlock(&q->l);
1034
1035 /* wakeup flow controlled writers */
1036 if(dowakeup){
1037 if(q->kick)
1038 q->kick(q->arg);
1039 Wakeup(&q->wr);
1040 }
1041 }
1042
1043 /*
1044 * get next block from a queue (up to a limit)
1045 */
1046 Block*
qbread(Queue * q,int len)1047 qbread(Queue *q, int len)
1048 {
1049 Block *b, *nb;
1050 int n;
1051
1052 qlock(&q->rlock);
1053 if(waserror()){
1054 qunlock(&q->rlock);
1055 nexterror();
1056 }
1057
1058 lock(&q->l);
1059 switch(qwait(q)){
1060 case 0:
1061 /* queue closed */
1062 unlock(&q->l);
1063 poperror();
1064 qunlock(&q->rlock);
1065 return nil;
1066 case -1:
1067 /* multiple reads on a closed queue */
1068 unlock(&q->l);
1069 error(q->err);
1070 }
1071
1072 /* if we get here, there's at least one block in the queue */
1073 b = qremove(q);
1074 n = BLEN(b);
1075
1076 /* split block if it's too big and this is not a message oriented queue */
1077 nb = b;
1078 if(n > len){
1079 if((q->state&Qmsg) == 0){
1080 n -= len;
1081 b = allocb(n);
1082 memmove(b->wp, nb->rp+len, n);
1083 b->wp += n;
1084 qputback(q, b);
1085 }
1086 nb->wp = nb->rp + len;
1087 }
1088
1089 /* restart producer */
1090 qwakeup_unlock(q);
1091
1092 poperror();
1093 qunlock(&q->rlock);
1094 return nb;
1095 }
1096
1097 /*
1098 * read a queue. if no data is queued, wait on its Rendez
1099 */
1100 long
qread(Queue * q,void * vp,int len)1101 qread(Queue *q, void *vp, int len)
1102 {
1103 Block *b, *first, **l;
1104 int m, n;
1105
1106 qlock(&q->rlock);
1107 if(waserror()){
1108 qunlock(&q->rlock);
1109 nexterror();
1110 }
1111
1112 lock(&q->l);
1113 again:
1114 switch(qwait(q)){
1115 case 0:
1116 /* queue closed */
1117 unlock(&q->l);
1118 poperror();
1119 qunlock(&q->rlock);
1120 return 0;
1121 case -1:
1122 /* multiple reads on a closed queue */
1123 unlock(&q->l);
1124 error(q->err);
1125 }
1126
1127 /* if we get here, there's at least one block in the queue */
1128 if(q->state & Qcoalesce){
1129 /* when coalescing, 0 length blocks just go away */
1130 b = q->bfirst;
1131 if(BLEN(b) <= 0){
1132 freeb(qremove(q));
1133 goto again;
1134 }
1135
1136 /* grab the first block plus as many
1137 * following blocks as will completely
1138 * fit in the read.
1139 */
1140 n = 0;
1141 l = &first;
1142 m = BLEN(b);
1143 for(;;) {
1144 *l = qremove(q);
1145 l = &b->next;
1146 n += m;
1147
1148 b = q->bfirst;
1149 if(b == nil)
1150 break;
1151 m = BLEN(b);
1152 if(n+m > len)
1153 break;
1154 }
1155 } else {
1156 first = qremove(q);
1157 n = BLEN(first);
1158 }
1159
1160 /* copy to user space outside of the ilock */
1161 unlock(&q->l);
1162 b = bl2mem(vp, first, len);
1163 lock(&q->l);
1164
1165 /* take care of any left over partial block */
1166 if(b != nil){
1167 n -= BLEN(b);
1168 if(q->state & Qmsg)
1169 freeb(b);
1170 else
1171 qputback(q, b);
1172 }
1173
1174 /* restart producer */
1175 qwakeup_unlock(q);
1176
1177 poperror();
1178 qunlock(&q->rlock);
1179 return n;
1180 }
1181
1182 static int
qnotfull(void * a)1183 qnotfull(void *a)
1184 {
1185 Queue *q = a;
1186
1187 return q->len < q->limit || (q->state & Qclosed);
1188 }
1189
1190 /*
1191 * add a block to a queue obeying flow control
1192 */
1193 long
qbwrite(Queue * q,Block * b)1194 qbwrite(Queue *q, Block *b)
1195 {
1196 int n, dowakeup;
1197 volatile struct {Block *b;} cb;
1198
1199 dowakeup = 0;
1200 n = BLEN(b);
1201 if(q->bypass){
1202 (*q->bypass)(q->arg, b);
1203 return n;
1204 }
1205 cb.b = b;
1206
1207 qlock(&q->wlock);
1208 if(waserror()){
1209 if(cb.b != nil)
1210 freeb(cb.b);
1211 qunlock(&q->wlock);
1212 nexterror();
1213 }
1214
1215 lock(&q->l);
1216
1217 /* give up if the queue is closed */
1218 if(q->state & Qclosed){
1219 unlock(&q->l);
1220 error(q->err);
1221 }
1222
1223 /* if nonblocking, don't queue over the limit */
1224 if(q->len >= q->limit){
1225 if(q->noblock){
1226 unlock(&q->l);
1227 freeb(b);
1228 poperror();
1229 qunlock(&q->wlock);
1230 return n;
1231 }
1232 }
1233
1234 /* queue the block */
1235 if(q->bfirst)
1236 q->blast->next = b;
1237 else
1238 q->bfirst = b;
1239 q->blast = b;
1240 b->next = 0;
1241 q->len += BALLOC(b);
1242 q->dlen += n;
1243 QDEBUG checkb(b, "qbwrite");
1244 cb.b = nil;
1245
1246 if(q->state & Qstarve){
1247 q->state &= ~Qstarve;
1248 dowakeup = 1;
1249 }
1250
1251 unlock(&q->l);
1252
1253 /* get output going again */
1254 if(q->kick && (dowakeup || (q->state&Qkick)))
1255 q->kick(q->arg);
1256
1257 if(dowakeup)
1258 Wakeup(&q->rr);
1259
1260 /*
1261 * flow control, wait for queue to get below the limit
1262 * before allowing the process to continue and queue
1263 * more. We do this here so that postnote can only
1264 * interrupt us after the data has been queued. This
1265 * means that things like 9p flushes and ssl messages
1266 * will not be disrupted by software interrupts.
1267 *
1268 * Note - this is moderately dangerous since a process
1269 * that keeps getting interrupted and rewriting will
1270 * queue infinite crud.
1271 */
1272 for(;;){
1273 if(q->noblock || qnotfull(q))
1274 break;
1275
1276 lock(&q->l);
1277 q->state |= Qflow;
1278 unlock(&q->l);
1279 Sleep(&q->wr, qnotfull, q);
1280 }
1281
1282 qunlock(&q->wlock);
1283 poperror();
1284 return n;
1285 }
1286
1287 /*
1288 * write to a queue. only Maxatomic bytes at a time is atomic.
1289 */
1290 int
qwrite(Queue * q,void * vp,int len)1291 qwrite(Queue *q, void *vp, int len)
1292 {
1293 int n, sofar;
1294 Block *b;
1295 uchar *p = vp;
1296
1297 sofar = 0;
1298 do {
1299 n = len-sofar;
1300 if(n > Maxatomic)
1301 n = Maxatomic;
1302
1303 b = allocb(n);
1304 setmalloctag(b, getcallerpc(&q));
1305 if(waserror()){
1306 freeb(b);
1307 nexterror();
1308 }
1309 memmove(b->wp, p+sofar, n);
1310 poperror();
1311 b->wp += n;
1312
1313 qbwrite(q, b);
1314
1315 sofar += n;
1316 } while(sofar < len && (q->state & Qmsg) == 0);
1317
1318 return len;
1319 }
1320
1321 /*
1322 * used by print() to write to a queue. Since we may be splhi or not in
1323 * a process, don't qlock.
1324 */
1325 int
qiwrite(Queue * q,void * vp,int len)1326 qiwrite(Queue *q, void *vp, int len)
1327 {
1328 int n, sofar, dowakeup;
1329 Block *b;
1330 uchar *p = vp;
1331
1332 dowakeup = 0;
1333
1334 sofar = 0;
1335 do {
1336 n = len-sofar;
1337 if(n > Maxatomic)
1338 n = Maxatomic;
1339
1340 b = iallocb(n);
1341 if (b == 0) {
1342 print("qiwrite: iallocb failed\n");
1343 break;
1344 }
1345 memmove(b->wp, p+sofar, n);
1346 b->wp += n;
1347
1348 lock(&q->l);
1349
1350 QDEBUG checkb(b, "qiwrite");
1351 if(q->bfirst)
1352 q->blast->next = b;
1353 else
1354 q->bfirst = b;
1355 q->blast = b;
1356 q->len += BALLOC(b);
1357 q->dlen += n;
1358
1359 if(q->state & Qstarve){
1360 q->state &= ~Qstarve;
1361 dowakeup = 1;
1362 }
1363
1364 unlock(&q->l);
1365
1366 if(dowakeup){
1367 if(q->kick)
1368 q->kick(q->arg);
1369 Wakeup(&q->rr);
1370 }
1371
1372 sofar += n;
1373 } while(sofar < len && (q->state & Qmsg) == 0);
1374
1375 return sofar;
1376 }
1377
1378 /*
1379 * be extremely careful when calling this,
1380 * as there is no reference accounting
1381 */
1382 void
qfree(Queue * q)1383 qfree(Queue *q)
1384 {
1385 qclose(q);
1386 free(q);
1387 }
1388
1389 /*
1390 * Mark a queue as closed. No further IO is permitted.
1391 * All blocks are released.
1392 */
1393 void
qclose(Queue * q)1394 qclose(Queue *q)
1395 {
1396 Block *bfirst;
1397
1398 if(q == nil)
1399 return;
1400
1401 /* mark it */
1402 lock(&q->l);
1403 q->state |= Qclosed;
1404 q->state &= ~(Qflow|Qstarve);
1405 strcpy(q->err, Ehungup);
1406 bfirst = q->bfirst;
1407 q->bfirst = 0;
1408 q->len = 0;
1409 q->dlen = 0;
1410 q->noblock = 0;
1411 unlock(&q->l);
1412
1413 /* free queued blocks */
1414 freeblist(bfirst);
1415
1416 /* wake up readers/writers */
1417 Wakeup(&q->rr);
1418 Wakeup(&q->wr);
1419 }
1420
1421 /*
1422 * Mark a queue as closed. Wakeup any readers. Don't remove queued
1423 * blocks.
1424 */
1425 void
qhangup(Queue * q,char * msg)1426 qhangup(Queue *q, char *msg)
1427 {
1428 /* mark it */
1429 lock(&q->l);
1430 q->state |= Qclosed;
1431 if(msg == 0 || *msg == 0)
1432 strcpy(q->err, Ehungup);
1433 else
1434 kstrcpy(q->err, msg, sizeof q->err);
1435 unlock(&q->l);
1436
1437 /* wake up readers/writers */
1438 Wakeup(&q->rr);
1439 Wakeup(&q->wr);
1440 }
1441
1442 /*
1443 * return non-zero if the q is hungup
1444 */
1445 int
qisclosed(Queue * q)1446 qisclosed(Queue *q)
1447 {
1448 return q->state & Qclosed;
1449 }
1450
1451 /*
1452 * mark a queue as no longer hung up
1453 */
1454 void
qreopen(Queue * q)1455 qreopen(Queue *q)
1456 {
1457 lock(&q->l);
1458 q->state &= ~Qclosed;
1459 q->state |= Qstarve;
1460 q->eof = 0;
1461 q->limit = q->inilim;
1462 unlock(&q->l);
1463 }
1464
1465 /*
1466 * return bytes queued
1467 */
1468 int
qlen(Queue * q)1469 qlen(Queue *q)
1470 {
1471 return q->dlen;
1472 }
1473
1474 /*
1475 * return space remaining before flow control
1476 */
1477 int
qwindow(Queue * q)1478 qwindow(Queue *q)
1479 {
1480 int l;
1481
1482 l = q->limit - q->len;
1483 if(l < 0)
1484 l = 0;
1485 return l;
1486 }
1487
1488 /*
1489 * return true if we can read without blocking
1490 */
1491 int
qcanread(Queue * q)1492 qcanread(Queue *q)
1493 {
1494 return q->bfirst!=0;
1495 }
1496
1497 /*
1498 * change queue limit
1499 */
1500 void
qsetlimit(Queue * q,int limit)1501 qsetlimit(Queue *q, int limit)
1502 {
1503 q->limit = limit;
1504 }
1505
1506 /*
1507 * set blocking/nonblocking
1508 */
1509 void
qnoblock(Queue * q,int onoff)1510 qnoblock(Queue *q, int onoff)
1511 {
1512 q->noblock = onoff;
1513 }
1514
1515 /*
1516 * flush the output queue
1517 */
1518 void
qflush(Queue * q)1519 qflush(Queue *q)
1520 {
1521 Block *bfirst;
1522
1523 /* mark it */
1524 lock(&q->l);
1525 bfirst = q->bfirst;
1526 q->bfirst = 0;
1527 q->len = 0;
1528 q->dlen = 0;
1529 unlock(&q->l);
1530
1531 /* free queued blocks */
1532 freeblist(bfirst);
1533
1534 /* wake up readers/writers */
1535 Wakeup(&q->wr);
1536 }
1537
1538 int
qfull(Queue * q)1539 qfull(Queue *q)
1540 {
1541 return q->state & Qflow;
1542 }
1543
1544 int
qstate(Queue * q)1545 qstate(Queue *q)
1546 {
1547 return q->state;
1548 }
1549
1550 int
qclosed(Queue * q)1551 qclosed(Queue *q)
1552 {
1553 return q->state & Qclosed;
1554 }
1555