Subversion Repositories planix.SVN

#include <u.h>
#include <libc.h>
#include <thread.h>
#include <sunrpc.h>

/*
 * Sun RPC server; for now, no reply cache
 */

static void sunRpcProc(void*);
static void sunRpcRequestThread(void*);
static void sunRpcReplyThread(void*);
static void sunRpcForkThread(void*);
static SunProg *sunFindProg(SunSrv*, SunMsg*, SunRpc*, Channel**);

typedef struct Targ Targ;
struct Targ
{
        void (*fn)(void*);
        void *arg;
};

SunSrv*
sunSrv(void)
{
        SunSrv *srv;

        srv = emalloc(sizeof(SunSrv));
        srv->chatty = 0;
        srv->crequest = chancreate(sizeof(SunMsg*), 16);
        srv->creply = chancreate(sizeof(SunMsg*), 16);
        srv->cthread = chancreate(sizeof(Targ), 4);

        proccreate(sunRpcProc, srv, SunStackSize);
        return srv;
}

void
sunSrvProg(SunSrv *srv, SunProg *prog, Channel *c)
{
        if(srv->nprog%16 == 0){
                srv->prog = erealloc(srv->prog, (srv->nprog+16)*sizeof(srv->prog[0]));
                srv->cdispatch = erealloc(srv->cdispatch, (srv->nprog+16)*sizeof(srv->cdispatch[0]));
        }
        srv->prog[srv->nprog] = prog;
        srv->cdispatch[srv->nprog] = c;
        srv->nprog++;
}

static void
sunRpcProc(void *v)
{
        threadcreate(sunRpcReplyThread, v, SunStackSize);
        threadcreate(sunRpcRequestThread, v, SunStackSize);
        threadcreate(sunRpcForkThread, v, SunStackSize);

}

static void
sunRpcForkThread(void *v)
{
        SunSrv *srv = v;
        Targ t;

        while(recv(srv->cthread, &t) == 1)
                threadcreate(t.fn, t.arg, SunStackSize);
}

void
sunSrvThreadCreate(SunSrv *srv, void (*fn)(void*), void *arg)
{
        Targ t;

        t.fn = fn;
        t.arg = arg;
        send(srv->cthread, &t);
}

static void
sunRpcRequestThread(void *v)
{
        uchar *p, *ep;
        Channel *c;
        SunSrv *srv = v;
        SunMsg *m;
        SunProg *pg;
        SunStatus ok;

        while((m = recvp(srv->crequest)) != nil){
                /* could look up in cache here? */

if(srv->chatty) fprint(2, "sun msg %p count %d\n", m, m->count);
                m->srv = srv;
                p = m->data;
                ep = p+m->count;
                if(sunRpcUnpack(p, ep, &p, &m->rpc) != SunSuccess){
                        fprint(2, "in: %.*H unpack failed\n", m->count, m->data);
                        sunMsgDrop(m);
                        continue;
                }
                if(srv->chatty)
                        fprint(2, "in: %B\n", &m->rpc);

                if(srv->alwaysReject){
                        if(srv->chatty)
                                fprint(2, "\trejecting\n");
                        sunMsgReplyError(m, SunAuthTooWeak);
                        continue;
                }

                if(!m->rpc.iscall){
                        sunMsgReplyError(m, SunGarbageArgs);
                        continue;
                }

                if((pg = sunFindProg(srv, m, &m->rpc, &c)) == nil){
                        /* sunFindProg sent error */
                        continue;
                }

                p = m->rpc.data;
                ep = p+m->rpc.ndata;
                m->call = nil;
                if((ok = sunCallUnpackAlloc(pg, m->rpc.proc<<1, p, ep, &p, &m->call)) != SunSuccess){
                        sunMsgReplyError(m, ok);
                        continue;
                }
                m->call->rpc = m->rpc;

                if(srv->chatty)
                        fprint(2, "\t%C\n", m->call);

                m->pg = pg;
                sendp(c, m);
        }
}

