ref: 8c9d28f4f8ab0305689f9f1f95e137e2f20e375c
dir: /sys/src/libthread/channel.c/
#include <u.h> #include <libc.h> #include <thread.h> #include "threadimpl.h" /* Value to indicate the channel is closed */ enum { CHANCLOSD = 0xc105ed, }; static char errcl[] = "channel was closed"; static Lock chanlock; /* central channel access lock */ static void enqueue(Alt*, Channel**); static void dequeue(Alt*); static int canexec(Alt*); static int altexec(Alt*, int); #define Closed ((void*)CHANCLOSD) #define Intred ((void*)~0) /* interrupted */ static void _chanfree(Channel *c) { int i, inuse; if(c->closed == 1) /* chanclose is ongoing */ inuse = 1; else{ inuse = 0; for(i = 0; i < c->nentry; i++) /* alt ongoing */ if(c->qentry[i]) inuse = 1; } if(inuse) c->freed = 1; else{ if(c->qentry) free(c->qentry); free(c); } } void chanfree(Channel *c) { lock(&chanlock); _chanfree(c); unlock(&chanlock); } Channel* chancreate(int elemsize, int elemcnt) { Channel *c; if(elemcnt < 0 || elemsize <= 0) return nil; c = _threadmalloc(sizeof(Channel)+elemsize*elemcnt, 1); c->e = elemsize; c->s = elemcnt; _threaddebug(DBGCHAN, "chancreate %p", c); return c; } static int isopenfor(Channel *c, int op) { return c->closed == 0 || (op == CHANRCV && c->n > 0); } int alt(Alt *alts) { Alt *a, *xa, *ca; Channel volatile *c; int n, s, waiting, allreadycl; void* r; Thread *t; /* * The point of going splhi here is that note handlers * might reasonably want to use channel operations, * but that will hang if the note comes while we hold the * chanlock. Instead, we delay the note until we've dropped * the lock. */ t = _threadgetproc()->thread; if(t->moribund || _threadexitsallstatus) yield(); /* won't return */ s = _procsplhi(); lock(&chanlock); t->alt = alts; t->chan = Chanalt; /* test whether any channels can proceed */ n = 0; a = nil; for(xa=alts; xa->op!=CHANEND && xa->op!=CHANNOBLK; xa++){ xa->entryno = -1; if(xa->op == CHANNOP) continue; c = xa->c; if(c==nil){ unlock(&chanlock); _procsplx(s); t->chan = Channone; return -1; } if(isopenfor(c, xa->op) && canexec(xa)) if(nrand(++n) == 0) a = xa; } if(a==nil){ /* nothing can proceed */ if(xa->op == CHANNOBLK){ unlock(&chanlock); _procsplx(s); t->chan = Channone; if(xa->op == CHANNOBLK) return xa - alts; } /* enqueue on all channels open for us. */ c = nil; ca = nil; waiting = 0; allreadycl = 0; for(xa=alts; xa->op!=CHANEND; xa++) if(xa->op==CHANNOP) continue; else if(isopenfor(xa->c, xa->op)){ waiting = 1; enqueue(xa, &c); } else if(xa->err != errcl) ca = xa; else allreadycl = 1; if(waiting == 0) if(ca != nil){ /* everything was closed, select last channel */ ca->err = errcl; unlock(&chanlock); _procsplx(s); t->chan = Channone; return ca - alts; } else if(allreadycl){ /* everything was already closed */ unlock(&chanlock); _procsplx(s); t->chan = Channone; return -1; } /* * wait for successful rendezvous. * we can't just give up if the rendezvous * is interrupted -- someone else might come * along and try to rendezvous with us, so * we need to be here. * if the channel was closed, the op is done * and we flag an error for the entry. */ Again: unlock(&chanlock); _procsplx(s); r = _threadrendezvous(&c, 0); s = _procsplhi(); lock(&chanlock); if(r==Intred){ /* interrupted */ if(c!=nil) /* someone will meet us; go back */ goto Again; c = (Channel*)~0; /* so no one tries to meet us */ } /* dequeue from channels, find selected one */ a = nil; for(xa=alts; xa->op!=CHANEND; xa++){ if(xa->op==CHANNOP) continue; if(xa->c == c){ a = xa; a->err = nil; if(r == Closed) a->err = errcl; } dequeue(xa); } unlock(&chanlock); _procsplx(s); if(a == nil){ /* we were interrupted */ assert(c==(Channel*)~0); return -1; } }else altexec(a, s); /* unlocks chanlock, does splx */ _sched(); t->chan = Channone; return a - alts; } int chanclose(Channel *c) { Alt *a; int i, s; s = _procsplhi(); /* note handlers; see :/^alt */ lock(&chanlock); if(c->closed){ /* Already close; we fail but it's ok. don't print */ unlock(&chanlock); _procsplx(s); return -1; } c->closed = 1; /* Being closed */ /* * Locate entries that will fail due to close * (send, and receive if nothing buffered) and wake them up. * the situation cannot change because all queries * should be committed by now and new ones will find the channel * closed. We still need to take the lock during the iteration * because we can wake threads on qentrys we have not seen yet * as in alt and there would be a race in the access to *a. */ for(i = 0; i < c->nentry; i++){ if((a = c->qentry[i]) == nil || *a->tag != nil) continue; if(a->op != CHANSND && (a->op != CHANRCV || c->n != 0)) continue; *a->tag = c; unlock(&chanlock); _procsplx(s); while(_threadrendezvous(a->tag, Closed) == Intred) ; s = _procsplhi(); lock(&chanlock); } c->closed = 2; /* Fully closed */ if(c->freed) _chanfree(c); unlock(&chanlock); _procsplx(s); return 0; } int chanclosing(Channel *c) { int n, s; s = _procsplhi(); /* note handlers; see :/^alt */ lock(&chanlock); if(c->closed == 0) n = -1; else n = c->n; unlock(&chanlock); _procsplx(s); return n; } /* * superseded by chanclosing int chanisclosed(Channel *c) { return chanisclosing(c) >= 0; } */ static int runop(int op, Channel *c, void *v, int nb) { int r; Alt a[2]; /* * we could do this without calling alt, * but the only reason would be performance, * and i'm not convinced it matters. */ a[0].op = op; a[0].c = c; a[0].v = v; a[0].err = nil; a[1].op = CHANEND; if(nb) a[1].op = CHANNOBLK; switch(r=alt(a)){ case -1: /* interrupted */ return -1; case 1: /* nonblocking, didn't accomplish anything */ assert(nb); return 0; case 0: /* * Okay, but return -1 if the op is done because of a close. */ if(a[0].err != nil) return -1; return 1; default: fprint(2, "ERROR: channel alt returned %d\n", r); abort(); return -1; } } int recv(Channel *c, void *v) { return runop(CHANRCV, c, v, 0); } int nbrecv(Channel *c, void *v) { return runop(CHANRCV, c, v, 1); } int send(Channel *c, void *v) { return runop(CHANSND, c, v, 0); } int nbsend(Channel *c, void *v) { return runop(CHANSND, c, v, 1); } static void channelsize(Channel *c, int sz) { if(c->e != sz){ fprint(2, "expected channel with elements of size %d, got size %d\n", sz, c->e); abort(); } } int sendul(Channel *c, ulong v) { channelsize(c, sizeof(ulong)); return send(c, &v); } ulong recvul(Channel *c) { ulong v; channelsize(c, sizeof(ulong)); if(recv(c, &v) < 0) return ~0; return v; } int sendp(Channel *c, void *v) { channelsize(c, sizeof(void*)); return send(c, &v); } void* recvp(Channel *c) { void *v; channelsize(c, sizeof(void*)); if(recv(c, &v) < 0) return nil; return v; } int nbsendul(Channel *c, ulong v) { channelsize(c, sizeof(ulong)); return nbsend(c, &v); } ulong nbrecvul(Channel *c) { ulong v; channelsize(c, sizeof(ulong)); if(nbrecv(c, &v) == 0) return 0; return v; } int nbsendp(Channel *c, void *v) { channelsize(c, sizeof(void*)); return nbsend(c, &v); } void* nbrecvp(Channel *c) { void *v; channelsize(c, sizeof(void*)); if(nbrecv(c, &v) == 0) return nil; return v; } static int emptyentry(Channel *c) { int i, extra; assert((c->nentry==0 && c->qentry==nil) || (c->nentry && c->qentry)); for(i=0; i<c->nentry; i++) if(c->qentry[i]==nil) return i; extra = 16; c->nentry += extra; c->qentry = realloc((void*)c->qentry, c->nentry*sizeof(c->qentry[0])); if(c->qentry == nil) sysfatal("realloc channel entries: %r"); memset(&c->qentry[i], 0, extra*sizeof(c->qentry[0])); return i; } static void enqueue(Alt *a, Channel **c) { int i; _threaddebug(DBGCHAN, "Queuing alt %p on channel %p", a, a->c); a->tag = c; i = emptyentry(a->c); a->c->qentry[i] = a; } static void dequeue(Alt *a) { int i; Channel *c; c = a->c; for(i=0; i<c->nentry; i++) if(c->qentry[i]==a){ _threaddebug(DBGCHAN, "Dequeuing alt %p from channel %p", a, a->c); c->qentry[i] = nil; /* release if freed and not closing */ if(c->freed && c->closed != 1) _chanfree(c); return; } } static int canexec(Alt *a) { int i, otherop; Channel *c; c = a->c; /* are there senders or receivers blocked? */ otherop = (CHANSND+CHANRCV) - a->op; for(i=0; i<c->nentry; i++) if(c->qentry[i] && c->qentry[i]->op==otherop && *c->qentry[i]->tag==nil){ _threaddebug(DBGCHAN, "can rendez alt %p chan %p", a, c); return 1; } /* is there room in the channel? */ if((a->op==CHANSND && c->n < c->s) || (a->op==CHANRCV && c->n > 0)){ _threaddebug(DBGCHAN, "can buffer alt %p chan %p", a, c); return 1; } return 0; } static void* altexecbuffered(Alt *a, int willreplace) { uchar *v; Channel *c; c = a->c; /* use buffered channel queue */ if(a->op==CHANRCV && c->n > 0){ _threaddebug(DBGCHAN, "buffer recv alt %p chan %p", a, c); v = c->v + c->e*(c->f%c->s); if(!willreplace) c->n--; c->f++; return v; } if(a->op==CHANSND && c->n < c->s){ _threaddebug(DBGCHAN, "buffer send alt %p chan %p", a, c); v = c->v + c->e*((c->f+c->n)%c->s); if(!willreplace) c->n++; return v; } abort(); return nil; } static void altcopy(void *dst, void *src, int sz) { if(dst){ if(src) memmove(dst, src, sz); else memset(dst, 0, sz); } } static int altexec(Alt *a, int spl) { volatile Alt *b; int i, n, otherop; Channel *c; void *me, *waiter, *buf; c = a->c; /* rendezvous with others */ otherop = (CHANSND+CHANRCV) - a->op; n = 0; b = nil; me = a->v; for(i=0; i<c->nentry; i++) if(c->qentry[i] && c->qentry[i]->op==otherop && *c->qentry[i]->tag==nil) if(nrand(++n) == 0) b = c->qentry[i]; if(b != nil){ _threaddebug(DBGCHAN, "rendez %s alt %p chan %p alt %p", a->op==CHANRCV?"recv":"send", a, c, b); waiter = b->v; if(c->s && c->n){ /* * if buffer is full and there are waiters * and we're meeting a waiter, * we must be receiving. * * we use the value in the channel buffer, * copy the waiter's value into the channel buffer * on behalf of the waiter, and then wake the waiter. */ if(a->op!=CHANRCV) abort(); buf = altexecbuffered(a, 1); altcopy(me, buf, c->e); altcopy(buf, waiter, c->e); }else{ if(a->op==CHANRCV) altcopy(me, waiter, c->e); else altcopy(waiter, me, c->e); } *b->tag = c; /* commits us to rendezvous */ _threaddebug(DBGCHAN, "unlocking the chanlock"); unlock(&chanlock); _procsplx(spl); _threaddebug(DBGCHAN, "chanlock is %lud", *(ulong*)&chanlock); while(_threadrendezvous(b->tag, 0) == Intred) ; return 1; } buf = altexecbuffered(a, 0); if(a->op==CHANRCV) altcopy(me, buf, c->e); else altcopy(buf, me, c->e); unlock(&chanlock); _procsplx(spl); return 1; }