/* * This file is part of the UCB release of Plan 9. It is subject to the license * terms in the LICENSE file found in the top-level directory of this * distribution and at http://akaros.cs.berkeley.edu/files/Plan9License. No * part of the UCB release of Plan 9, including this file, may be copied, * modified, propagated, or distributed except according to the terms contained * in the LICENSE file. */ #include "stdinc.h"#include "dat.h"#include "fns.h" typedef struct LumpQueue LumpQueue;typedef struct WLump WLump; enum{ MaxLumpQ = 1 << 3 /* max. lumps on a single write queue, must be pow 2 */}; struct WLump{ Lump *u; Packet *p; int creator; int gen; uint ms;}; struct LumpQueue{ QLock lock; Rendez flush; Rendez full; Rendez empty; WLump q[MaxLumpQ]; int w; int r;}; static LumpQueue *lumpqs;static int nqs; static QLock glk;static int gen; static void queueproc(void *vq); intinitlumpqueues(int nq){ LumpQueue *q; int i; nqs = nq; lumpqs = MKNZ(LumpQueue, nq); for(i = 0; i < nq; i++){ q = &lumpqs[i]; q->full.l = &q->lock; q->empty.l = &q->lock; q->flush.l = &q->lock; if(vtproc(queueproc, q) < 0){ seterr(EOk, "can't start write queue slave: %r"); return -1; } } return 0;} /* * queue a lump & it's packet data for writing */intqueuewrite(Lump *u, Packet *p, int creator, uint ms){ LumpQueue *q; int i; trace(TraceProc, "queuewrite"); i = indexsect(mainindex, u->score); if(i < 0 || i >= nqs){ seterr(EBug, "internal error: illegal index section in queuewrite"); return -1; } q = &lumpqs[i]; qlock(&q->lock); while(q->r == ((q->w + 1) & (MaxLumpQ - 1))){ trace(TraceProc, "queuewrite sleep"); rsleep(&q->full); } q->q[q->w].u = u; q->q[q->w].p = p; q->q[q->w].creator = creator; q->q[q->w].ms = ms; q->q[q->w].gen = gen; q->w = (q->w + 1) & (MaxLumpQ - 1); trace(TraceProc, "queuewrite wakeup"); rwakeup(&q->empty); qunlock(&q->lock); return 0;} voidflushqueue(void){ int i; LumpQueue *q; if(!lumpqs) return; trace(TraceProc, "flushqueue"); qlock(&glk); gen++; qunlock(&glk); for(i=0; i<mainindex->nsects; i++){ q = &lumpqs[i]; qlock(&q->lock); while(q->w != q->r && gen - q->q[q->r].gen > 0){ trace(TraceProc, "flushqueue sleep q%d", i); rsleep(&q->flush); } qunlock(&q->lock); }} static voidqueueproc(void *vq){ LumpQueue *q; Lump *u; Packet *p; int creator; uint ms; threadsetname("queueproc"); q = vq; for(;;){ qlock(&q->lock); while(q->w == q->r){ trace(TraceProc, "queueproc sleep empty"); rsleep(&q->empty); } u = q->q[q->r].u; p = q->q[q->r].p; creator = q->q[q->r].creator; ms = q->q[q->r].ms; q->r = (q->r + 1) & (MaxLumpQ - 1); trace(TraceProc, "queueproc wakeup flush"); rwakeupall(&q->flush); trace(TraceProc, "queueproc wakeup full"); rwakeup(&q->full); qunlock(&q->lock); trace(TraceProc, "queueproc writelump %V", u->score); if(writeqlump(u, p, creator, ms) < 0) fprint(2, "failed to write lump for %V: %r", u->score); trace(TraceProc, "queueproc wrotelump %V", u->score); putlump(u); }}
/*
* This file is part of the UCB release of Plan 9. It is subject to the license
* terms in the LICENSE file found in the top-level directory of this
* distribution and at http://akaros.cs.berkeley.edu/files/Plan9License. No
* part of the UCB release of Plan 9, including this file, may be copied,
* modified, propagated, or distributed except according to the terms contained
* in the LICENSE file.
*/
#include "stdinc.h"
#include "dat.h"
#include "fns.h"
typedef struct LumpQueue LumpQueue;
typedef struct WLump WLump;
enum
{
MaxLumpQ = 1 << 3 /* max. lumps on a single write queue, must be pow 2 */
};
struct WLump
{
Lump *u;
Packet *p;
int creator;
int gen;
uint ms;
};
struct LumpQueue
{
QLock lock;
Rendez flush;
Rendez full;
Rendez empty;
WLump q[MaxLumpQ];
int w;
int r;
};
static LumpQueue *lumpqs;
static int nqs;
static QLock glk;
static int gen;
static void queueproc(void *vq);
int
initlumpqueues(int nq)
{
LumpQueue *q;
int i;
nqs = nq;
lumpqs = MKNZ(LumpQueue, nq);
for(i = 0; i < nq; i++){
q = &lumpqs[i];
q->full.l = &q->lock;
q->empty.l = &q->lock;
q->flush.l = &q->lock;
if(vtproc(queueproc, q) < 0){
seterr(EOk, "can't start write queue slave: %r");
return -1;
}
}
return 0;
}
/*
* queue a lump & it's packet data for writing
*/
int
queuewrite(Lump *u, Packet *p, int creator, uint ms)
{
LumpQueue *q;
int i;
trace(TraceProc, "queuewrite");
i = indexsect(mainindex, u->score);
if(i < 0 || i >= nqs){
seterr(EBug, "internal error: illegal index section in queuewrite");
return -1;
}
q = &lumpqs[i];
qlock(&q->lock);
while(q->r == ((q->w + 1) & (MaxLumpQ - 1))){
trace(TraceProc, "queuewrite sleep");
rsleep(&q->full);
}
q->q[q->w].u = u;
q->q[q->w].p = p;
q->q[q->w].creator = creator;
q->q[q->w].ms = ms;
q->q[q->w].gen = gen;
q->w = (q->w + 1) & (MaxLumpQ - 1);
trace(TraceProc, "queuewrite wakeup");
rwakeup(&q->empty);
qunlock(&q->lock);
return 0;
}
void
flushqueue(void)
{
int i;
LumpQueue *q;
if(!lumpqs)
return;
trace(TraceProc, "flushqueue");
qlock(&glk);
gen++;
qunlock(&glk);
for(i=0; i<mainindex->nsects; i++){
q = &lumpqs[i];
qlock(&q->lock);
while(q->w != q->r && gen - q->q[q->r].gen > 0){
trace(TraceProc, "flushqueue sleep q%d", i);
rsleep(&q->flush);
}
qunlock(&q->lock);
}
}
static void
queueproc(void *vq)
{
LumpQueue *q;
Lump *u;
Packet *p;
int creator;
uint ms;
threadsetname("queueproc");
q = vq;
for(;;){
qlock(&q->lock);
while(q->w == q->r){
trace(TraceProc, "queueproc sleep empty");
rsleep(&q->empty);
}
u = q->q[q->r].u;
p = q->q[q->r].p;
creator = q->q[q->r].creator;
ms = q->q[q->r].ms;
q->r = (q->r + 1) & (MaxLumpQ - 1);
trace(TraceProc, "queueproc wakeup flush");
rwakeupall(&q->flush);
trace(TraceProc, "queueproc wakeup full");
rwakeup(&q->full);
qunlock(&q->lock);
trace(TraceProc, "queueproc writelump %V", u->score);
if(writeqlump(u, p, creator, ms) < 0)
fprint(2, "failed to write lump for %V: %r", u->score);
trace(TraceProc, "queueproc wrotelump %V", u->score);
putlump(u);
}
}