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