| rev |
line source |
|
Me@0
|
1 /*
|
|
Me@1
|
2 * Copyright 2009 OpenSourceStewardshipFoundation.org
|
|
Me@0
|
3 * Licensed under GNU General Public License version 2
|
|
Me@0
|
4 *
|
|
Me@0
|
5 * Author: seanhalle@yahoo.com
|
|
Me@0
|
6 */
|
|
Me@0
|
7
|
|
Me@0
|
8
|
|
Me@0
|
9 #include <stdio.h>
|
|
Me@0
|
10 #include <errno.h>
|
|
Me@0
|
11 #include <pthread.h>
|
|
Me@0
|
12 #include <stdlib.h>
|
|
Me@1
|
13 #include <sched.h>
|
|
Me@1
|
14 #include <windows.h>
|
|
Me@0
|
15
|
|
Me@0
|
16 #include "BlockingQueue.h"
|
|
Me@0
|
17
|
|
Me@0
|
18 #define INC(x) (++x == 1024) ? (x) = 0 : (x)
|
|
Me@0
|
19
|
|
Me@1
|
20 #define SPINLOCK_TRIES 100000
|
|
Me@0
|
21
|
|
Me@0
|
22 //===========================================================================
|
|
Me@0
|
23 //Normal pthread Q
|
|
Me@0
|
24
|
|
Me@4
|
25 PThdQueueStruc* makePThdQ()
|
|
Me@0
|
26 {
|
|
Me@4
|
27 PThdQueueStruc* retQ;
|
|
Me@0
|
28 int status;
|
|
Me@4
|
29 retQ = (PThdQueueStruc *) malloc( sizeof( PThdQueueStruc ) );
|
|
Me@0
|
30
|
|
Me@0
|
31
|
|
Me@0
|
32 status = pthread_mutex_init( &retQ->mutex_t, NULL);
|
|
Me@0
|
33 if (status < 0)
|
|
Me@0
|
34 {
|
|
Me@0
|
35 perror("Error in creating mutex:");
|
|
Me@0
|
36 exit(1);
|
|
Me@0
|
37 return NULL;
|
|
Me@0
|
38 }
|
|
Me@0
|
39
|
|
Me@0
|
40 status = pthread_cond_init ( &retQ->cond_w_t, NULL);
|
|
Me@0
|
41 if (status < 0)
|
|
Me@0
|
42 {
|
|
Me@0
|
43 perror("Error in creating cond_var:");
|
|
Me@0
|
44 exit(1);
|
|
Me@0
|
45 return NULL;
|
|
Me@0
|
46 }
|
|
Me@0
|
47
|
|
Me@0
|
48 status = pthread_cond_init ( &retQ->cond_r_t, NULL);
|
|
Me@0
|
49 if (status < 0)
|
|
Me@0
|
50 {
|
|
Me@0
|
51 perror("Error in creating cond_var:");
|
|
Me@0
|
52 exit(1);
|
|
Me@0
|
53 return NULL;
|
|
Me@0
|
54 }
|
|
Me@0
|
55
|
|
Me@0
|
56 retQ->count = 0;
|
|
Me@0
|
57 retQ->readPos = 0;
|
|
Me@0
|
58 retQ->writePos = 0;
|
|
Me@0
|
59 retQ -> w_empty = retQ -> w_full = 0;
|
|
Me@0
|
60
|
|
Me@0
|
61 return retQ;
|
|
Me@0
|
62 }
|
|
Me@0
|
63
|
|
Me@4
|
64 void * readPThdQ( PThdQueueStruc *Q )
|
|
Me@0
|
65 { void *ret;
|
|
Me@0
|
66 int status, wt;
|
|
Me@0
|
67 pthread_mutex_lock( &Q->mutex_t );
|
|
Me@0
|
68 {
|
|
Me@0
|
69 while( Q -> count == 0 )
|
|
Me@0
|
70 { Q -> w_empty = 1;
|
|
Me@1
|
71 // pthread_cond_broadcast( &Q->cond_w_t );
|
|
Me@0
|
72 status = pthread_cond_wait( &Q->cond_r_t, &Q->mutex_t );
|
|
Me@0
|
73 if (status != 0)
|
|
Me@0
|
74 { perror("Thread wait error: ");
|
|
Me@0
|
75 exit(1);
|
|
Me@0
|
76 }
|
|
Me@0
|
77 }
|
|
Me@0
|
78 Q -> w_empty = 0;
|
|
Me@0
|
79 Q -> count -= 1;
|
|
Me@0
|
80 ret = Q->data[ Q->readPos ];
|
|
Me@0
|
81 INC( Q->readPos );
|
|
Me@0
|
82 wt = Q -> w_full;
|
|
Me@0
|
83 Q -> w_full = 0;
|
|
Me@1
|
84 //pthread_cond_broadcast( &Q->cond_w_t );
|
|
Me@0
|
85 }
|
|
Me@0
|
86 pthread_mutex_unlock( &Q->mutex_t );
|
|
Me@0
|
87 if (wt) pthread_cond_signal( &Q->cond_w_t );
|
|
Me@0
|
88
|
|
Me@0
|
89 return( ret );
|
|
Me@0
|
90 }
|
|
Me@0
|
91
|
|
Me@4
|
92 void writePThdQ( void * in, PThdQueueStruc* Q )
|
|
Me@0
|
93 {
|
|
Me@0
|
94 int status, wt;
|
|
Me@0
|
95 pthread_mutex_lock( &Q->mutex_t );
|
|
Me@0
|
96 {
|
|
Me@0
|
97 while( Q->count >= 1024 )
|
|
Me@0
|
98 {
|
|
Me@0
|
99 Q -> w_full = 1;
|
|
Me@0
|
100 // pthread_cond_broadcast( &Q->cond_r_t );
|
|
Me@0
|
101 status = pthread_cond_wait( &Q->cond_w_t, &Q->mutex_t );
|
|
Me@0
|
102 if (status != 0)
|
|
Me@0
|
103 { perror("Thread wait error: ");
|
|
Me@0
|
104 exit(1);
|
|
Me@0
|
105 }
|
|
Me@0
|
106 }
|
|
Me@0
|
107 Q -> w_full = 0;
|
|
Me@0
|
108 Q->count += 1;
|
|
Me@0
|
109 Q->data[ Q->writePos ] = in;
|
|
Me@0
|
110 INC( Q->writePos );
|
|
Me@0
|
111 wt = Q -> w_empty;
|
|
Me@0
|
112 Q -> w_empty = 0;
|
|
Me@0
|
113 // pthread_cond_broadcast( &Q->cond_r_t );
|
|
Me@0
|
114 }
|
|
Me@0
|
115 pthread_mutex_unlock( &Q->mutex_t );
|
|
Me@0
|
116 if( wt ) pthread_cond_signal( &Q->cond_r_t );
|
|
Me@0
|
117 }
|
|
Me@0
|
118
|
|
Me@0
|
119
|
|
Me@0
|
120 //===========================================================================
|
|
Me@0
|
121 // multi reader multi writer fast Q via CAS
|
|
Me@0
|
122 #ifndef _GNU_SOURCE
|
|
Me@0
|
123 #define _GNU_SOURCE
|
|
Me@0
|
124
|
|
Me@0
|
125 /*This is a blocking queue, but it uses CAS instr plus yield() when empty
|
|
Me@0
|
126 * or full
|
|
Me@0
|
127 *It uses CAS because it's meant to have more than one reader and more than
|
|
Me@0
|
128 * one writer.
|
|
Me@0
|
129 */
|
|
Me@0
|
130
|
|
Me@0
|
131 CASQueueStruc* makeCASQ()
|
|
Me@0
|
132 {
|
|
Me@0
|
133 CASQueueStruc* retQ;
|
|
Me@0
|
134 retQ = (CASQueueStruc *) malloc( sizeof( CASQueueStruc ) );
|
|
Me@0
|
135
|
|
Me@0
|
136 retQ->insertLock = UNLOCKED;
|
|
Me@0
|
137 retQ->extractLock= UNLOCKED;
|
|
Me@0
|
138 //TODO: check got pointer syntax right
|
|
Me@0
|
139 retQ->extractPos = &(retQ->startOfData[0]); //side by side == empty
|
|
Me@0
|
140 retQ->insertPos = &(retQ->startOfData[1]); // so start pos's have to be
|
|
Me@0
|
141 retQ->endOfData = &(retQ->startOfData[1023]);
|
|
Me@0
|
142
|
|
Me@0
|
143 return retQ;
|
|
Me@0
|
144 }
|
|
Me@0
|
145
|
|
Me@0
|
146
|
|
Me@0
|
147 void* readCASQ( CASQueueStruc* Q )
|
|
Me@1
|
148 { void *out = 0;
|
|
Me@1
|
149 int tries = 0;
|
|
Me@1
|
150 void **startOfData = Q->startOfData;
|
|
Me@1
|
151 void **endOfData = Q->endOfData;
|
|
Me@1
|
152
|
|
Me@0
|
153 int success = FALSE;
|
|
Me@0
|
154
|
|
Me@4
|
155 while( TRUE )
|
|
Me@0
|
156 { success =
|
|
Me@0
|
157 __sync_bool_compare_and_swap( &(Q->extractLock), UNLOCKED, LOCKED );
|
|
Me@4
|
158 //NOTE: checked assy, and it does lock correctly..
|
|
Me@0
|
159 if( success )
|
|
Me@0
|
160 {
|
|
Me@1
|
161 void **insertPos = Q->insertPos;
|
|
Me@1
|
162 void **extractPos = Q->extractPos;
|
|
Me@0
|
163
|
|
Me@0
|
164 //if not empty -- extract just below insert when empty
|
|
Me@0
|
165 if( insertPos - extractPos != 1 &&
|
|
Me@0
|
166 !(extractPos == endOfData && insertPos == startOfData))
|
|
Me@0
|
167 { //move before read
|
|
Me@0
|
168 if( extractPos == endOfData ) //write new pos exactly once, correctly
|
|
Me@0
|
169 { Q->extractPos = startOfData; //can't overrun then fix it 'cause
|
|
Me@0
|
170 } // other thread might read bad pos
|
|
Me@0
|
171 else
|
|
Me@0
|
172 { Q->extractPos++;
|
|
Me@0
|
173 }
|
|
Me@0
|
174 out = *(Q->extractPos);
|
|
Me@0
|
175 Q->extractLock = UNLOCKED;
|
|
Me@0
|
176 return out;
|
|
Me@0
|
177 }
|
|
Me@0
|
178 else //Q is empty
|
|
Me@0
|
179 { success = FALSE;
|
|
Me@0
|
180 Q->extractLock = UNLOCKED;//have to try again, release for others
|
|
Me@0
|
181 }
|
|
Me@0
|
182 }
|
|
Me@0
|
183 //Q is busy or empty
|
|
Me@0
|
184 tries++;
|
|
Me@1
|
185 if( tries > SPINLOCK_TRIES ) SwitchToThread(); //WinAPI yield()
|
|
Me@0
|
186 }
|
|
Me@0
|
187 }
|
|
Me@0
|
188
|
|
Me@0
|
189 void writeCASQ( void * in, CASQueueStruc* Q )
|
|
Me@0
|
190 {
|
|
Me@0
|
191 int tries = 0;
|
|
Me@1
|
192 //TODO: need to make Q volatile? Want to do this Q in assembly!
|
|
Me@1
|
193 //Have no idea what GCC's going to do to this code
|
|
Me@1
|
194 void **startOfData = Q->startOfData;
|
|
Me@1
|
195 void **endOfData = Q->endOfData;
|
|
Me@1
|
196
|
|
Me@0
|
197 int success = FALSE;
|
|
Me@0
|
198
|
|
Me@4
|
199 while( TRUE )
|
|
Me@0
|
200 { success =
|
|
Me@0
|
201 __sync_bool_compare_and_swap( &(Q->insertLock), UNLOCKED, LOCKED );
|
|
Me@0
|
202 if( success )
|
|
Me@0
|
203 {
|
|
Me@1
|
204 void **insertPos = Q->insertPos;
|
|
Me@1
|
205 void **extractPos = Q->extractPos;
|
|
Me@0
|
206
|
|
Me@0
|
207 //check if room to insert.. can't use a count variable
|
|
Me@0
|
208 // 'cause both insertor Thd and extractor Thd would write it
|
|
Me@0
|
209 if( extractPos - insertPos != 1 &&
|
|
Me@0
|
210 !(insertPos == endOfData && extractPos == startOfData))
|
|
Me@1
|
211 { *(Q->insertPos) = in; //insert before move
|
|
Me@4
|
212 if( insertPos == endOfData )
|
|
Me@0
|
213 { Q->insertPos = startOfData;
|
|
Me@0
|
214 }
|
|
Me@0
|
215 else
|
|
Me@0
|
216 { Q->insertPos++;
|
|
Me@0
|
217 }
|
|
Me@0
|
218 Q->insertLock = UNLOCKED;
|
|
Me@0
|
219 return;
|
|
Me@0
|
220 }
|
|
Me@0
|
221 else //Q is full
|
|
Me@0
|
222 { success = FALSE;
|
|
Me@0
|
223 Q->insertLock = UNLOCKED;//have to try again, release for others
|
|
Me@0
|
224 }
|
|
Me@0
|
225 }
|
|
Me@0
|
226 tries++;
|
|
Me@1
|
227 if( tries > SPINLOCK_TRIES ) SwitchToThread(); //Win yield()
|
|
Me@0
|
228 }
|
|
Me@0
|
229 }
|
|
Me@0
|
230
|
|
Me@0
|
231 #endif //_GNU_SOURCE
|
|
Me@0
|
232
|
|
Me@1
|
233
|
|
Me@0
|
234 //===========================================================================
|
|
Me@0
|
235 //Single reader single writer super fast Q.. no atomic instrs..
|
|
Me@0
|
236
|
|
Me@0
|
237
|
|
Me@0
|
238 /*This is a blocking queue, but it uses no atomic instructions, just does
|
|
Me@1
|
239 * yield() when empty or full
|
|
Me@0
|
240 *
|
|
Me@0
|
241 *It doesn't need any atomic instructions because only a single thread
|
|
Me@0
|
242 * extracts and only a single thread inserts, and it has no locations that
|
|
Me@0
|
243 * are written by both. It writes before moving and moves before reading,
|
|
Me@0
|
244 * and never lets write position and read position be the same, so dis-
|
|
Me@0
|
245 * synchrony can only ever cause an unnecessary call to yield(), never a
|
|
Me@0
|
246 * wrong value (by monotonicity of movement of pointers, plus single writer
|
|
Me@0
|
247 * to pointers, plus sequence of write before change pointer, plus
|
|
Me@0
|
248 * assumptions that if thread A semantically writes X before Y, then thread
|
|
Me@0
|
249 * B will see the writes in that order.)
|
|
Me@0
|
250 */
|
|
Me@0
|
251
|
|
Me@0
|
252 SRSWQueueStruc* makeSRSWQ()
|
|
Me@0
|
253 {
|
|
Me@0
|
254 SRSWQueueStruc* retQ;
|
|
Me@0
|
255 retQ = (SRSWQueueStruc *) malloc( sizeof( SRSWQueueStruc ) );
|
|
Me@0
|
256
|
|
Me@0
|
257 retQ->extractPos = &(retQ->startOfData[0]); //side by side == empty
|
|
Me@0
|
258 retQ->insertPos = &(retQ->startOfData[1]); // so start pos's have to be
|
|
Me@0
|
259 retQ->endOfData = &(retQ->startOfData[1023]);
|
|
Me@0
|
260
|
|
Me@0
|
261 return retQ;
|
|
Me@0
|
262 }
|
|
Me@0
|
263
|
|
Me@0
|
264
|
|
Me@0
|
265 void* readSRSWQ( SRSWQueueStruc* Q )
|
|
Me@0
|
266 { void *out = 0;
|
|
Me@0
|
267 int tries = 0;
|
|
Me@0
|
268
|
|
Me@0
|
269 while( TRUE )
|
|
Me@1
|
270 {
|
|
Me@1
|
271 if( Q->insertPos - Q->extractPos != 1 &&
|
|
Me@1
|
272 !(Q->extractPos == Q->endOfData && Q->insertPos == Q->startOfData))
|
|
Me@1
|
273 { if( Q->extractPos >= Q->endOfData ) Q->extractPos = Q->startOfData;
|
|
Me@1
|
274 else Q->extractPos++; //move before read
|
|
Me@0
|
275 out = *(Q->extractPos);
|
|
Me@0
|
276 return out;
|
|
Me@0
|
277 }
|
|
Me@0
|
278 //Q is empty
|
|
Me@0
|
279 tries++;
|
|
Me@1
|
280 if( tries > SPINLOCK_TRIES ) SwitchToThread(); //Win yield()
|
|
Me@0
|
281 }
|
|
Me@0
|
282 }
|
|
Me@0
|
283
|
|
Me@1
|
284
|
|
Me@1
|
285 void* readSRSWQ_NonBlocking( SRSWQueueStruc* Q )
|
|
Me@1
|
286 { void *out = 0;
|
|
Me@1
|
287 int tries = 0;
|
|
Me@1
|
288
|
|
Me@1
|
289 while( TRUE )
|
|
Me@1
|
290 {
|
|
Me@1
|
291 if( Q->insertPos - Q->extractPos != 1 &&
|
|
Me@1
|
292 !(Q->extractPos == Q->endOfData && Q->insertPos == Q->startOfData))
|
|
Me@1
|
293 { Q->extractPos++; //move before read
|
|
Me@1
|
294 if( Q->extractPos > Q->endOfData ) Q->extractPos = Q->startOfData;
|
|
Me@1
|
295 out = *(Q->extractPos);
|
|
Me@1
|
296 return out;
|
|
Me@1
|
297 }
|
|
Me@1
|
298 //Q is empty
|
|
Me@1
|
299 tries++;
|
|
Me@1
|
300 if( tries > 2 ) return 0; //long enough for writer to finish
|
|
Me@1
|
301 }
|
|
Me@1
|
302 }
|
|
Me@1
|
303
|
|
Me@1
|
304
|
|
Me@0
|
305 void writeSRSWQ( void * in, SRSWQueueStruc* Q )
|
|
Me@0
|
306 {
|
|
Me@0
|
307 int tries = 0;
|
|
Me@0
|
308
|
|
Me@0
|
309 while( TRUE )
|
|
Me@1
|
310 {
|
|
Me@1
|
311 if( Q->extractPos - Q->insertPos != 1 &&
|
|
Me@1
|
312 !(Q->insertPos == Q->endOfData && Q->extractPos == Q->startOfData))
|
|
Me@1
|
313 { *(Q->insertPos) = in; //insert before move
|
|
Me@1
|
314 if( Q->insertPos >= Q->endOfData ) Q->insertPos = Q->startOfData;
|
|
Me@1
|
315 else Q->insertPos++;
|
|
Me@0
|
316 return;
|
|
Me@0
|
317 }
|
|
Me@0
|
318 //Q is full
|
|
Me@0
|
319 tries++;
|
|
Me@1
|
320 if( tries > SPINLOCK_TRIES ) SwitchToThread(); //Win yield()
|
|
Me@0
|
321 }
|
|
Me@0
|
322 }
|
|
Me@1
|
323
|
|
Me@1
|
324
|
|
Me@1
|
325
|
|
Me@1
|
326 //===========================================================================
|
|
Me@1
|
327 //Single reader Multiple writer super fast Q.. no atomic instrs..
|
|
Me@1
|
328
|
|
Me@1
|
329
|
|
Me@1
|
330 /*This is a blocking queue, but it uses no atomic instructions, just does
|
|
Me@1
|
331 * yield() when empty or full
|
|
Me@1
|
332 *
|
|
Me@1
|
333 *It doesn't need any atomic instructions because only a single thread
|
|
Me@1
|
334 * extracts and only a single thread inserts, and it has no locations that
|
|
Me@1
|
335 * are written by both. It writes before moving and moves before reading,
|
|
Me@1
|
336 * and never lets write position and read position be the same, so dis-
|
|
Me@1
|
337 * synchrony can only ever cause an unnecessary call to yield(), never a
|
|
Me@1
|
338 * wrong value (by monotonicity of movement of pointers, plus single writer
|
|
Me@1
|
339 * to pointers, plus sequence of write before change pointer, plus
|
|
Me@1
|
340 * assumptions that if thread A semantically writes X before Y, then thread
|
|
Me@1
|
341 * B will see the writes in that order.)
|
|
Me@1
|
342 *
|
|
Me@1
|
343 *The multi-writer version is implemented as a hierarchy. Each writer has
|
|
Me@1
|
344 * its own single-reader single-writer queue. The reader simply does a
|
|
Me@1
|
345 * round-robin harvesting from them.
|
|
Me@1
|
346 *
|
|
Me@1
|
347 *A writer must first register itself with the queue, and receives an ID back
|
|
Me@1
|
348 * It then uses that ID on each write operation.
|
|
Me@1
|
349 *
|
|
Me@1
|
350 *The implementation is:
|
|
Me@1
|
351 *Physically:
|
|
Me@1
|
352 * -] the SRMWQueueStruc holds an array of SRSWQueueStruc s
|
|
Me@1
|
353 * -] it also has read-pointer to the last queue a write was taken from.
|
|
Me@1
|
354 *
|
|
Me@1
|
355 *Action-Patterns:
|
|
Me@1
|
356 * -] To add a writer
|
|
Me@1
|
357 * --]] writer-thread calls addWriterToQ(), remember the ID it returns
|
|
Me@1
|
358 * --]] internally addWriterToQ does:
|
|
Me@1
|
359 * ---]]] if needs more room, makes a larger writer-array
|
|
Me@1
|
360 * ---]]] copies the old writer-array into the new
|
|
Me@1
|
361 * ---]]] makes a new SRSW queue an puts it into the array
|
|
Me@1
|
362 * ---]]] returns the index to the new SRSW queue as the ID
|
|
Me@1
|
363 * -] To write
|
|
Me@1
|
364 * --]] writer thread calls writeSRMWQ, passing the Q struc and its writer-ID
|
|
Me@1
|
365 * --]] this call may block, via repeated yield() calls
|
|
Me@1
|
366 * --]] internally, writeSRMWQ does:
|
|
Me@1
|
367 * ---]]] uses the writerID as index to get the SRSW queue for that writer
|
|
Me@1
|
368 * ---]]] performs writeQ on that queue (may block via repeated yield calls)
|
|
Me@1
|
369 * -] To Read
|
|
Me@1
|
370 * --]] reader calls readSRMWQ, passing the Q struc
|
|
Me@1
|
371 * --]] this call may block, via repeated yield() calls
|
|
Me@1
|
372 * --]] internally, readSRMWQ does:
|
|
Me@1
|
373 * ---]]] gets saved index of last SRSW queue read from
|
|
Me@1
|
374 * ---]]] increments index and gets indexed queue
|
|
Me@1
|
375 * ---]]] does a non-blocking read of that queue
|
|
Me@1
|
376 * ---]]] if gets something, saves index and returns that value
|
|
Me@1
|
377 * ---]]] if gets null, then goes to next queue
|
|
Me@1
|
378 * ---]]] if got null from all the queues then does yield() then tries again
|
|
Me@1
|
379 *
|
|
Me@1
|
380 *Note: "0" is used as the value null, so SRSW queues must only contain
|
|
Me@1
|
381 * pointers, and cannot use 0 as a valid pointer value.
|
|
Me@1
|
382 *
|
|
Me@1
|
383 */
|
|
Me@1
|
384
|
|
Me@1
|
385 SRMWQueueStruc* makeSRMWQ()
|
|
Me@1
|
386 { SRMWQueueStruc* retQ;
|
|
Me@1
|
387
|
|
Me@1
|
388 retQ = (SRMWQueueStruc *) malloc( sizeof( SRMWQueueStruc ) );
|
|
Me@1
|
389
|
|
Me@1
|
390 retQ->numInternalQs = 0;
|
|
Me@1
|
391 retQ->internalQsSz = 10;
|
|
Me@1
|
392 retQ->internalQs = malloc( retQ->internalQsSz * sizeof(SRSWQueueStruc *));
|
|
Me@1
|
393
|
|
Me@1
|
394 retQ->lastQReadFrom = 0;
|
|
Me@1
|
395
|
|
Me@1
|
396 return retQ;
|
|
Me@1
|
397 }
|
|
Me@1
|
398
|
|
Me@1
|
399 /* ---]]] if needs more room, makes a larger writer-array
|
|
Me@1
|
400 * ---]]] copies the old writer-array into the new
|
|
Me@1
|
401 * ---]]] makes a new SRSW queue an puts it into the array
|
|
Me@1
|
402 * ---]]] returns the index to the new SRSW queue as the ID
|
|
Me@1
|
403 *
|
|
Me@1
|
404 *NOTE: assuming all adds are completed before any writes or reads are
|
|
Me@1
|
405 * performed.. otherwise, this needs to be re-done carefully, probably with
|
|
Me@1
|
406 * a lock.
|
|
Me@1
|
407 */
|
|
Me@1
|
408 int addWriterToSRMWQ( SRMWQueueStruc* Q )
|
|
Me@1
|
409 { int oldSz, i;
|
|
Me@1
|
410 SRSWQueueStruc * *oldArray;
|
|
Me@1
|
411
|
|
Me@1
|
412 (Q->numInternalQs)++;
|
|
Me@1
|
413 if( Q->numInternalQs >= Q->internalQsSz )
|
|
Me@1
|
414 { //full, so make bigger
|
|
Me@1
|
415 oldSz = Q->internalQsSz;
|
|
Me@1
|
416 oldArray = Q->internalQs;
|
|
Me@1
|
417 Q->internalQsSz *= 2;
|
|
Me@1
|
418 Q->internalQs = malloc( Q->internalQsSz * sizeof(SRSWQueueStruc *));
|
|
Me@1
|
419 for( i = 0; i < oldSz; i++ )
|
|
Me@1
|
420 { Q->internalQs[i] = oldArray[i];
|
|
Me@1
|
421 }
|
|
Me@1
|
422 free( oldArray );
|
|
Me@1
|
423 }
|
|
Me@1
|
424 Q->internalQs[ Q->numInternalQs - 1 ] = makeSRSWQ();
|
|
Me@1
|
425 return Q->numInternalQs - 1;
|
|
Me@1
|
426 }
|
|
Me@1
|
427
|
|
Me@1
|
428
|
|
Me@1
|
429 /* ---]]] gets saved index of last SRSW queue read-from
|
|
Me@1
|
430 * ---]]] increments index and gets indexed queue
|
|
Me@1
|
431 * ---]]] does a non-blocking read of that queue
|
|
Me@1
|
432 * ---]]] if gets something, saves index and returns that value
|
|
Me@1
|
433 * ---]]] if gets null, then goes to next queue
|
|
Me@1
|
434 * ---]]] if got null from all the queues then does yield() then tries again
|
|
Me@1
|
435 */
|
|
Me@1
|
436 void* readSRMWQ( SRMWQueueStruc* Q )
|
|
Me@1
|
437 { SRSWQueueStruc *readQ;
|
|
Me@1
|
438 void *readValue = 0;
|
|
Me@1
|
439 int tries = 0;
|
|
Me@1
|
440 int QToReadFrom = 0;
|
|
Me@1
|
441
|
|
Me@1
|
442 QToReadFrom = Q->lastQReadFrom;
|
|
Me@1
|
443
|
|
Me@1
|
444 while( TRUE )
|
|
Me@1
|
445 { QToReadFrom++;
|
|
Me@1
|
446 if( QToReadFrom >= Q->numInternalQs ) QToReadFrom = 0;
|
|
Me@1
|
447 readQ = Q->internalQs[ QToReadFrom ];
|
|
Me@1
|
448 readValue = readSRSWQ_NonBlocking( readQ );
|
|
Me@1
|
449
|
|
Me@1
|
450 if( readValue != 0 ) //got a value, return it
|
|
Me@1
|
451 { Q->lastQReadFrom = QToReadFrom;
|
|
Me@1
|
452 return readValue;
|
|
Me@1
|
453 }
|
|
Me@1
|
454 else //SRSW Q just read is empty
|
|
Me@1
|
455 { //check if all queues have been tried
|
|
Me@1
|
456 if( QToReadFrom == Q->lastQReadFrom ) //all the queues tried & empty
|
|
Me@1
|
457 { tries++; //give a writer a chance to finish before yield
|
|
Me@1
|
458 if( tries > SPINLOCK_TRIES ) SwitchToThread(); //Win yield()
|
|
Me@1
|
459 }
|
|
Me@1
|
460 }
|
|
Me@1
|
461 }
|
|
Me@1
|
462 }
|
|
Me@1
|
463
|
|
Me@1
|
464
|
|
Me@1
|
465 /*
|
|
Me@1
|
466 * ---]]] uses the writerID as index to get the SRSW queue for that writer
|
|
Me@1
|
467 * ---]]] performs writeQ on that queue (may block via repeated yield calls)
|
|
Me@1
|
468 */
|
|
Me@1
|
469 void writeSRMWQ( void * in, SRMWQueueStruc* Q, int writerID )
|
|
Me@1
|
470 {
|
|
Me@1
|
471 if( in == 0 ) printf( "error, wrote 0 to SRMW Q" );//TODO: throw an error
|
|
Me@1
|
472
|
|
Me@1
|
473 writeSRSWQ( in, Q->internalQs[ writerID ] );
|
|
Me@1
|
474 }
|
