| 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@0
|
25 QueueStruc* makeQ()
|
|
Me@0
|
26 {
|
|
Me@0
|
27 QueueStruc* retQ;
|
|
Me@0
|
28 int status;
|
|
Me@0
|
29 retQ = (QueueStruc *) malloc( sizeof( QueueStruc ) );
|
|
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@0
|
64 void * readQ( QueueStruc *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@0
|
92 void writeQ( void * in, QueueStruc* 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@0
|
155 while( !success )
|
|
Me@0
|
156 { success =
|
|
Me@0
|
157 __sync_bool_compare_and_swap( &(Q->extractLock), UNLOCKED, LOCKED );
|
|
Me@0
|
158 if( success )
|
|
Me@0
|
159 {
|
|
Me@1
|
160 void **insertPos = Q->insertPos;
|
|
Me@1
|
161 void **extractPos = Q->extractPos;
|
|
Me@0
|
162
|
|
Me@0
|
163 //if not empty -- extract just below insert when empty
|
|
Me@0
|
164 if( insertPos - extractPos != 1 &&
|
|
Me@0
|
165 !(extractPos == endOfData && insertPos == startOfData))
|
|
Me@0
|
166 { //move before read
|
|
Me@0
|
167 if( extractPos == endOfData ) //write new pos exactly once, correctly
|
|
Me@0
|
168 { Q->extractPos = startOfData; //can't overrun then fix it 'cause
|
|
Me@0
|
169 } // other thread might read bad pos
|
|
Me@0
|
170 else
|
|
Me@0
|
171 { Q->extractPos++;
|
|
Me@0
|
172 }
|
|
Me@0
|
173 out = *(Q->extractPos);
|
|
Me@0
|
174 Q->extractLock = UNLOCKED;
|
|
Me@0
|
175 return out;
|
|
Me@0
|
176 }
|
|
Me@0
|
177 else //Q is empty
|
|
Me@0
|
178 { success = FALSE;
|
|
Me@0
|
179 Q->extractLock = UNLOCKED;//have to try again, release for others
|
|
Me@0
|
180 }
|
|
Me@0
|
181 }
|
|
Me@0
|
182 //Q is busy or empty
|
|
Me@0
|
183 tries++;
|
|
Me@1
|
184 if( tries > SPINLOCK_TRIES ) SwitchToThread(); //WinAPI yield()
|
|
Me@0
|
185 }
|
|
Me@0
|
186 }
|
|
Me@0
|
187
|
|
Me@0
|
188 void writeCASQ( void * in, CASQueueStruc* Q )
|
|
Me@0
|
189 {
|
|
Me@0
|
190 int tries = 0;
|
|
Me@1
|
191 //TODO: need to make Q volatile? Want to do this Q in assembly!
|
|
Me@1
|
192 //Have no idea what GCC's going to do to this code
|
|
Me@1
|
193 void **startOfData = Q->startOfData;
|
|
Me@1
|
194 void **endOfData = Q->endOfData;
|
|
Me@1
|
195
|
|
Me@0
|
196 int success = FALSE;
|
|
Me@0
|
197
|
|
Me@0
|
198 while( !success )
|
|
Me@0
|
199 { success =
|
|
Me@0
|
200 __sync_bool_compare_and_swap( &(Q->insertLock), UNLOCKED, LOCKED );
|
|
Me@0
|
201 if( success )
|
|
Me@0
|
202 {
|
|
Me@1
|
203 void **insertPos = Q->insertPos;
|
|
Me@1
|
204 void **extractPos = Q->extractPos;
|
|
Me@0
|
205
|
|
Me@0
|
206 //check if room to insert.. can't use a count variable
|
|
Me@0
|
207 // 'cause both insertor Thd and extractor Thd would write it
|
|
Me@0
|
208 if( extractPos - insertPos != 1 &&
|
|
Me@0
|
209 !(insertPos == endOfData && extractPos == startOfData))
|
|
Me@1
|
210 { *(Q->insertPos) = in; //insert before move
|
|
Me@0
|
211 if( insertPos == endOfData ) //write new pos exactly once, correctly
|
|
Me@0
|
212 { Q->insertPos = startOfData;
|
|
Me@0
|
213 }
|
|
Me@0
|
214 else
|
|
Me@0
|
215 { Q->insertPos++;
|
|
Me@0
|
216 }
|
|
Me@0
|
217 Q->insertLock = UNLOCKED;
|
|
Me@0
|
218 return;
|
|
Me@0
|
219 }
|
|
Me@0
|
220 else //Q is full
|
|
Me@0
|
221 { success = FALSE;
|
|
Me@0
|
222 Q->insertLock = UNLOCKED;//have to try again, release for others
|
|
Me@0
|
223 }
|
|
Me@0
|
224 }
|
|
Me@0
|
225 tries++;
|
|
Me@1
|
226 if( tries > SPINLOCK_TRIES ) SwitchToThread(); //Win yield()
|
|
Me@0
|
227 }
|
|
Me@0
|
228 }
|
|
Me@0
|
229
|
|
Me@0
|
230 #endif //_GNU_SOURCE
|
|
Me@0
|
231
|
|
Me@1
|
232
|
|
Me@0
|
233 //===========================================================================
|
|
Me@0
|
234 //Single reader single writer super fast Q.. no atomic instrs..
|
|
Me@0
|
235
|
|
Me@0
|
236
|
|
Me@0
|
237 /*This is a blocking queue, but it uses no atomic instructions, just does
|
|
Me@1
|
238 * yield() when empty or full
|
|
Me@0
|
239 *
|
|
Me@0
|
240 *It doesn't need any atomic instructions because only a single thread
|
|
Me@0
|
241 * extracts and only a single thread inserts, and it has no locations that
|
|
Me@0
|
242 * are written by both. It writes before moving and moves before reading,
|
|
Me@0
|
243 * and never lets write position and read position be the same, so dis-
|
|
Me@0
|
244 * synchrony can only ever cause an unnecessary call to yield(), never a
|
|
Me@0
|
245 * wrong value (by monotonicity of movement of pointers, plus single writer
|
|
Me@0
|
246 * to pointers, plus sequence of write before change pointer, plus
|
|
Me@0
|
247 * assumptions that if thread A semantically writes X before Y, then thread
|
|
Me@0
|
248 * B will see the writes in that order.)
|
|
Me@0
|
249 */
|
|
Me@0
|
250
|
|
Me@0
|
251 SRSWQueueStruc* makeSRSWQ()
|
|
Me@0
|
252 {
|
|
Me@0
|
253 SRSWQueueStruc* retQ;
|
|
Me@0
|
254 retQ = (SRSWQueueStruc *) malloc( sizeof( SRSWQueueStruc ) );
|
|
Me@0
|
255
|
|
Me@0
|
256 retQ->extractPos = &(retQ->startOfData[0]); //side by side == empty
|
|
Me@0
|
257 retQ->insertPos = &(retQ->startOfData[1]); // so start pos's have to be
|
|
Me@0
|
258 retQ->endOfData = &(retQ->startOfData[1023]);
|
|
Me@0
|
259
|
|
Me@0
|
260 return retQ;
|
|
Me@0
|
261 }
|
|
Me@0
|
262
|
|
Me@0
|
263
|
|
Me@0
|
264 void* readSRSWQ( SRSWQueueStruc* Q )
|
|
Me@0
|
265 { void *out = 0;
|
|
Me@0
|
266 int tries = 0;
|
|
Me@0
|
267
|
|
Me@0
|
268 while( TRUE )
|
|
Me@1
|
269 {
|
|
Me@1
|
270 if( Q->insertPos - Q->extractPos != 1 &&
|
|
Me@1
|
271 !(Q->extractPos == Q->endOfData && Q->insertPos == Q->startOfData))
|
|
Me@1
|
272 { if( Q->extractPos >= Q->endOfData ) Q->extractPos = Q->startOfData;
|
|
Me@1
|
273 else Q->extractPos++; //move before read
|
|
Me@0
|
274 out = *(Q->extractPos);
|
|
Me@0
|
275 return out;
|
|
Me@0
|
276 }
|
|
Me@0
|
277 //Q is empty
|
|
Me@0
|
278 tries++;
|
|
Me@1
|
279 if( tries > SPINLOCK_TRIES ) SwitchToThread(); //Win yield()
|
|
Me@0
|
280 }
|
|
Me@0
|
281 }
|
|
Me@0
|
282
|
|
Me@1
|
283
|
|
Me@1
|
284 void* readSRSWQ_NonBlocking( SRSWQueueStruc* Q )
|
|
Me@1
|
285 { void *out = 0;
|
|
Me@1
|
286 int tries = 0;
|
|
Me@1
|
287
|
|
Me@1
|
288 while( TRUE )
|
|
Me@1
|
289 {
|
|
Me@1
|
290 if( Q->insertPos - Q->extractPos != 1 &&
|
|
Me@1
|
291 !(Q->extractPos == Q->endOfData && Q->insertPos == Q->startOfData))
|
|
Me@1
|
292 { Q->extractPos++; //move before read
|
|
Me@1
|
293 if( Q->extractPos > Q->endOfData ) Q->extractPos = Q->startOfData;
|
|
Me@1
|
294 out = *(Q->extractPos);
|
|
Me@1
|
295 return out;
|
|
Me@1
|
296 }
|
|
Me@1
|
297 //Q is empty
|
|
Me@1
|
298 tries++;
|
|
Me@1
|
299 if( tries > 2 ) return 0; //long enough for writer to finish
|
|
Me@1
|
300 }
|
|
Me@1
|
301 }
|
|
Me@1
|
302
|
|
Me@1
|
303
|
|
Me@0
|
304 void writeSRSWQ( void * in, SRSWQueueStruc* Q )
|
|
Me@0
|
305 {
|
|
Me@0
|
306 int tries = 0;
|
|
Me@0
|
307
|
|
Me@0
|
308 while( TRUE )
|
|
Me@1
|
309 {
|
|
Me@1
|
310 if( Q->extractPos - Q->insertPos != 1 &&
|
|
Me@1
|
311 !(Q->insertPos == Q->endOfData && Q->extractPos == Q->startOfData))
|
|
Me@1
|
312 { *(Q->insertPos) = in; //insert before move
|
|
Me@1
|
313 if( Q->insertPos >= Q->endOfData ) Q->insertPos = Q->startOfData;
|
|
Me@1
|
314 else Q->insertPos++;
|
|
Me@0
|
315 return;
|
|
Me@0
|
316 }
|
|
Me@0
|
317 //Q is full
|
|
Me@0
|
318 tries++;
|
|
Me@1
|
319 if( tries > SPINLOCK_TRIES ) SwitchToThread(); //Win yield()
|
|
Me@0
|
320 }
|
|
Me@0
|
321 }
|
|
Me@1
|
322
|
|
Me@1
|
323
|
|
Me@1
|
324
|
|
Me@1
|
325 //===========================================================================
|
|
Me@1
|
326 //Single reader Multiple writer super fast Q.. no atomic instrs..
|
|
Me@1
|
327
|
|
Me@1
|
328
|
|
Me@1
|
329 /*This is a blocking queue, but it uses no atomic instructions, just does
|
|
Me@1
|
330 * yield() when empty or full
|
|
Me@1
|
331 *
|
|
Me@1
|
332 *It doesn't need any atomic instructions because only a single thread
|
|
Me@1
|
333 * extracts and only a single thread inserts, and it has no locations that
|
|
Me@1
|
334 * are written by both. It writes before moving and moves before reading,
|
|
Me@1
|
335 * and never lets write position and read position be the same, so dis-
|
|
Me@1
|
336 * synchrony can only ever cause an unnecessary call to yield(), never a
|
|
Me@1
|
337 * wrong value (by monotonicity of movement of pointers, plus single writer
|
|
Me@1
|
338 * to pointers, plus sequence of write before change pointer, plus
|
|
Me@1
|
339 * assumptions that if thread A semantically writes X before Y, then thread
|
|
Me@1
|
340 * B will see the writes in that order.)
|
|
Me@1
|
341 *
|
|
Me@1
|
342 *The multi-writer version is implemented as a hierarchy. Each writer has
|
|
Me@1
|
343 * its own single-reader single-writer queue. The reader simply does a
|
|
Me@1
|
344 * round-robin harvesting from them.
|
|
Me@1
|
345 *
|
|
Me@1
|
346 *A writer must first register itself with the queue, and receives an ID back
|
|
Me@1
|
347 * It then uses that ID on each write operation.
|
|
Me@1
|
348 *
|
|
Me@1
|
349 *The implementation is:
|
|
Me@1
|
350 *Physically:
|
|
Me@1
|
351 * -] the SRMWQueueStruc holds an array of SRSWQueueStruc s
|
|
Me@1
|
352 * -] it also has read-pointer to the last queue a write was taken from.
|
|
Me@1
|
353 *
|
|
Me@1
|
354 *Action-Patterns:
|
|
Me@1
|
355 * -] To add a writer
|
|
Me@1
|
356 * --]] writer-thread calls addWriterToQ(), remember the ID it returns
|
|
Me@1
|
357 * --]] internally addWriterToQ does:
|
|
Me@1
|
358 * ---]]] if needs more room, makes a larger writer-array
|
|
Me@1
|
359 * ---]]] copies the old writer-array into the new
|
|
Me@1
|
360 * ---]]] makes a new SRSW queue an puts it into the array
|
|
Me@1
|
361 * ---]]] returns the index to the new SRSW queue as the ID
|
|
Me@1
|
362 * -] To write
|
|
Me@1
|
363 * --]] writer thread calls writeSRMWQ, passing the Q struc and its writer-ID
|
|
Me@1
|
364 * --]] this call may block, via repeated yield() calls
|
|
Me@1
|
365 * --]] internally, writeSRMWQ does:
|
|
Me@1
|
366 * ---]]] uses the writerID as index to get the SRSW queue for that writer
|
|
Me@1
|
367 * ---]]] performs writeQ on that queue (may block via repeated yield calls)
|
|
Me@1
|
368 * -] To Read
|
|
Me@1
|
369 * --]] reader calls readSRMWQ, passing the Q struc
|
|
Me@1
|
370 * --]] this call may block, via repeated yield() calls
|
|
Me@1
|
371 * --]] internally, readSRMWQ does:
|
|
Me@1
|
372 * ---]]] gets saved index of last SRSW queue read from
|
|
Me@1
|
373 * ---]]] increments index and gets indexed queue
|
|
Me@1
|
374 * ---]]] does a non-blocking read of that queue
|
|
Me@1
|
375 * ---]]] if gets something, saves index and returns that value
|
|
Me@1
|
376 * ---]]] if gets null, then goes to next queue
|
|
Me@1
|
377 * ---]]] if got null from all the queues then does yield() then tries again
|
|
Me@1
|
378 *
|
|
Me@1
|
379 *Note: "0" is used as the value null, so SRSW queues must only contain
|
|
Me@1
|
380 * pointers, and cannot use 0 as a valid pointer value.
|
|
Me@1
|
381 *
|
|
Me@1
|
382 */
|
|
Me@1
|
383
|
|
Me@1
|
384 SRMWQueueStruc* makeSRMWQ()
|
|
Me@1
|
385 { SRMWQueueStruc* retQ;
|
|
Me@1
|
386
|
|
Me@1
|
387 retQ = (SRMWQueueStruc *) malloc( sizeof( SRMWQueueStruc ) );
|
|
Me@1
|
388
|
|
Me@1
|
389 retQ->numInternalQs = 0;
|
|
Me@1
|
390 retQ->internalQsSz = 10;
|
|
Me@1
|
391 retQ->internalQs = malloc( retQ->internalQsSz * sizeof(SRSWQueueStruc *));
|
|
Me@1
|
392
|
|
Me@1
|
393 retQ->lastQReadFrom = 0;
|
|
Me@1
|
394
|
|
Me@1
|
395 return retQ;
|
|
Me@1
|
396 }
|
|
Me@1
|
397
|
|
Me@1
|
398 /* ---]]] if needs more room, makes a larger writer-array
|
|
Me@1
|
399 * ---]]] copies the old writer-array into the new
|
|
Me@1
|
400 * ---]]] makes a new SRSW queue an puts it into the array
|
|
Me@1
|
401 * ---]]] returns the index to the new SRSW queue as the ID
|
|
Me@1
|
402 *
|
|
Me@1
|
403 *NOTE: assuming all adds are completed before any writes or reads are
|
|
Me@1
|
404 * performed.. otherwise, this needs to be re-done carefully, probably with
|
|
Me@1
|
405 * a lock.
|
|
Me@1
|
406 */
|
|
Me@1
|
407 int addWriterToSRMWQ( SRMWQueueStruc* Q )
|
|
Me@1
|
408 { int oldSz, i;
|
|
Me@1
|
409 SRSWQueueStruc * *oldArray;
|
|
Me@1
|
410
|
|
Me@1
|
411 (Q->numInternalQs)++;
|
|
Me@1
|
412 if( Q->numInternalQs >= Q->internalQsSz )
|
|
Me@1
|
413 { //full, so make bigger
|
|
Me@1
|
414 oldSz = Q->internalQsSz;
|
|
Me@1
|
415 oldArray = Q->internalQs;
|
|
Me@1
|
416 Q->internalQsSz *= 2;
|
|
Me@1
|
417 Q->internalQs = malloc( Q->internalQsSz * sizeof(SRSWQueueStruc *));
|
|
Me@1
|
418 for( i = 0; i < oldSz; i++ )
|
|
Me@1
|
419 { Q->internalQs[i] = oldArray[i];
|
|
Me@1
|
420 }
|
|
Me@1
|
421 free( oldArray );
|
|
Me@1
|
422 }
|
|
Me@1
|
423 Q->internalQs[ Q->numInternalQs - 1 ] = makeSRSWQ();
|
|
Me@1
|
424 return Q->numInternalQs - 1;
|
|
Me@1
|
425 }
|
|
Me@1
|
426
|
|
Me@1
|
427
|
|
Me@1
|
428 /* ---]]] gets saved index of last SRSW queue read-from
|
|
Me@1
|
429 * ---]]] increments index and gets indexed queue
|
|
Me@1
|
430 * ---]]] does a non-blocking read of that queue
|
|
Me@1
|
431 * ---]]] if gets something, saves index and returns that value
|
|
Me@1
|
432 * ---]]] if gets null, then goes to next queue
|
|
Me@1
|
433 * ---]]] if got null from all the queues then does yield() then tries again
|
|
Me@1
|
434 */
|
|
Me@1
|
435 void* readSRMWQ( SRMWQueueStruc* Q )
|
|
Me@1
|
436 { SRSWQueueStruc *readQ;
|
|
Me@1
|
437 void *readValue = 0;
|
|
Me@1
|
438 int tries = 0;
|
|
Me@1
|
439 int QToReadFrom = 0;
|
|
Me@1
|
440
|
|
Me@1
|
441 QToReadFrom = Q->lastQReadFrom;
|
|
Me@1
|
442
|
|
Me@1
|
443 while( TRUE )
|
|
Me@1
|
444 { QToReadFrom++;
|
|
Me@1
|
445 if( QToReadFrom >= Q->numInternalQs ) QToReadFrom = 0;
|
|
Me@1
|
446 readQ = Q->internalQs[ QToReadFrom ];
|
|
Me@1
|
447 readValue = readSRSWQ_NonBlocking( readQ );
|
|
Me@1
|
448
|
|
Me@1
|
449 if( readValue != 0 ) //got a value, return it
|
|
Me@1
|
450 { Q->lastQReadFrom = QToReadFrom;
|
|
Me@1
|
451 return readValue;
|
|
Me@1
|
452 }
|
|
Me@1
|
453 else //SRSW Q just read is empty
|
|
Me@1
|
454 { //check if all queues have been tried
|
|
Me@1
|
455 if( QToReadFrom == Q->lastQReadFrom ) //all the queues tried & empty
|
|
Me@1
|
456 { tries++; //give a writer a chance to finish before yield
|
|
Me@1
|
457 if( tries > SPINLOCK_TRIES ) SwitchToThread(); //Win yield()
|
|
Me@1
|
458 }
|
|
Me@1
|
459 }
|
|
Me@1
|
460 }
|
|
Me@1
|
461 }
|
|
Me@1
|
462
|
|
Me@1
|
463
|
|
Me@1
|
464 /*
|
|
Me@1
|
465 * ---]]] uses the writerID as index to get the SRSW queue for that writer
|
|
Me@1
|
466 * ---]]] performs writeQ on that queue (may block via repeated yield calls)
|
|
Me@1
|
467 */
|
|
Me@1
|
468 void writeSRMWQ( void * in, SRMWQueueStruc* Q, int writerID )
|
|
Me@1
|
469 {
|
|
Me@1
|
470 if( in == 0 ) printf( "error, wrote 0 to SRMW Q" );//TODO: throw an error
|
|
Me@1
|
471
|
|
Me@1
|
472 writeSRSWQ( in, Q->internalQs[ writerID ] );
|
|
Me@1
|
473 }
|
