1 a0d146ed 2005-07-12 devnull #include "stdinc.h"
  2 a0d146ed 2005-07-12 devnull #include "dat.h"
  3 a0d146ed 2005-07-12 devnull #include "fns.h"
  4 a0d146ed 2005-07-12 devnull 
  5 a0d146ed 2005-07-12 devnull typedef struct LumpQueue	LumpQueue;
  6 a0d146ed 2005-07-12 devnull typedef struct WLump		WLump;
  7 a0d146ed 2005-07-12 devnull 
  8 a0d146ed 2005-07-12 devnull enum
  9 a0d146ed 2005-07-12 devnull {
 10 a0d146ed 2005-07-12 devnull 	MaxLumpQ	= 1 << 3	/* max. lumps on a single write queue, must be pow 2 */
 11 a0d146ed 2005-07-12 devnull };
 12 a0d146ed 2005-07-12 devnull 
 13 a0d146ed 2005-07-12 devnull struct WLump
 14 a0d146ed 2005-07-12 devnull {
 15 a0d146ed 2005-07-12 devnull 	Lump	*u;
 16 a0d146ed 2005-07-12 devnull 	Packet	*p;
 17 a0d146ed 2005-07-12 devnull 	int	creator;
 18 a0d146ed 2005-07-12 devnull 	int	gen;
 19 a0d146ed 2005-07-12 devnull 	uint	ms;
 20 a0d146ed 2005-07-12 devnull };
 21 a0d146ed 2005-07-12 devnull 
 22 a0d146ed 2005-07-12 devnull struct LumpQueue
 23 a0d146ed 2005-07-12 devnull {
 24 a0d146ed 2005-07-12 devnull 	QLock	lock;
 25 a0d146ed 2005-07-12 devnull 	Rendez 	flush;
 26 a0d146ed 2005-07-12 devnull 	Rendez	full;
 27 a0d146ed 2005-07-12 devnull 	Rendez	empty;
 28 a0d146ed 2005-07-12 devnull 	WLump	q[MaxLumpQ];
 29 a0d146ed 2005-07-12 devnull 	int	w;
 30 a0d146ed 2005-07-12 devnull 	int	r;
 31 a0d146ed 2005-07-12 devnull };
 32 a0d146ed 2005-07-12 devnull 
 33 a0d146ed 2005-07-12 devnull static LumpQueue	*lumpqs;
 34 a0d146ed 2005-07-12 devnull static int		nqs;
 35 a0d146ed 2005-07-12 devnull 
 36 a0d146ed 2005-07-12 devnull static QLock		glk;
 37 a0d146ed 2005-07-12 devnull static int		gen;
 38 a0d146ed 2005-07-12 devnull 
 39 a0d146ed 2005-07-12 devnull static void	queueproc(void *vq);
 40 a0d146ed 2005-07-12 devnull 
 41 a0d146ed 2005-07-12 devnull int
 42 a0d146ed 2005-07-12 devnull initlumpqueues(int nq)
 43 a0d146ed 2005-07-12 devnull {
 44 a0d146ed 2005-07-12 devnull 	LumpQueue *q;
 45 a0d146ed 2005-07-12 devnull 
 46 a0d146ed 2005-07-12 devnull 	int i;
 47 a0d146ed 2005-07-12 devnull 	nqs = nq;
 48 a0d146ed 2005-07-12 devnull 
 49 a0d146ed 2005-07-12 devnull 	lumpqs = MKNZ(LumpQueue, nq);
 50 a0d146ed 2005-07-12 devnull 
 51 a0d146ed 2005-07-12 devnull 	for(i = 0; i < nq; i++){
 52 a0d146ed 2005-07-12 devnull 		q = &lumpqs[i];
 53 a0d146ed 2005-07-12 devnull 		q->full.l = &q->lock;
 54 a0d146ed 2005-07-12 devnull 		q->empty.l = &q->lock;
 55 a0d146ed 2005-07-12 devnull 		q->flush.l = &q->lock;
 56 a0d146ed 2005-07-12 devnull 
 57 a0d146ed 2005-07-12 devnull 		if(vtproc(queueproc, q) < 0){
 58 a0d146ed 2005-07-12 devnull 			seterr(EOk, "can't start write queue slave: %r");
 59 a0d146ed 2005-07-12 devnull 			return -1;
 60 a0d146ed 2005-07-12 devnull 		}
 61 a0d146ed 2005-07-12 devnull 		if(vtproc(queueproc, q) < 0){
 62 a0d146ed 2005-07-12 devnull 			seterr(EOk, "can't start write queue slave: %r");
 63 a0d146ed 2005-07-12 devnull 			return -1;
 64 a0d146ed 2005-07-12 devnull 		}
 65 a0d146ed 2005-07-12 devnull 		if(vtproc(queueproc, q) < 0){
 66 a0d146ed 2005-07-12 devnull 			seterr(EOk, "can't start write queue slave: %r");
 67 a0d146ed 2005-07-12 devnull 			return -1;
 68 a0d146ed 2005-07-12 devnull 		}
 69 a0d146ed 2005-07-12 devnull 		if(vtproc(queueproc, q) < 0){
 70 a0d146ed 2005-07-12 devnull 			seterr(EOk, "can't start write queue slave: %r");
 71 a0d146ed 2005-07-12 devnull 			return -1;
 72 a0d146ed 2005-07-12 devnull 		}
 73 a0d146ed 2005-07-12 devnull 		if(vtproc(queueproc, q) < 0){
 74 a0d146ed 2005-07-12 devnull 			seterr(EOk, "can't start write queue slave: %r");
 75 a0d146ed 2005-07-12 devnull 			return -1;
 76 a0d146ed 2005-07-12 devnull 		}
 77 a0d146ed 2005-07-12 devnull 	}
 78 a0d146ed 2005-07-12 devnull 
 79 a0d146ed 2005-07-12 devnull 	return 0;
 80 a0d146ed 2005-07-12 devnull }
 81 a0d146ed 2005-07-12 devnull 
 82 a0d146ed 2005-07-12 devnull /*
 83 a0d146ed 2005-07-12 devnull  * queue a lump & it's packet data for writing
 84 a0d146ed 2005-07-12 devnull  */
 85 a0d146ed 2005-07-12 devnull int
 86 a0d146ed 2005-07-12 devnull queuewrite(Lump *u, Packet *p, int creator, uint ms)
 87 a0d146ed 2005-07-12 devnull {
 88 a0d146ed 2005-07-12 devnull 	LumpQueue *q;
 89 a0d146ed 2005-07-12 devnull 	int i;
 90 a0d146ed 2005-07-12 devnull 
 91 a0d146ed 2005-07-12 devnull 	trace(TraceProc, "queuewrite");
 92 a0d146ed 2005-07-12 devnull 	i = indexsect(mainindex, u->score);
 93 a0d146ed 2005-07-12 devnull 	if(i < 0 || i >= nqs){
 94 a0d146ed 2005-07-12 devnull 		seterr(EBug, "internal error: illegal index section in queuewrite");
 95 a0d146ed 2005-07-12 devnull 		return -1;
 96 a0d146ed 2005-07-12 devnull 	}
 97 a0d146ed 2005-07-12 devnull 
 98 a0d146ed 2005-07-12 devnull 	q = &lumpqs[i];
 99 a0d146ed 2005-07-12 devnull 
100 a0d146ed 2005-07-12 devnull 	qlock(&q->lock);
101 a0d146ed 2005-07-12 devnull 	while(q->r == ((q->w + 1) & (MaxLumpQ - 1))){
102 a0d146ed 2005-07-12 devnull 		trace(TraceProc, "queuewrite sleep");
103 a0d146ed 2005-07-12 devnull 		rsleep(&q->full);
104 a0d146ed 2005-07-12 devnull 	}
105 a0d146ed 2005-07-12 devnull 
106 a0d146ed 2005-07-12 devnull 	q->q[q->w].u = u;
107 a0d146ed 2005-07-12 devnull 	q->q[q->w].p = p;
108 a0d146ed 2005-07-12 devnull 	q->q[q->w].creator = creator;
109 a0d146ed 2005-07-12 devnull 	q->q[q->w].ms = ms;
110 a0d146ed 2005-07-12 devnull 	q->q[q->w].gen = gen;
111 a0d146ed 2005-07-12 devnull 	q->w = (q->w + 1) & (MaxLumpQ - 1);
112 a0d146ed 2005-07-12 devnull 
113 a0d146ed 2005-07-12 devnull 	trace(TraceProc, "queuewrite wakeup");
114 a0d146ed 2005-07-12 devnull 	rwakeup(&q->empty);
115 a0d146ed 2005-07-12 devnull 
116 a0d146ed 2005-07-12 devnull 	qunlock(&q->lock);
117 a0d146ed 2005-07-12 devnull 
118 a0d146ed 2005-07-12 devnull 	return 0;
119 a0d146ed 2005-07-12 devnull }
120 a0d146ed 2005-07-12 devnull 
121 a0d146ed 2005-07-12 devnull void
122 a0d146ed 2005-07-12 devnull flushqueue(void)
123 a0d146ed 2005-07-12 devnull {
124 a0d146ed 2005-07-12 devnull 	int i;
125 a0d146ed 2005-07-12 devnull 	LumpQueue *q;
126 a0d146ed 2005-07-12 devnull 
127 a0d146ed 2005-07-12 devnull 	if(!lumpqs)
128 a0d146ed 2005-07-12 devnull 		return;
129 a0d146ed 2005-07-12 devnull 
130 a0d146ed 2005-07-12 devnull 	trace(TraceProc, "flushqueue");
131 a0d146ed 2005-07-12 devnull 
132 a0d146ed 2005-07-12 devnull 	qlock(&glk);
133 a0d146ed 2005-07-12 devnull 	gen++;
134 a0d146ed 2005-07-12 devnull 	qunlock(&glk);
135 a0d146ed 2005-07-12 devnull 
136 a0d146ed 2005-07-12 devnull 	for(i=0; i<mainindex->nsects; i++){
137 a0d146ed 2005-07-12 devnull 		q = &lumpqs[i];
138 a0d146ed 2005-07-12 devnull 		qlock(&q->lock);
139 a0d146ed 2005-07-12 devnull 		while(q->w != q->r && gen - q->q[q->r].gen > 0){
140 a0d146ed 2005-07-12 devnull 			trace(TraceProc, "flushqueue sleep q%d", i);
141 a0d146ed 2005-07-12 devnull 			rsleep(&q->flush);
142 a0d146ed 2005-07-12 devnull 		}
143 a0d146ed 2005-07-12 devnull 		qunlock(&q->lock);
144 a0d146ed 2005-07-12 devnull 	}
145 a0d146ed 2005-07-12 devnull }
146 a0d146ed 2005-07-12 devnull 	
147 a0d146ed 2005-07-12 devnull static void
148 a0d146ed 2005-07-12 devnull queueproc(void *vq)
149 a0d146ed 2005-07-12 devnull {
150 a0d146ed 2005-07-12 devnull 	LumpQueue *q;
151 a0d146ed 2005-07-12 devnull 	Lump *u;
152 a0d146ed 2005-07-12 devnull 	Packet *p;
153 a0d146ed 2005-07-12 devnull 	int creator;
154 a0d146ed 2005-07-12 devnull 	uint ms;
155 a0d146ed 2005-07-12 devnull 
156 a0d146ed 2005-07-12 devnull 	threadsetname("queueproc");
157 a0d146ed 2005-07-12 devnull 
158 a0d146ed 2005-07-12 devnull 	q = vq;
159 a0d146ed 2005-07-12 devnull 	for(;;){
160 a0d146ed 2005-07-12 devnull 		qlock(&q->lock);
161 a0d146ed 2005-07-12 devnull 		while(q->w == q->r){
162 a0d146ed 2005-07-12 devnull 			trace(TraceProc, "queueproc sleep empty");
163 a0d146ed 2005-07-12 devnull 			rsleep(&q->empty);
164 a0d146ed 2005-07-12 devnull 		}
165 a0d146ed 2005-07-12 devnull 
166 a0d146ed 2005-07-12 devnull 		u = q->q[q->r].u;
167 a0d146ed 2005-07-12 devnull 		p = q->q[q->r].p;
168 a0d146ed 2005-07-12 devnull 		creator = q->q[q->r].creator;
169 a0d146ed 2005-07-12 devnull 		ms = q->q[q->r].ms;
170 a0d146ed 2005-07-12 devnull 
171 a0d146ed 2005-07-12 devnull 		q->r = (q->r + 1) & (MaxLumpQ - 1);
172 a0d146ed 2005-07-12 devnull 		trace(TraceProc, "queueproc wakeup flush");
173 a0d146ed 2005-07-12 devnull 		rwakeupall(&q->flush);
174 a0d146ed 2005-07-12 devnull 
175 a0d146ed 2005-07-12 devnull 		trace(TraceProc, "queueproc wakeup full");
176 a0d146ed 2005-07-12 devnull 		rwakeup(&q->full);
177 a0d146ed 2005-07-12 devnull 
178 a0d146ed 2005-07-12 devnull 		qunlock(&q->lock);
179 a0d146ed 2005-07-12 devnull 
180 a0d146ed 2005-07-12 devnull 		trace(TraceProc, "queueproc writelump %V", u->score);
181 a0d146ed 2005-07-12 devnull 		if(writeqlump(u, p, creator, ms) < 0)
182 a0d146ed 2005-07-12 devnull 			fprint(2, "failed to write lump for %V: %r", u->score);
183 a0d146ed 2005-07-12 devnull 		trace(TraceProc, "queueproc wrotelump %V", u->score);
184 a0d146ed 2005-07-12 devnull 
185 a0d146ed 2005-07-12 devnull 		putlump(u);
186 a0d146ed 2005-07-12 devnull 	}
187 a0d146ed 2005-07-12 devnull }