ref: a97255112d90b1c776d610ac360a857af87556f3
dir: /kern/qio.c/
#include "u.h"
#include "lib.h"
#include "dat.h"
#include "fns.h"
#include "error.h"
static ulong padblockcnt;
static ulong concatblockcnt;
static ulong pullupblockcnt;
static ulong copyblockcnt;
static ulong consumecnt;
static ulong producecnt;
static ulong qcopycnt;
static int debugging;
#define QDEBUG if(0)
/*
* IO queues
*/
struct Queue
{
Lock lk;
Block* bfirst; /* buffer */
Block* blast;
int len; /* bytes allocated to queue */
int dlen; /* data bytes in queue */
int limit; /* max bytes in queue */
int inilim; /* initial limit */
int state;
int noblock; /* true if writes return immediately when q full */
int eof; /* number of eofs read by user */
void (*kick)(void*); /* restart output */
void (*bypass)(void*, Block*); /* bypass queue altogether */
void* arg; /* argument to kick */
QLock rlock; /* mutex for reading processes */
Rendez rr; /* process waiting to read */
QLock wlock; /* mutex for writing processes */
Rendez wr; /* process waiting to write */
char err[ERRMAX];
};
enum
{
Maxatomic = 64*1024,
};
uint qiomaxatomic = Maxatomic;
void
ixsummary(void)
{
debugging ^= 1;
iallocsummary();
print("pad %lud, concat %lud, pullup %lud, copy %lud\n",
padblockcnt, concatblockcnt, pullupblockcnt, copyblockcnt);
print("consume %lud, produce %lud, qcopy %lud\n",
consumecnt, producecnt, qcopycnt);
}
/*
* free a list of blocks
*/
void
freeblist(Block *b)
{
Block *next;
for(; b != 0; b = next){
next = b->next;
b->next = 0;
freeb(b);
}
}
/*
* pad a block to the front (or the back if size is negative)
*/
Block*
padblock(Block *bp, int size)
{
int n;
Block *nbp;
QDEBUG checkb(bp, "padblock 1");
if(size >= 0){
if(bp->rp - bp->base >= size){
bp->rp -= size;
return bp;
}
if(bp->next)
panic("padblock 0x%p", getcallerpc(&bp));
n = BLEN(bp);
padblockcnt++;
nbp = allocb(size+n);
nbp->rp += size;
nbp->wp = nbp->rp;
memmove(nbp->wp, bp->rp, n);
nbp->wp += n;
freeb(bp);
nbp->rp -= size;
} else {
size = -size;
if(bp->next)
panic("padblock 0x%p", getcallerpc(&bp));
if(bp->lim - bp->wp >= size)
return bp;
n = BLEN(bp);
padblockcnt++;
nbp = allocb(size+n);
memmove(nbp->wp, bp->rp, n);
nbp->wp += n;
freeb(bp);
}
QDEBUG checkb(nbp, "padblock 1");
return nbp;
}
/*
* return count of bytes in a string of blocks
*/
int
blocklen(Block *bp)
{
int len;
len = 0;
while(bp) {
len += BLEN(bp);
bp = bp->next;
}
return len;
}
/*
* return count of space in blocks
*/
int
blockalloclen(Block *bp)
{
int len;
len = 0;
while(bp) {
len += BALLOC(bp);
bp = bp->next;
}
return len;
}
/*
* copy the string of blocks into
* a single block and free the string
*/
Block*
concatblock(Block *bp)
{
int len;
Block *nb, *f;
if(bp->next == 0)
return bp;
nb = allocb(blocklen(bp));
for(f = bp; f; f = f->next) {
len = BLEN(f);
memmove(nb->wp, f->rp, len);
nb->wp += len;
}
concatblockcnt += BLEN(nb);
freeblist(bp);
QDEBUG checkb(nb, "concatblock 1");
return nb;
}
/*
* make sure the first block has at least n bytes
*/
Block*
pullupblock(Block *bp, int n)
{
int i;
Block *nbp;
/*
* this should almost always be true, it's
* just to avoid every caller checking.
*/
if(BLEN(bp) >= n)
return bp;
/*
* if not enough room in the first block,
* add another to the front of the list.
*/
if(bp->lim - bp->rp < n){
nbp = allocb(n);
nbp->next = bp;
bp = nbp;
}
/*
* copy bytes from the trailing blocks into the first
*/
n -= BLEN(bp);
while((nbp = bp->next)){
i = BLEN(nbp);
if(i > n) {
memmove(bp->wp, nbp->rp, n);
pullupblockcnt++;
bp->wp += n;
nbp->rp += n;
QDEBUG checkb(bp, "pullupblock 1");
return bp;
} else {
/* shouldn't happen but why crash if it does */
if(i < 0){
print("pullup negative length packet\n");
i = 0;
}
memmove(bp->wp, nbp->rp, i);
pullupblockcnt++;
bp->wp += i;
bp->next = nbp->next;
nbp->next = 0;
freeb(nbp);
n -= i;
if(n == 0){
QDEBUG checkb(bp, "pullupblock 2");
return bp;
}
}
}
freeb(bp);
return 0;
}
/*
* make sure the first block has at least n bytes
*/
Block*
pullupqueue(Queue *q, int n)
{
Block *b;
if(BLEN(q->bfirst) >= n)
return q->bfirst;
q->bfirst = pullupblock(q->bfirst, n);
for(b = q->bfirst; b != nil && b->next != nil; b = b->next)
;
q->blast = b;
return q->bfirst;
}
/*
* trim to len bytes starting at offset
*/
Block *
trimblock(Block *bp, int offset, int len)
{
ulong l;
Block *nb, *startb;
QDEBUG checkb(bp, "trimblock 1");
if(blocklen(bp) < offset+len) {
freeblist(bp);
return nil;
}
while((l = BLEN(bp)) < offset) {
offset -= l;
nb = bp->next;
bp->next = nil;
freeb(bp);
bp = nb;
}
startb = bp;
bp->rp += offset;
while((l = BLEN(bp)) < len) {
len -= l;
bp = bp->next;
}
bp->wp -= (BLEN(bp) - len);
if(bp->next) {
freeblist(bp->next);
bp->next = nil;
}
return startb;
}
/*
* copy 'count' bytes into a new block
*/
Block*
copyblock(Block *bp, int count)
{
int l;
Block *nbp;
QDEBUG checkb(bp, "copyblock 0");
nbp = allocb(count);
for(; count > 0 && bp != 0; bp = bp->next){
l = BLEN(bp);
if(l > count)
l = count;
memmove(nbp->wp, bp->rp, l);
nbp->wp += l;
count -= l;
}
if(count > 0){
memset(nbp->wp, 0, count);
nbp->wp += count;
}
copyblockcnt++;
QDEBUG checkb(nbp, "copyblock 1");
return nbp;
}
Block*
adjustblock(Block* bp, int len)
{
int n;
Block *nbp;
if(len < 0){
freeb(bp);
return nil;
}
if(bp->rp+len > bp->lim){
nbp = copyblock(bp, len);
freeblist(bp);
QDEBUG checkb(nbp, "adjustblock 1");
return nbp;
}
n = BLEN(bp);
if(len > n)
memset(bp->wp, 0, len-n);
bp->wp = bp->rp+len;
QDEBUG checkb(bp, "adjustblock 2");
return bp;
}
/*
* throw away up to count bytes from a
* list of blocks. Return count of bytes
* thrown away.
*/
int
pullblock(Block **bph, int count)
{
Block *bp;
int n, bytes;
bytes = 0;
if(bph == nil)
return 0;
while(*bph != nil && count != 0) {
bp = *bph;
n = BLEN(bp);
if(count < n)
n = count;
bytes += n;
count -= n;
bp->rp += n;
QDEBUG checkb(bp, "pullblock ");
if(BLEN(bp) == 0) {
*bph = bp->next;
bp->next = nil;
freeb(bp);
}
}
return bytes;
}
/*
* get next block from a queue, return null if nothing there
*/
Block*
qget(Queue *q)
{
int dowakeup;
Block *b;
/* sync with qwrite */
ilock(&q->lk);
b = q->bfirst;
if(b == nil){
q->state |= Qstarve;
iunlock(&q->lk);
return nil;
}
q->bfirst = b->next;
b->next = 0;
q->len -= BALLOC(b);
q->dlen -= BLEN(b);
QDEBUG checkb(b, "qget");
/* if writer flow controlled, restart */
if((q->state & Qflow) && q->len < q->limit/2){
q->state &= ~Qflow;
dowakeup = 1;
} else
dowakeup = 0;
iunlock(&q->lk);
if(dowakeup)
wakeup(&q->wr);
return b;
}
/*
* throw away the next 'len' bytes in the queue
*/
int
qdiscard(Queue *q, int len)
{
Block *b;
int dowakeup, n, sofar;
ilock(&q->lk);
for(sofar = 0; sofar < len; sofar += n){
b = q->bfirst;
if(b == nil)
break;
QDEBUG checkb(b, "qdiscard");
n = BLEN(b);
if(n <= len - sofar){
q->bfirst = b->next;
b->next = 0;
q->len -= BALLOC(b);
q->dlen -= BLEN(b);
freeb(b);
} else {
n = len - sofar;
b->rp += n;
q->dlen -= n;
}
}
/*
* if writer flow controlled, restart
*
* This used to be
* q->len < q->limit/2
* but it slows down tcp too much for certain write sizes.
* I really don't understand it completely. It may be
* due to the queue draining so fast that the transmission
* stalls waiting for the app to produce more data. - presotto
*/
if((q->state & Qflow) && q->len < q->limit){
q->state &= ~Qflow;
dowakeup = 1;
} else
dowakeup = 0;
iunlock(&q->lk);
if(dowakeup)
wakeup(&q->wr);
return sofar;
}
/*
* Interrupt level copy out of a queue, return # bytes copied.
*/
int
qconsume(Queue *q, void *vp, int len)
{
Block *b;
int n, dowakeup;
uchar *p = vp;
Block *tofree = nil;
/* sync with qwrite */
ilock(&q->lk);
for(;;) {
b = q->bfirst;
if(b == 0){
q->state |= Qstarve;
iunlock(&q->lk);
return -1;
}
QDEBUG checkb(b, "qconsume 1");
n = BLEN(b);
if(n > 0)
break;
q->bfirst = b->next;
q->len -= BALLOC(b);
/* remember to free this */
b->next = tofree;
tofree = b;
};
if(n < len)
len = n;
memmove(p, b->rp, len);
consumecnt += n;
b->rp += len;
q->dlen -= len;
/* discard the block if we're done with it */
if((q->state & Qmsg) || len == n){
q->bfirst = b->next;
b->next = 0;
q->len -= BALLOC(b);
q->dlen -= BLEN(b);
/* remember to free this */
b->next = tofree;
tofree = b;
}
/* if writer flow controlled, restart */
if((q->state & Qflow) && q->len < q->limit/2){
q->state &= ~Qflow;
dowakeup = 1;
} else
dowakeup = 0;
iunlock(&q->lk);
if(dowakeup)
wakeup(&q->wr);
if(tofree != nil)
freeblist(tofree);
return len;
}
int
qpass(Queue *q, Block *b)
{
int dlen, len, dowakeup;
/* sync with qread */
dowakeup = 0;
ilock(&q->lk);
if(q->len >= q->limit){
freeblist(b);
iunlock(&q->lk);
return -1;
}
if(q->state & Qclosed){
freeblist(b);
iunlock(&q->lk);
return BALLOC(b);
}
/* add buffer to queue */
if(q->bfirst)
q->blast->next = b;
else
q->bfirst = b;
len = BALLOC(b);
dlen = BLEN(b);
QDEBUG checkb(b, "qpass");
while(b->next){
b = b->next;
QDEBUG checkb(b, "qpass");
len += BALLOC(b);
dlen += BLEN(b);
}
q->blast = b;
q->len += len;
q->dlen += dlen;
if(q->len >= q->limit/2)
q->state |= Qflow;
if(q->state & Qstarve){
q->state &= ~Qstarve;
dowakeup = 1;
}
iunlock(&q->lk);
if(dowakeup)
wakeup(&q->rr);
return len;
}
int
qpassnolim(Queue *q, Block *b)
{
int dlen, len, dowakeup;
/* sync with qread */
dowakeup = 0;
ilock(&q->lk);
if(q->state & Qclosed){
freeblist(b);
iunlock(&q->lk);
return BALLOC(b);
}
/* add buffer to queue */
if(q->bfirst)
q->blast->next = b;
else
q->bfirst = b;
len = BALLOC(b);
dlen = BLEN(b);
QDEBUG checkb(b, "qpass");
while(b->next){
b = b->next;
QDEBUG checkb(b, "qpass");
len += BALLOC(b);
dlen += BLEN(b);
}
q->blast = b;
q->len += len;
q->dlen += dlen;
if(q->len >= q->limit/2)
q->state |= Qflow;
if(q->state & Qstarve){
q->state &= ~Qstarve;
dowakeup = 1;
}
iunlock(&q->lk);
if(dowakeup)
wakeup(&q->rr);
return len;
}
/*
* if the allocated space is way out of line with the used
* space, reallocate to a smaller block
*/
Block*
packblock(Block *bp)
{
Block **l, *nbp;
int n;
for(l = &bp; *l; l = &(*l)->next){
nbp = *l;
n = BLEN(nbp);
if((n<<2) < BALLOC(nbp)){
*l = allocb(n);
memmove((*l)->wp, nbp->rp, n);
(*l)->wp += n;
(*l)->next = nbp->next;
freeb(nbp);
}
}
return bp;
}
int
qproduce(Queue *q, void *vp, int len)
{
Block *b;
int dowakeup;
uchar *p = vp;
/* sync with qread */
dowakeup = 0;
ilock(&q->lk);
/* no waiting receivers, room in buffer? */
if(q->len >= q->limit){
q->state |= Qflow;
iunlock(&q->lk);
return -1;
}
/* save in buffer */
b = iallocb(len);
if(b == 0){
iunlock(&q->lk);
return 0;
}
memmove(b->wp, p, len);
producecnt += len;
b->wp += len;
if(q->bfirst)
q->blast->next = b;
else
q->bfirst = b;
q->blast = b;
/* b->next = 0; done by iallocb() */
q->len += BALLOC(b);
q->dlen += BLEN(b);
QDEBUG checkb(b, "qproduce");
if(q->state & Qstarve){
q->state &= ~Qstarve;
dowakeup = 1;
}
if(q->len >= q->limit)
q->state |= Qflow;
iunlock(&q->lk);
if(dowakeup)
wakeup(&q->rr);
return len;
}
/*
* copy from offset in the queue
*/
Block*
qcopy(Queue *q, int len, ulong offset)
{
int sofar;
int n;
Block *b, *nb;
uchar *p;
nb = allocb(len);
ilock(&q->lk);
/* go to offset */
b = q->bfirst;
for(sofar = 0; ; sofar += n){
if(b == nil){
iunlock(&q->lk);
return nb;
}
n = BLEN(b);
if(sofar + n > offset){
p = b->rp + offset - sofar;
n -= offset - sofar;
break;
}
QDEBUG checkb(b, "qcopy");
b = b->next;
}
/* copy bytes from there */
for(sofar = 0; sofar < len;){
if(n > len - sofar)
n = len - sofar;
memmove(nb->wp, p, n);
qcopycnt += n;
sofar += n;
nb->wp += n;
b = b->next;
if(b == nil)
break;
n = BLEN(b);
p = b->rp;
}
iunlock(&q->lk);
return nb;
}
/*
* called by non-interrupt code
*/
Queue*
qopen(int limit, int msg, void (*kick)(void*), void *arg)
{
Queue *q;
q = malloc(sizeof(Queue));
if(q == 0)
return 0;
q->limit = q->inilim = limit;
q->kick = kick;
q->arg = arg;
q->state = msg;
q->state |= Qstarve;
q->eof = 0;
q->noblock = 0;
return q;
}
/* open a queue to be bypassed */
Queue*
qbypass(void (*bypass)(void*, Block*), void *arg)
{
Queue *q;
q = malloc(sizeof(Queue));
if(q == 0)
return 0;
q->limit = 0;
q->arg = arg;
q->bypass = bypass;
q->state = 0;
return q;
}
static int
notempty(void *a)
{
Queue *q = a;
return (q->state & Qclosed) || q->bfirst != 0;
}
/*
* wait for the queue to be non-empty or closed.
* called with q ilocked.
*/
static int
qwait(Queue *q)
{
/* wait for data */
for(;;){
if(q->bfirst != nil)
break;
if(q->state & Qclosed){
if(++q->eof > 3)
return -1;
if(*q->err && strcmp(q->err, Ehungup) != 0)
return -1;
return 0;
}
q->state |= Qstarve; /* flag requesting producer to wake me */
iunlock(&q->lk);
sleep(&q->rr, notempty, q);
ilock(&q->lk);
}
return 1;
}
/*
* add a block list to a queue
*/
void
qaddlist(Queue *q, Block *b)
{
/* queue the block */
if(q->bfirst)
q->blast->next = b;
else
q->bfirst = b;
q->len += blockalloclen(b);
q->dlen += blocklen(b);
while(b->next)
b = b->next;
q->blast = b;
}
/*
* called with q ilocked
*/
Block*
qremove(Queue *q)
{
Block *b;
b = q->bfirst;
if(b == nil)
return nil;
q->bfirst = b->next;
b->next = nil;
q->dlen -= BLEN(b);
q->len -= BALLOC(b);
QDEBUG checkb(b, "qremove");
return b;
}
/*
* copy the contents of a string of blocks into
* memory. emptied blocks are freed. return
* pointer to first unconsumed block.
*/
Block*
bl2mem(uchar *p, Block *b, int n)
{
int i;
Block *next;
for(; b != nil; b = next){
i = BLEN(b);
if(i > n){
memmove(p, b->rp, n);
b->rp += n;
return b;
}
memmove(p, b->rp, i);
n -= i;
p += i;
b->rp += i;
next = b->next;
freeb(b);
}
return nil;
}
/*
* copy the contents of memory into a string of blocks.
* return nil on error.
*/
Block*
mem2bl(uchar *p, int len)
{
int n;
Block *b, *first, **l;
first = nil;
l = &first;
if(waserror()){
freeblist(first);
nexterror();
}
do {
n = len;
if(n > Maxatomic)
n = Maxatomic;
*l = b = allocb(n);
/* setmalloctag(b, (up->text[0]<<24)|(up->text[1]<<16)|(up->text[2]<<8)|up->text[3]); */
memmove(b->wp, p, n);
b->wp += n;
p += n;
len -= n;
l = &b->next;
} while(len > 0);
poperror();
return first;
}
/*
* put a block back to the front of the queue
* called with q ilocked
*/
void
qputback(Queue *q, Block *b)
{
b->next = q->bfirst;
if(q->bfirst == nil)
q->blast = b;
q->bfirst = b;
q->len += BALLOC(b);
q->dlen += BLEN(b);
}
/*
* flow control, get producer going again
* called with q ilocked
*/
static void
qwakeup_iunlock(Queue *q)
{
int dowakeup = 0;
/* if writer flow controlled, restart */
if((q->state & Qflow) && q->len < q->limit/2){
q->state &= ~Qflow;
dowakeup = 1;
}
iunlock(&q->lk);
/* wakeup flow controlled writers */
if(dowakeup){
if(q->kick)
q->kick(q->arg);
wakeup(&q->wr);
}
}
/*
* get next block from a queue (up to a limit)
*/
Block*
qbread(Queue *q, int len)
{
Block *b, *nb;
int n;
qlock(&q->rlock);
if(waserror()){
qunlock(&q->rlock);
nexterror();
}
ilock(&q->lk);
switch(qwait(q)){
case 0:
/* queue closed */
iunlock(&q->lk);
qunlock(&q->rlock);
poperror();
return nil;
case -1:
/* multiple reads on a closed queue */
iunlock(&q->lk);
error(q->err);
}
/* if we get here, there's at least one block in the queue */
b = qremove(q);
n = BLEN(b);
/* split block if it's too big and this is not a message queue */
nb = b;
if(n > len){
if((q->state&Qmsg) == 0){
n -= len;
b = allocb(n);
memmove(b->wp, nb->rp+len, n);
b->wp += n;
qputback(q, b);
}
nb->wp = nb->rp + len;
}
/* restart producer */
qwakeup_iunlock(q);
poperror();
qunlock(&q->rlock);
return nb;
}
/*
* read a queue. if no data is queued, post a Block
* and wait on its Rendez.
*/
long
qread(Queue *q, void *vp, int len)
{
Block *b, *first, **l;
int m, n;
qlock(&q->rlock);
if(waserror()){
qunlock(&q->rlock);
nexterror();
}
ilock(&q->lk);
again:
switch(qwait(q)){
case 0:
/* queue closed */
iunlock(&q->lk);
qunlock(&q->rlock);
poperror();
return 0;
case -1:
/* multiple reads on a closed queue */
iunlock(&q->lk);
error(q->err);
}
/* if we get here, there's at least one block in the queue */
if(q->state & Qcoalesce){
/* when coalescing, 0 length blocks just go away */
b = q->bfirst;
if(BLEN(b) <= 0){
freeb(qremove(q));
goto again;
}
/* grab the first block plus as many
* following blocks as will completely
* fit in the read.
*/
n = 0;
l = &first;
m = BLEN(b);
for(;;) {
*l = qremove(q);
l = &b->next;
n += m;
b = q->bfirst;
if(b == nil)
break;
m = BLEN(b);
if(n+m > len)
break;
}
} else {
first = qremove(q);
n = BLEN(first);
}
/* copy to user space outside of the ilock */
iunlock(&q->lk);
b = bl2mem(vp, first, len);
ilock(&q->lk);
/* take care of any left over partial block */
if(b != nil){
n -= BLEN(b);
if(q->state & Qmsg)
freeb(b);
else
qputback(q, b);
}
/* restart producer */
qwakeup_iunlock(q);
poperror();
qunlock(&q->rlock);
return n;
}
static int
qnotfull(void *a)
{
Queue *q = a;
return q->len < q->limit || (q->state & Qclosed);
}
ulong noblockcnt;
/*
* add a block to a queue obeying flow control
*/
long
qbwrite(Queue *q, Block *b)
{
int n, dowakeup;
n = BLEN(b);
if(q->bypass){
(*q->bypass)(q->arg, b);
return n;
}
dowakeup = 0;
qlock(&q->wlock);
if(waserror()){
if(b != nil)
freeb(b);
qunlock(&q->wlock);
nexterror();
}
ilock(&q->lk);
/* give up if the queue is closed */
if(q->state & Qclosed){
iunlock(&q->lk);
error(q->err);
}
/* if nonblocking, don't queue over the limit */
if(q->len >= q->limit){
if(q->noblock){
iunlock(&q->lk);
freeb(b);
noblockcnt += n;
qunlock(&q->wlock);
poperror();
return n;
}
}
/* queue the block */
if(q->bfirst)
q->blast->next = b;
else
q->bfirst = b;
q->blast = b;
b->next = 0;
q->len += BALLOC(b);
q->dlen += n;
QDEBUG checkb(b, "qbwrite");
b = nil;
/* make sure other end gets awakened */
if(q->state & Qstarve){
q->state &= ~Qstarve;
dowakeup = 1;
}
iunlock(&q->lk);
/* get output going again */
if(q->kick && (dowakeup || (q->state&Qkick)))
q->kick(q->arg);
/* wakeup anyone consuming at the other end */
if(dowakeup){
wakeup(&q->rr);
/* if we just wokeup a higher priority process, let it run */
/*
p = wakeup(&q->rr);
if(p != nil && p->priority > up->priority)
sched();
*/
}
/*
* flow control, wait for queue to get below the limit
* before allowing the process to continue and queue
* more. We do this here so that postnote can only
* interrupt us after the data has been queued. This
* means that things like 9p flushes and ssl messages
* will not be disrupted by software interrupts.
*
* Note - this is moderately dangerous since a process
* that keeps getting interrupted and rewriting will
* queue infinite crud.
*/
for(;;){
if(q->noblock || qnotfull(q))
break;
ilock(&q->lk);
q->state |= Qflow;
iunlock(&q->lk);
sleep(&q->wr, qnotfull, q);
}
USED(b);
qunlock(&q->wlock);
poperror();
return n;
}
/*
* write to a queue. only Maxatomic bytes at a time is atomic.
*/
int
qwrite(Queue *q, void *vp, int len)
{
int n, sofar;
Block *b;
uchar *p = vp;
QDEBUG if(!islo())
print("qwrite hi %p\n", getcallerpc(&q));
sofar = 0;
do {
n = len-sofar;
if(n > Maxatomic)
n = Maxatomic;
b = allocb(n);
/* setmalloctag(b, (up->text[0]<<24)|(up->text[1]<<16)|(up->text[2]<<8)|up->text[3]); */
if(waserror()){
freeb(b);
nexterror();
}
memmove(b->wp, p+sofar, n);
poperror();
b->wp += n;
qbwrite(q, b);
sofar += n;
} while(sofar < len && (q->state & Qmsg) == 0);
return len;
}
/*
* used by print() to write to a queue. Since we may be splhi or not in
* a process, don't qlock.
*/
int
qiwrite(Queue *q, void *vp, int len)
{
int n, sofar, dowakeup;
Block *b;
uchar *p = vp;
dowakeup = 0;
sofar = 0;
do {
n = len-sofar;
if(n > Maxatomic)
n = Maxatomic;
b = iallocb(n);
if(b == nil)
break;
memmove(b->wp, p+sofar, n);
b->wp += n;
ilock(&q->lk);
QDEBUG checkb(b, "qiwrite");
if(q->bfirst)
q->blast->next = b;
else
q->bfirst = b;
q->blast = b;
q->len += BALLOC(b);
q->dlen += n;
if(q->state & Qstarve){
q->state &= ~Qstarve;
dowakeup = 1;
}
iunlock(&q->lk);
if(dowakeup){
if(q->kick)
q->kick(q->arg);
wakeup(&q->rr);
}
sofar += n;
} while(sofar < len && (q->state & Qmsg) == 0);
return sofar;
}
/*
* be extremely careful when calling this,
* as there is no reference accounting
*/
void
qfree(Queue *q)
{
qclose(q);
free(q);
}
/*
* Mark a queue as closed. No further IO is permitted.
* All blocks are released.
*/
void
qclose(Queue *q)
{
Block *bfirst;
if(q == nil)
return;
/* mark it */
ilock(&q->lk);
q->state |= Qclosed;
q->state &= ~(Qflow|Qstarve);
strcpy(q->err, Ehungup);
bfirst = q->bfirst;
q->bfirst = 0;
q->len = 0;
q->dlen = 0;
q->noblock = 0;
iunlock(&q->lk);
/* free queued blocks */
freeblist(bfirst);
/* wake up readers/writers */
wakeup(&q->rr);
wakeup(&q->wr);
}
/*
* Mark a queue as closed. Wakeup any readers. Don't remove queued
* blocks.
*/
void
qhangup(Queue *q, char *msg)
{
/* mark it */
ilock(&q->lk);
q->state |= Qclosed;
if(msg == 0 || *msg == 0)
strcpy(q->err, Ehungup);
else
strncpy(q->err, msg, ERRMAX-1);
iunlock(&q->lk);
/* wake up readers/writers */
wakeup(&q->rr);
wakeup(&q->wr);
}
/*
* return non-zero if the q is hungup
*/
int
qisclosed(Queue *q)
{
return q->state & Qclosed;
}
/*
* mark a queue as no longer hung up
*/
void
qreopen(Queue *q)
{
ilock(&q->lk);
q->state &= ~Qclosed;
q->state |= Qstarve;
q->eof = 0;
q->limit = q->inilim;
iunlock(&q->lk);
}
/*
* return bytes queued
*/
int
qlen(Queue *q)
{
return q->dlen;
}
/*
* return space remaining before flow control
*/
int
qwindow(Queue *q)
{
int l;
l = q->limit - q->len;
if(l < 0)
l = 0;
return l;
}
/*
* return true if we can read without blocking
*/
int
qcanread(Queue *q)
{
return q->bfirst!=0;
}
/*
* change queue limit
*/
void
qsetlimit(Queue *q, int limit)
{
q->limit = limit;
}
/*
* set blocking/nonblocking
*/
void
qnoblock(Queue *q, int onoff)
{
q->noblock = onoff;
}
/*
* flush the output queue
*/
void
qflush(Queue *q)
{
Block *bfirst;
/* mark it */
ilock(&q->lk);
bfirst = q->bfirst;
q->bfirst = 0;
q->len = 0;
q->dlen = 0;
iunlock(&q->lk);
/* free queued blocks */
freeblist(bfirst);
/* wake up readers/writers */
wakeup(&q->wr);
}
int
qfull(Queue *q)
{
return q->state & Qflow;
}
int
qstate(Queue *q)
{
return q->state;
}