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