arti/irix-657m-src
1
0
Fork 0
Code Activity
Files
main
irix-657m-src/irix/cmd/tk/test3.c
calmsacibis995 8fc8fa8089 Source code upload
2022-09-29 17:59:04 +03:00

1228 lines
27 KiB
C
Raw Permalink Blame History

#include "assert.h"
#include "errno.h"
#include "tkm.h"
#include "unistd.h"
#include "memory.h"
#include "stdlib.h"
#include "stdio.h"
#include "strings.h"
#include "getopt.h"
#include "ulocks.h"
#include "mutex.h"
#include "time.h"
#include "signal.h"
#include "sys/types.h"
#include "sys/prctl.h"
#include "mbox.h"
#include "syncv.h"
#define STKSIZE (1024*1024*1)
static void sig(int);
static void node(int, int);
#define TEST3_LOOKUP 1 /* c->s RPC */
#define TEST3_GETVALUE 2 /* c->s RPC */
#define TEST3_REVOKE 3 /* server -> client 1-way */
#define TEST3_RVALUE 5 /* c->s IPC/RPC */
union test3_req {
struct lookup_req {
unsigned long id;
} lookup_req;
struct getvalue_req {
tk_set_t obtain;
tk_set_t toreturn;
tk_disp_t why;
void *handle;
unsigned long value;
} getvalue_req;
struct revoke_req {
tk_set_t torevoke;
tk_set_t eh;
tk_disp_t which;
void *handle;
unsigned long id; /* used for debugging */
unsigned long value;
} revoke_req;
};
union test3_res {
struct lookup_res {
tk_set_t existance;
void *handle;
} lookup_res;
struct getvalue_res {
tk_set_t granted;
tk_set_t already;
unsigned long value;
unsigned long id;
} getvalue_res;
};
/*
* Service info
*/
#define MAXSERVICES 1
struct service_info {
int server_node; /* node that will act as server */
int service_id;
int cmpl; /* client multi-programming level */
void (*wait)(mesg_t *, int);
void (*start)(void *, size_t);
void *sdata; /* private server data */
void **cdata; /* private client data / node */
} services[MAXSERVICES];
#define S1_CP(n) (services[SERVICE1_ID].cdata[n])/* get private data for node */
#define S1_SP() (services[SERVICE1_ID].sdata)/* get server data */
#define SERVICE1_ID 0
/* are we the service1 server? */
#define S1S() (services[SERVICE1_ID].server_node)
#define AM_S1S(n) (n == services[SERVICE1_ID].server_node)
/*
* tokens
*/
#define EXIST_CLASS 0
#define EXIST_READ TK_MAKE(EXIST_CLASS, TK_READ)
#define DATA_CLASS 1
#define DATA_READ TK_MAKE(DATA_CLASS, TK_READ)
#define SDATA_READ TK_MAKE_SINGLETON(DATA_CLASS, TK_READ)
#define DATA_UPDATE TK_MAKE(DATA_CLASS, TK_WRITE)
#define SDATA_UPDATE TK_MAKE_SINGLETON(DATA_CLASS, TK_WRITE)
#define DATA_SET TK_ADD_SET(DATA_READ, DATA_UPDATE)
#define NTOKENS 2
/*
* server side
* data per node
*/
struct server {
usema_t *upd;
ulock_t lckvalue;
struct sdata *data; /* list of active sdata's */
};
/* server side distributed object */
struct sdata {
TKS_DECL(ss, NTOKENS);
unsigned long id; /* object id */
unsigned long value; /* object value */
struct sdata *next;
};
/*
* client side
* data per node
*/
struct client {
tks_ch_t h; /* handle (contains node we're on) */
int node; /* node we're on (for debug) */
int ndone; /* # threads done */
ulock_t *upd;
sv_t wait; /* wait for in-progress objects */
struct cdata *data; /* list of active cdata's */
};
/* client side of distributed object */
struct cdata {
TKC_DECL(cs, NTOKENS);
unsigned long id; /* object value */
unsigned long value; /* object value */
tks_ch_t h; /* client handle ... should be global */
void *handle; /* server handle */
int flags;
int ref;
ulock_t *upd;
struct cdata *next;
};
#define CL_INPROGRESS 1
/*
* htonode - convert a handle to a node number
*/
#define htonode(h) (h)
#define nodetoh(n) (n)
int nloops = 10;
pid_t ppid;
unsigned long ndoneclients;
unsigned long amlogging;
int verbose;
/*
* the following arena is used for all locks and semaphores for all nodes
* and all services ...
*/
usptr_t *usptr;
char *Cmd;
int nodes;
int dumplog = 0;
int maxobjects = 100;
static void service1_wait(mesg_t *, int);
static void service1_start(void *, size_t);
static void service1_test(int node);
int
main(int argc, char **argv)
{
int c, i;
int cmpl;
unsigned long totclients;
int nwaiters;
int usedebuglocks = 0;
setlinebuf(stdout);
setlinebuf(stderr);
Cmd = strdup(argv[0]);
ppid = getpid();
prctl(PR_COREPID, 0, 1);
cmpl = 3;
nodes = 4;
while ((c = getopt(argc, argv, "o:dTN:vLm:n:")) != EOF)
switch (c) {
case 'd':
usedebuglocks = 1;
break;
case 'T':
__tk_tracenow = 1;
break;
case 'v':
verbose++;
break;
case 'L':
dumplog++;
break;
case 'm': /* node multi-programming level */
cmpl = atoi(optarg);
break;
case 'o':
maxobjects = atoi(optarg);
break;
case 'N':
nodes = atoi(optarg);
break;
case 'n':
nloops = atoi(optarg);
break;
default:
fprintf(stderr, "test3:illegal option %c\n", c);
fprintf(stderr, "Usage:test3 [options]\n");
fprintf(stderr, "\t-n #\t# loops\n");
fprintf(stderr, "\t-N #\t# nodes\n");
fprintf(stderr, "\t-o #\t# objects\n");
fprintf(stderr, "\t-m #\tclient multi-programming level\n");
fprintf(stderr, "\t-v\tverbose\n");
exit(1);
}
printf("test3: nodes %d loops %d client-multiprogramming %d objects %d\n",
nodes, nloops, cmpl, maxobjects);
/*
* setup services
*/
for (i = 0; i < MAXSERVICES; i++)
services[i].service_id = -1;
services[0].server_node = 0;
services[0].service_id = SERVICE1_ID;
services[0].cmpl = cmpl;
services[0].wait = service1_wait;
services[0].start = service1_start;
services[0].cdata = (void **)malloc(nodes * sizeof(void *));
/*
* XXX we really need more why??
* 1) each service has 'cmpl' client threads per node
* 2) Each service server needs a thread per potential client thread
* that is not on its node
* 3) each client not on the server needs a 'wait' thread.
* XX each client thread could be calling the server but
* the server could also send out some RPCs that must be answered.
*/
totclients = nodes * cmpl;
nwaiters = (nodes+1) * cmpl;
nwaiters *= 2;
nwaiters += (nodes - 1);
usconfig(CONF_INITUSERS, nwaiters + totclients + 1);
usconfig(CONF_INITSIZE, (32*1024) * nodes);
sigset(SIGTERM, sig);
sigset(SIGABRT, sig);
sigset(SIGUSR1, sig);
sigset(SIGUSR2, sig);
sigset(SIGURG, sig);
prctl(PR_SETEXITSIG, SIGTERM);
/* initialize RPC */
initmbox("/usr/tmp/test3mbox");
/* call tkc_init to get arenas started up in correct sequence */
tkc_init();
tks_init();
/*
* alloc an area for locks - note that all nodes 'share' this
* but that should be OK
*/
if (usedebuglocks)
usconfig(CONF_LOCKTYPE, US_DEBUGPLUS);
usconfig(CONF_ARENATYPE, US_SHAREDONLY);
if ((usptr = usinit("/usr/tmp/test3")) == NULL)
abort();
/* create 'nodes' */
for (i = 0; i < nodes; i++)
node(i, nodes);
while (ndoneclients != totclients)
sginap(100);
if (dumplog)
tk_printlog(stdout, 1000, dumplog > 2 ? TK_LOG_ALL :
(dumplog > 1 ? TK_LOG_TS : 0), NULL);
return 0;
}
static void
sig(int s)
{
static FILE *f = NULL;
if (s != SIGTERM && f == NULL)
f = fopen("test3.LOG", "w");
if (s == SIGABRT) {
/* make sure only one process logs */
sighold(SIGTERM);
if (f == NULL)
f = stderr;
fprintf(f, "\n\n======ABORT=============\n\n");
if (test_and_set(&amlogging, 1) == 0)
tk_printlog(f, -1, TK_LOG_ALL, NULL);
fflush(f);
sigset(SIGABRT, SIG_DFL);
abort();
} else if (s == SIGUSR1 || s == SIGUSR2) {
if (f == NULL)
f = stdout;
if (amlogging)
return;
fprintf(f, "\n\n===SIGUSR===============\n\n");
tk_printlog(f, -1, s == SIGUSR2 ? TK_LOG_ALL : 0, NULL);
fflush(f);
return;
} else if (s == SIGURG) {
char criteria[128];
fprintf(stdout, "Criteria?:");
fflush(stdout);
if (fgets(criteria, sizeof(criteria), stdin) == NULL)
criteria[0] = '\0';
if (f == NULL)
f = stdout;
fprintf(f, "\n\n===SIGURG===============\n\n");
tk_printlog(f, -1, TK_LOG_ALL, criteria);
fflush(f);
return;
}
sighold(SIGTERM);
exit(0);
}
/*
* Node
*/
/*
* do all waits for ootb messages - both client and server
*/
/* ARGSUSED */
static void
node_wait(void *a, size_t sz)
{
int node = (int)(ptrdiff_t)a;
struct mbox *mbox;
auto mesg_t *m;
mbox = ntombox(node);
for (;;) {
readmbox(mbox, &m);
(services[m->service_id].wait)(m, node);
if (m->flags & MESG_RPC)
replymbox(mbox, m);
else
freemesg(m);
}
}
static void
node(int mynode, int nodes)
{
int i, j;
struct mbox *mb;
pid_t spid;
mb = allocmbox();
setmbox(mynode, mb);
for (i = 0; i < MAXSERVICES; i++) {
if (services[i].service_id < 0)
continue;
/* start up wait threads -
* if we are the server then we need 'cmpl' server threads
* per client node AND client wait threads.
* if we are a client we need 'cmpl' client wait threads
* but no server wait threads ..
*/
if (services[i].server_node == mynode) {
int ns;
ns = (nodes+1) * services[SERVICE1_ID].cmpl;
for (j = 0; j < ns; j++) {
if ((spid = sprocsp(node_wait, PR_SALL,
(void *)(ptrdiff_t)mynode,
NULL, STKSIZE)) < 0) {
perror("sproc");
exit(1);
}
if (verbose)
printf("%s:started up server wait thread on node %d pid %d\n",
Cmd, mynode, spid);
}
} else {
int ns;
ns = services[SERVICE1_ID].cmpl;
for (j = 0; j < ns; j++) {
if ((spid = sprocsp(node_wait, PR_SALL,
(void *)(ptrdiff_t)mynode,
NULL, STKSIZE)) < 0) {
perror("sproc");
exit(1);
}
if (verbose)
printf("%s:started up client wait thread on node %d pid %d\n",
Cmd, mynode, spid);
}
}
/*
* start up 'threads' that do testing
*/
if ((spid = sprocsp(services[i].start, PR_SALL,
(void *)(ptrdiff_t)mynode,
NULL, STKSIZE)) < 0) {
perror("sproc");
exit(1);
}
if (verbose)
printf("%s:started up service %d on node %d pid %d\n",
Cmd, i, mynode, spid);
}
}
/*
* token callout modules
*/
static void client_obtain(void *, tk_set_t, tk_set_t, tk_disp_t, tk_set_t *);
static void client_return(tkc_state_t, void *, tk_set_t, tk_set_t, tk_disp_t);
static void server_recall(void *o, tks_ch_t h, tk_set_t r, tk_disp_t);
static void server_recalled(void *, tk_set_t, tk_set_t);
static void server_idle(void *, tk_set_t);
tks_ifstate_t svriface = {
server_recall,
server_recalled,
server_idle
};
tkc_ifstate_t iface = {
client_obtain,
client_return
};
/*
* stubs and server implementation modules
*/
static int invoke_lookup(tks_ch_t, unsigned long, tk_set_t *, void **);
static int invoke_getvalue(tks_ch_t, tk_set_t, tk_set_t, tk_disp_t,
tk_set_t *, tk_set_t *, void *, unsigned long *, unsigned long *);
static int server_lookup(tks_ch_t h, unsigned long, tk_set_t *, void **);
static int server_getvalue(tks_ch_t, tk_set_t, tk_set_t, tk_disp_t,
tk_set_t *, tk_set_t *, void *, unsigned long *, unsigned long *);
static void server_return(tks_ch_t, unsigned long, void *, tk_set_t, tk_set_t, tk_disp_t);
static int getlock(struct cdata *, int);
static void putlock(struct cdata *, int);
static struct cdata *getid(struct client *cp, unsigned long id);
static void releaseid(struct client *, struct cdata *);
/* ARGSUSED */
static void
service1_launch(void *a, size_t sz)
{
service1_test((int)(ptrdiff_t)a);
/* NOTREACHED */
}
/*
* called once per node - sets up minimal state needed
*/
/* ARGSUSED */
void
service1_start(void *a, size_t sz)
{
int mynode = (int)(ptrdiff_t)a;
int i;
pid_t spid;
struct server sd;
struct client cd; /* this is what we're manipulating */
/* init a client data object */
cd.h = nodetoh(mynode); /* handle is just node number */
cd.node = mynode;
cd.upd = usnewlock(usptr);
sv_create(&cd.wait);
cd.data = NULL;
cd.ndone = 0;
/* place local data in 'fake' private memory */
S1_CP(mynode) = &cd;
if (AM_S1S(mynode)) {
/* init server side */
sd.upd = usnewsema(usptr, 1);
sd.data = NULL;
/* place local data in 'fake' private memory */
S1_SP() = &sd;
}
/*
* each client / server is multi-threaded
*/
for (i = 1; i < services[SERVICE1_ID].cmpl; i++) {
if ((spid = sprocsp(service1_launch, PR_SALL,
(void *)(ptrdiff_t)mynode,
NULL, STKSIZE)) < 0) {
perror("sproc");
exit(1);
}
if (verbose)
printf("%s:started up client on node %d pid %d\n",
Cmd, mynode, spid);
}
service1_test(mynode);
/* NOTREACHED */
}
/*
* called by all threads - they all simulate accessing a variety of
* like-objects
* each thread selects an object and a mode (read/write), gets access
* to it, plays with it a while and releases it. Upon last us
* the object is flushed back to the server
*/
static void
service1_test(int node)
{
struct client *cp;
struct cdata *cd;
int i, j, rw;
unsigned long id, nvalue;
auto unsigned int seed;
seed = getpid();
cp = S1_CP(node);
for (j = 0; j < nloops; j++) {
id = (unsigned long)(rand_r(&seed) % maxobjects);
rw = (rand_r(&seed) % 2) == 0? TK_READ : TK_WRITE;
if ((cd = getid(cp, id)) == NULL) {
/* some form of error */
if (verbose)
printf("test3:client %d failed to get object %lu\n",
node, id);
sginap(4);
continue;
}
if (verbose > 1)
printf("test3:client %d has object %d\n",
node, id);
/* play a bit */
for (i = 0; i < 10; i++) {
rw = (rand_r(&seed) % 2) == 0? TK_READ : TK_WRITE;
if (getlock(cd, rw)) {
/* error - server dead */
assert(0);
}
assert(cd->value);
if (rw == TK_WRITE) {
do {
nvalue = rand_r(&seed);
} while (nvalue == cd->value || nvalue == 0);
cd->value = nvalue;
}
sginap(rand_r(&seed) % 5);
putlock(cd, rw);
}
/* release object */
releaseid(cp, cd);
}
ussetlock(cp->upd);
cp->ndone++;
usunsetlock(cp->upd);
test_then_add(&ndoneclients, 1);
for (;;)
pause();
}
static struct cdata *
getid(struct client *cp, unsigned long id)
{
struct cdata *cd, **pcd;
tk_set_t granted;
int rv;
/* grab lock protecting client chain */
retry:
ussetlock(cp->upd);
for (cd = cp->data; cd; cd = cd->next) {
if (cd->id == id) {
/* have one already */
ussetlock(cd->upd);
if (cd->flags & CL_INPROGRESS) {
usunsetlock(cd->upd);
sv_wait(&cp->wait, cp->upd);
goto retry;
}
assert(cd->ref > 0);
cd->ref++;
usunsetlock(cd->upd);
usunsetlock(cp->upd);
return cd;
}
}
/* none there - create one, put onlist and mark INPROGRESS */
cd = malloc(sizeof(*cd));
cd->handle = NULL;
cd->id = id;
cd->flags = CL_INPROGRESS;
cd->ref = 1;
cd->h = cp->h;
cd->upd = usnewlock(usptr);
cd->next = cp->data;
cp->data = cd;
usunsetlock(cp->upd);
/* look up on server */
if (AM_S1S(cp->node)) {
rv = server_lookup(cp->h, id, &granted, &cd->handle);
} else {
rv = invoke_lookup(cp->h, id, &granted, &cd->handle);
}
assert(rv == 0);
/*
* if rv != 0 then the server id dead
* if granted is TK_NULLSET then server didn't have what we wanted
*/
if (rv == 0 && granted != TK_NULLSET) {
/* we now have existance token */
tkc_create("client", cd->cs, (void *)cd, &iface, NTOKENS,
granted, (void *)id);
ussetlock(cd->upd);
assert(cd->flags & CL_INPROGRESS);
cd->flags &= ~CL_INPROGRESS;
usunsetlock(cd->upd);
ussetlock(cp->upd);
sv_broadcast(&cp->wait);
usunsetlock(cp->upd);
} else {
/* some form of error */
ussetlock(cp->upd);
for (pcd = &cp->data; *pcd; pcd = &(*pcd)->next) {
if (*pcd == cd) {
*pcd = cd->next;
break;
}
}
usfreelock(cd->upd, usptr);
free(cd);
sv_broadcast(&cp->wait);
usunsetlock(cp->upd);
cd = NULL;
}
return cd;
}
/*
* releaseid - client all done with object
* called either from client app code or client_recall
*/
static void
releaseid(struct client *cp, struct cdata *cd)
{
tk_set_t toreturn;
tk_disp_t dofret;
struct cdata **pcd;
ussetlock(cd->upd);
assert(cd->ref > 0);
if (--cd->ref > 0) {
usunsetlock(cd->upd);
return;
}
/*
* ref count 0 - mark as INPROGRESS, return to server, then
* delete from client
*/
cd->flags |= CL_INPROGRESS;
usunsetlock(cd->upd);
tkc_release(cd->cs, EXIST_READ);
/*
* Must pass TK_WAIT in since otherwise we can get the
* following race:
* thread gets WRITE token and then releases it and it's ref on
* the object (the WRITE token gets CACHED).
* a revoke comes in from the server, the WRITE token goes into
* the RETURNING state.
* another client thread drops its ref count, the ref count goes
* to 0 and it calls tkc_returning. If it doesn't wait
* for the WRITE token to finish returning, this client
* could come back from tkc_returning, send an RPC to the
* server. get the response and call tkc_destroy - all before
* the thread that is in client_revoke() gets back and runs
* and calls tkc_return ...
*/
tkc_returning(cd->cs, TK_ADD_SET(EXIST_READ, DATA_SET), &toreturn,
&dofret, TK_WAIT);
assert(TK_SUB_SET(toreturn, TK_ADD_SET(EXIST_READ, DATA_SET)) == TK_NULLSET);
/*
* send back tokens and data
* Turns out that the revoke callout is just the thing we need
* It also calls tkc_return();
* Note that the token server will not return to us until any
* outstanding revokes it has (that would come into client_recall())
* have completed. Thus once we return there can be no more outstanding
* messages for this object.
*/
client_return(NULL, cd, toreturn, TK_NULLSET, dofret);
/* delete from client list */
ussetlock(cp->upd);
for (pcd = &cp->data; *pcd; pcd = &(*pcd)->next) {
if (*pcd == cd) {
*pcd = cd->next;
break;
}
}
sv_broadcast(&cp->wait);
usunsetlock(cp->upd);
usfreelock(cd->upd, usptr);
tkc_destroy(cd->cs);
free(cd);
}
/*
* Can get the lock for either read or update
*/
static int
getlock(struct cdata *cd, int type)
{
int rv;
if (type == TK_READ)
rv = tkc_acquire1(cd->cs, SDATA_READ);
else
rv = tkc_acquire1(cd->cs, SDATA_UPDATE);
if (rv == 0)
ussetlock(cd->upd);
return rv;
}
static void
putlock(struct cdata *cd, int type)
{
usunsetlock(cd->upd);
if (type == TK_READ)
tkc_release1(cd->cs, SDATA_READ);
else
tkc_release1(cd->cs, SDATA_UPDATE);
}
/*===================================================================*/
/*
* client message ops - these correspond 1-1 with server routines
* that are called from _wait
*/
static int
invoke_lookup(tks_ch_t h, unsigned long id, tk_set_t *granted, void **handle)
{
mesg_t *m;
union test3_req *req;
union test3_res *res;
int error;
m = getmesg();
m->op = TEST3_LOOKUP;
m->service_id = SERVICE1_ID;
m->handle = h;
req = (union test3_req *)&m->request;
req->lookup_req.id = id;
error = callmbox(ntombox(S1S()), m);
if (error == 0) {
res = (union test3_res *)&m->response;
*granted = res->lookup_res.existance;
*handle = res->lookup_res.handle;
}
freemesg(m);
return error;
}
static int
invoke_getvalue(tks_ch_t h,
tk_set_t obtain,
tk_set_t toreturn,
tk_disp_t why,
tk_set_t *granted,
tk_set_t *already,
void *handle,
unsigned long *value,
unsigned long *id)
{
mesg_t *m;
union test3_req *req;
union test3_res *res;
int error;
m = getmesg();
m->op = TEST3_GETVALUE;
m->service_id = SERVICE1_ID;
m->handle = h;
req = (union test3_req *)&m->request;
req->getvalue_req.obtain = obtain;
req->getvalue_req.toreturn = toreturn;
req->getvalue_req.why = why;
req->getvalue_req.handle = handle;
req->getvalue_req.value = *value;
error = callmbox(ntombox(S1S()), m);
if (error == 0) {
res = (union test3_res *)&m->response;
*granted = res->getvalue_res.granted;
*already = res->getvalue_res.already;
*value = res->getvalue_res.value;
*id = res->getvalue_res.id;
}
freemesg(m);
return error;
}
/*=================================================================*/
/*
* server routines - invoked from _wait and from local clients
*/
static int
server_lookup(tks_ch_t h, unsigned long id, tk_set_t *granted, void **handle)
{
auto tk_set_t already;
struct server *sp;
struct sdata *sd;
assert(id < maxobjects);
sp = (struct server *)S1_SP();
if (sp == NULL) {
/* race .. return equiv of ENOENT */
*granted = TK_NULLSET;
return 0;
}
uspsema(sp->upd);
for (sd = sp->data; sd; sd = sd->next) {
if (sd->id == id)
break;
}
if (sd) {
/* got one already */
tks_obtain(sd->ss, h, EXIST_READ, granted, NULL, &already);
} else {
/* create a new one */
sd = malloc(sizeof(*sd));
sd->id = id;
sd->value = 1;
sd->next = sp->data;
sp->data = sd;
tks_create("server", sd->ss, (void *)sd, &svriface, NTOKENS,
(void *)id);
/* all set - respond with existance token */
tks_obtain(sd->ss, h, EXIST_READ, granted, NULL, &already);
}
*handle = sd;
usvsema(sp->upd);
tks_notify_idle(sd->ss, EXIST_READ);
assert(already == TK_NULLSET);
assert(*granted == EXIST_READ);
return 0;
}
/*
* respond to GETVALUE request from client
*/
static int
server_getvalue(tks_ch_t h,
tk_set_t obtain,
tk_set_t toreturn,
tk_disp_t why,
tk_set_t *granted,
tk_set_t *already,
void *handle,
unsigned long *value,
unsigned long *id)
{
struct sdata *sd = (struct sdata *)handle;
if (toreturn != TK_NULLSET) {
if (TK_IS_IN_SET(toreturn, DATA_UPDATE)) {
/* shouldn't need a lock since we have WRITE token */
sd->value = *value;
}
assert(!TK_IS_IN_SET(toreturn, EXIST_READ));
tks_return(sd->ss, h, toreturn, TK_NULLSET, TK_NULLSET,
why);
}
tks_obtain(sd->ss, h, obtain, granted, NULL, already);
/* always give back value */
*value = sd->value;
*id = sd->id;
return 0;
}
/*
* server function to handle a return from client - either via a recall
* or a revoke or a client-initiated return.
*/
static void
server_return(tks_ch_t h,
unsigned long value,
void *handle,
tk_set_t ok,
tk_set_t eh,
tk_disp_t which)
{
struct sdata *sd = (struct sdata *)handle;
if (TK_IS_IN_SET(ok, DATA_UPDATE)) {
/* shouldn't need lock - no one else can touch */
sd->value = value;
}
/* this could trigger the tks_idle callback */
tks_return(sd->ss, h, ok, TK_NULLSET, eh, which);
return;
}
/*
* callout function - called when a token has been recalled and can now be
* reclaimed/destroyed
*/
/* ARGSUSED */
static void
server_recalled(void *o, tk_set_t r, tk_set_t refused)
{
assert(0);
}
static void
server_idle(void *o, tk_set_t idle)
{
struct sdata *sd = (struct sdata *)o;
struct sdata **psd;
struct server *sp;
auto tk_set_t st;
int found;
sp = (struct server *)S1_SP();
assert(idle == EXIST_READ);
uspsema(sp->upd);
tks_getstate(sd->ss, &st);
if (TK_GET_LEVEL(st, EXIST_CLASS)) {
/* someone grabbed it again */
usvsema(sp->upd);
return;
}
assert(st == TK_NULLSET);
/* remove from list on server */
for (psd = &sp->data, found = 0; *psd; psd = &(*psd)->next) {
if (*psd == sd) {
*psd = sd->next;
found = 1;
break;
}
}
assert(found);
usvsema(sp->upd);
tks_destroy(sd->ss);
free(sd);
}
/*
* token callout for obtaining tokens
*/
/* ARGSUSED */
static void
client_obtain(void *o,
tk_set_t obtain,
tk_set_t toreturn,
tk_disp_t why,
tk_set_t *refused)
{
struct cdata *cd = (struct cdata *)o;
tk_set_t granted, already;
auto unsigned long value, id;
int error;
if (AM_S1S(cd->h)) {
error = server_getvalue(cd->h, obtain, toreturn, why,
&granted, &already, cd->handle, &value, &id);
} else {
error = invoke_getvalue(cd->h, obtain, toreturn, why,
&granted, &already, cd->handle, &value, &id);
}
assert(error == 0);
assert(id == cd->id);
*refused = TK_SUB_SET(obtain, TK_ADD_SET(granted, already));
assert(*refused == TK_NULLSET);
assert(already == TK_NULLSET);
if (TK_IS_IN_SET(granted, DATA_SET))
cd->value = value;
}
/*
* client_return - callout from token client module
*/
/* ARGSUSED */
static void
client_return(
tkc_state_t ci,
void *o,
tk_set_t revoke,
tk_set_t eh,
tk_disp_t which)
{
mesg_t *m;
union test3_req *req;
struct cdata *cd = (struct cdata *)o;
/* REFERENCED */
int error = 0;
if (AM_S1S(cd->h)) {
/* we are server */
server_return(cd->h, cd->value, cd->handle, revoke, eh, which);
} else {
m = getmesg();
m->op = TEST3_RVALUE;
m->service_id = SERVICE1_ID;
m->handle = cd->h;
req = (union test3_req *)&m->request;
req->revoke_req.torevoke = revoke;
req->revoke_req.eh = eh;
req->revoke_req.which = which;
req->revoke_req.value = cd->value;
req->revoke_req.handle = cd->handle;
if (TK_IS_ANY_CLIENT(which)) {
error = callmbox(ntombox(S1S()), m);
freemesg(m);
} else {
error = writembox(ntombox(S1S()), m);
}
}
/* inform client token module that all is done */
tkc_returned(cd->cs, revoke, TK_NULLSET);
}
/*
* client_recall - called for a server invoked REVOKE or RECALL
* Note that because of the token server protocol, it won't respond
* to a client_initiated return until all server initiated revokes are
* completed, we know that the handle presented here is valid.
*/
static void
client_recall(
struct client *cp,
void *handle,
unsigned long id,
tk_set_t toreturn,
tk_disp_t which)
{
struct cdata *cd;
/* assert that handle is still valid */
ussetlock(cp->upd);
for (cd = cp->data; cd; cd = cd->next) {
if (cd->handle == handle)
break;
}
assert(cd);
assert(cd->id == id);
usunsetlock(cp->upd);
assert(!TK_IS_IN_SET(toreturn, EXIST_READ));
tkc_recall(cd->cs, toreturn, which);
#if OLD
ussetlock(cd->upd);
usunsetlock(cp->upd);
/*
* as soon as we release this lock another client thread could
* release this object. We don't want it to go away until
* we're done with it
*/
assert(cd->ref > 0);
cd->ref++;
usunsetlock(cd->upd);
assert(!TK_IS_IN_SET(toreturn, EXIST_READ));
(void) tkc_recall(cd->cs, toreturn, which);
assert(cd->ref > 0);
releaseid(cp, cd);
#endif
}
/*
* get client handle from within token module - used for TRACING only
*/
tks_ch_t
getch_t(void *o)
{
struct cdata *cd = (struct cdata *)o;
assert(cd);
return(cd->h);
}
/*
* callout function that is called from the token server module when a
* token (set) needs to be revoked. It sends a message to the appropriate
* client. This sends a 1 way message
*/
/* ARGSUSED */
static void
server_recall(void *o, tks_ch_t h, tk_set_t r, tk_disp_t which)
{
mesg_t *m;
union test3_req *req;
struct sdata *sd = (struct sdata *)o;
if (AM_S1S(htonode(h))) {
struct client *cp;
cp = (struct client *)S1_CP(htonode(h));
client_recall(cp, sd, sd->id, r, which);
return;
}
m = getmesg();
m->op = TEST3_REVOKE;
m->service_id = SERVICE1_ID;
m->handle = h;
req = (union test3_req *)&m->request;
req->revoke_req.torevoke = r;
req->revoke_req.which = which;
req->revoke_req.handle = sd;
req->revoke_req.id = sd->id;
(void)writembox(ntombox(htonode(h)), m);
}
/*
* this thread handles any out-of-the-blue messages
* Both client and server.
* 'node' is which node we are.
*/
static void
service1_wait(mesg_t *m, int node)
{
struct client *cp; /* client side data */
union test3_req *req;
union test3_res *res;
req = (union test3_req *)&m->request;
res = (union test3_res *)&m->response;
switch (m->op) {
/* client messages */
case TEST3_REVOKE:
assert(!AM_S1S(node));
cp = (struct client *)S1_CP(node);
client_recall(cp, req->revoke_req.handle,
req->revoke_req.id,
req->revoke_req.torevoke,
req->revoke_req.which);
break;
/*
* server messages
*/
case TEST3_RVALUE:
{
assert(AM_S1S(node));
server_return(m->handle,
req->revoke_req.value,
req->revoke_req.handle,
req->revoke_req.torevoke,
req->revoke_req.eh,
req->revoke_req.which);
break;
}
case TEST3_GETVALUE:
{
auto unsigned long value;
assert(AM_S1S(node));
value = req->getvalue_req.value;
server_getvalue(m->handle,
req->getvalue_req.obtain,
req->getvalue_req.toreturn,
req->getvalue_req.why,
&res->getvalue_res.granted,
&res->getvalue_res.already,
req->getvalue_req.handle,
&value,
&res->getvalue_res.id);
res->getvalue_res.value = value;
break;
}
case TEST3_LOOKUP:
assert(AM_S1S(node));
server_lookup(m->handle, req->lookup_req.id,
&res->lookup_res.existance,
&res->lookup_res.handle);
break;
default:
fprintf(stderr, "service1-wait:Unknown msg op:%d\n", m->op);
abort();
break;
}
return;
}
View Git Blame Copy Permalink