Subversion Repositories planix.SVN

#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);
}

int
chaninit(Channel *c, int elemsize, int elemcnt)
{
        if(elemcnt < 0 || elemsize <= 0 || c == nil)
                return -1;
        c->f = 0;
        c->n = 0;
        c->closed = 0;
        c->freed = 0;
        c->e = elemsize;
        c->s = elemcnt;
        _threaddebug(DBGCHAN, "chaninit %p", c);
        return 1;
}

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;
}