static SunProg*
sunFindProg(SunSrv *srv, SunMsg *m, SunRpc *rpc, Channel **pc)
{
        int i, vlo, vhi;
        SunProg *pg;

        vlo = 0x7fffffff;
        vhi = -1;

        for(i=0; i<srv->nprog; i++){
                pg = srv->prog[i];
                if(pg->prog != rpc->prog)
                        continue;
                if(pg->vers == rpc->vers){
                        *pc = srv->cdispatch[i];
                        return pg;
                }
                /* right program, wrong version: record range */
                if(pg->vers < vlo)
                        vlo = pg->vers;
                if(pg->vers > vhi)
                        vhi = pg->vers;
        }
        if(vhi == -1){
                if(srv->chatty)
                        fprint(2, "\tprogram %ud unavailable\n", rpc->prog);
                sunMsgReplyError(m, SunProgUnavail);
        }else{
                /* putting these in rpc is a botch */
                rpc->low = vlo;
                rpc->high = vhi;
                if(srv->chatty)
                        fprint(2, "\tversion %ud unavailable; have %d-%d\n", rpc->vers, vlo, vhi);
                sunMsgReplyError(m, SunProgMismatch);
        }
        return nil;
}

static void
sunRpcReplyThread(void *v)
{
        SunMsg *m;
        SunSrv *srv = v;

        while((m = recvp(srv->creply)) != nil){
                /* could record in cache here? */
                sendp(m->creply, m);
        }       
}

int
sunMsgReplyError(SunMsg *m, SunStatus error)
{
        uchar *p, *bp, *ep;
        int n;

        m->rpc.status = error;
        m->rpc.iscall = 0;
        m->rpc.verf.flavor = SunAuthNone;
        m->rpc.data = nil;
        m->rpc.ndata = 0;

        if(m->srv->chatty)
                fprint(2, "out: %B\n", &m->rpc);

        n = sunRpcSize(&m->rpc);
        bp = emalloc(n);
        ep = bp+n;
        p = bp;
        if(sunRpcPack(p, ep, &p, &m->rpc) < 0){
                fprint(2, "sunRpcPack failed\n");
                sunMsgDrop(m);
                return 0;
        }
        if(p != ep){
                fprint(2, "sunMsgReplyError: rpc sizes didn't work out\n");
                sunMsgDrop(m);
                return 0;
        }
        free(m->data);
        m->data = bp;
        m->count = n;
        sendp(m->srv->creply, m);
        return 0;
}

int
sunMsgReply(SunMsg *m, SunCall *c)
{
        int n1, n2;
        uchar *bp, *p, *ep;

        c->type = m->call->type+1;
        c->rpc.iscall = 0;
        c->rpc.prog = m->rpc.prog;
        c->rpc.vers = m->rpc.vers;
        c->rpc.proc = m->rpc.proc;
        c->rpc.xid = m->rpc.xid;

        if(m->srv->chatty){
                fprint(2, "out: %B\n", &c->rpc);
                fprint(2, "\t%C\n", c);
        }

        n1 = sunRpcSize(&c->rpc);
        n2 = sunCallSize(m->pg, c);

        bp = emalloc(n1+n2);
        ep = bp+n1+n2;
        p = bp;
        if(sunRpcPack(p, ep, &p, &c->rpc) != SunSuccess){
                fprint(2, "sunRpcPack failed\n");
                return sunMsgDrop(m);
        }
        if(sunCallPack(m->pg, p, ep, &p, c) != SunSuccess){
                fprint(2, "pg->pack failed\n");
                return sunMsgDrop(m);
        }
        if(p != ep){
                fprint(2, "sunMsgReply: sizes didn't work out\n");
                return sunMsgDrop(m);
        }
        free(m->data);
        m->data = bp;
        m->count = n1+n2;

        sendp(m->srv->creply, m);
        return 0;
}

int
sunMsgDrop(SunMsg *m)
{
        free(m->data);
        free(m->call);
        memset(m, 0xFB, sizeof *m);
        free(m);
        return 0;
}