1: 
  2: #define _MT 1
  3: 
  4: #include <stdio.h>
  5: #include <stdlib.h>
  6: static char __file__[] = __FILE__;      /* for tassert.h                */
  7: #include        "tassert.h"
  8: 
  9: #if _WIN32
 10: 
 11: #include <windows.h>
 12: #include <errno.h>
 13: #include <process.h>
 14: 
 15: #include "root.h"
 16: #include "rmem.h"
 17: 
 18: static unsigned __stdcall startthread(void *p);
 19: 
 20: struct FileData
 21: {
 22:     File *file;
 23:     int result;
 24:     HANDLE event;
 25: };
 26: 
 27: struct AsyncRead
 28: {
 29:     static AsyncRead *create(size_t nfiles);
 30:     void addFile(File *file);
 31:     void start();
 32:     int read(size_t i);
 33:     static void dispose(AsyncRead *);
 34: 
 35:     HANDLE hThread;
 36: 
 37:     size_t filesdim;
 38:     size_t filesmax;
 39:     FileData files[1];
 40: };
 41: 
 42: 
 43: AsyncRead *AsyncRead::create(size_t nfiles)
 44: {
 45:     AsyncRead *aw = (AsyncRead *)mem.calloc(1, sizeof(AsyncRead) +
 46:                                 (nfiles - 1) * sizeof(FileData));
 47:     aw->filesmax = nfiles;
 48:     return aw;
 49: }
 50: 
 51: void AsyncRead::addFile(File *file)
 52: {
 53:     //printf("addFile(file = %p)\n", file);
 54:     //printf("filesdim = %d, filesmax = %d\n", filesdim, filesmax);
 55:     assert(filesdim < filesmax);
 56:     files[filesdim].file = file;
 57:     files[filesdim].event = CreateEvent(NULL, TRUE, FALSE, NULL);
 58:     ResetEvent(files[filesdim].event);
warning C6387: 'argument 1' might be '0': this does not adhere to the specification for the function 'ResetEvent': Lines: 55, 56, 57, 58
 59:     filesdim++;
 60: }
 61: 
 62: void AsyncRead::start()
 63: {
 64:     //printf("aw->filesdim = %p %d\n", this, filesdim);
 65:     if (filesdim)
 66:     {
 67:         unsigned threadaddr;
 68:         hThread = (HANDLE) _beginthreadex(NULL,
 69:             0,
 70:             &startthread,
 71:             this,
 72:             0,
 73:             (unsigned *)&threadaddr);
 74: 
 75:         if (hThread)
 76:         {
 77:             SetThreadPriority(hThread, THREAD_PRIORITY_HIGHEST);
 78:         }
 79:         else
 80:         {
 81:             assert(0);
 82:         }
 83:     }
 84: }
 85: 
 86: int AsyncRead::read(size_t i)
 87: {
 88:     FileData *f = &files[i];
 89:     WaitForSingleObject(f->event, INFINITE);
 90:     Sleep(0);                   // give up time slice
 91:     return f->result;
 92: }
 93: 
 94: void AsyncRead::dispose(AsyncRead *aw)
 95: {
 96:     delete aw;
 97: }
 98: 
 99: 
100: 
101: unsigned __stdcall startthread(void *p)
102: {
103:     AsyncRead *aw = (AsyncRead *)p;
104: 
105:     //printf("aw->filesdim = %p %d\n", aw, aw->filesdim);
106:     for (size_t i = 0; i < aw->filesdim; i++)
107:     {   FileData *f = &aw->files[i];
108: 
109:         f->result = f->file->read();
110:         SetEvent(f->event);
111:     }
112:     _endthreadex(EXIT_SUCCESS);
113:     return EXIT_SUCCESS;                // if skidding
114: }
115: 
116: #elif linux  // Posix
117: 
118: #include <errno.h>
119: #include <pthread.h>
120: #include <time.h>
121: 
122: #include "root.h"
123: #include "rmem.h"
124: 
125: void *startthread(void *arg);
126: 
127: void err_abort(int status, const char *msg)
128: {
129:     fprintf(stderr, "fatal error = %d, %s\n", status, msg);
130:     exit(EXIT_FAILURE);
131: }
132: 
133: struct FileData
134: {
135:     File *file;
136:     int result;
137: 
138:     pthread_mutex_t mutex;
139:     pthread_cond_t cond;
140:     int value;
141: };
142: 
143: struct AsyncRead
144: {
145:     static AsyncRead *create(size_t nfiles);
146:     void addFile(File *file);
147:     void start();
148:     int read(size_t i);
149:     static void dispose(AsyncRead *);
150: 
151:     size_t filesdim;
152:     size_t filesmax;
153:     FileData files[1];
154: };
155: 
156: 
157: AsyncRead *AsyncRead::create(size_t nfiles)
158: {
159:     AsyncRead *aw = (AsyncRead *)mem.calloc(1, sizeof(AsyncRead) +
160:                                 (nfiles - 1) * sizeof(FileData));
161:     aw->filesmax = nfiles;
162:     return aw;
163: }
164: 
165: void AsyncRead::addFile(File *file)
166: {
167:     //printf("addFile(file = %p)\n", file);
168:     //printf("filesdim = %d, filesmax = %d\n", filesdim, filesmax);
169:     assert(filesdim < filesmax);
170:     FileData *f = &files[filesdim];
171:     f->file = file;
172: 
173:     int status = pthread_mutex_init(&f->mutex, NULL);
174:     if (status != 0)
175:         err_abort(status, "init mutex");
176:     status = pthread_cond_init(&f->cond, NULL);
177:     if (status != 0)
178:         err_abort(status, "init cond");
179: 
180:     filesdim++;
181: }
182: 
183: void AsyncRead::start()
184: {
185:     //printf("aw->filesdim = %p %d\n", this, filesdim);
186:     if (filesdim)
187:     {
188:         pthread_t thread_id;
189:         int status = pthread_create(&thread_id,
190:             NULL,
191:             &startthread,
192:             this);
193:         if (status != 0)
194:             err_abort(status, "create thread");
195:     }
196: }
197: 
198: int AsyncRead::read(size_t i)
199: {
200:     FileData *f = &files[i];
201: 
202:     // Wait for the event
203:     int status = pthread_mutex_lock(&f->mutex);
204:     if (status != 0)
205:         err_abort(status, "lock mutex");
206:     while (f->value == 0)
207:     {
208:         status = pthread_cond_wait(&f->cond, &f->mutex);
209:         if (status != 0)
210:             err_abort(status, "wait on condition");
211:     }
212:     status = pthread_mutex_unlock(&f->mutex);
213:     if (status != 0)
214:         err_abort(status, "unlock mutex");
215: 
216:     return f->result;
217: }
218: 
219: void AsyncRead::dispose(AsyncRead *aw)
220: {
221:     //printf("AsyncRead::dispose()\n");
222:     for (int i = 0; i < aw->filesdim; i++)
223:     {
224:         FileData *f = &aw->files[i];
225:         int status = pthread_cond_destroy(&f->cond);
226:         if (status != 0)
227:             err_abort(status, "cond destroy");
228:         status = pthread_mutex_destroy(&f->mutex);
229:         if (status != 0)
230:             err_abort(status, "mutex destroy");
231:     }
232:     delete aw;
233: }
234: 
235: 
236: void *startthread(void *p)
237: {
238:     AsyncRead *aw = (AsyncRead *)p;
239: 
240:     //printf("startthread: aw->filesdim = %p %d\n", aw, aw->filesdim);
241:     size_t dim = aw->filesdim;
242:     for (size_t i = 0; i < dim; i++)
243:     {   FileData *f = &aw->files[i];
244: 
245:         f->result = f->file->read();
246: 
247:         // Set event
248:         int status = pthread_mutex_lock(&f->mutex);
249:         if (status != 0)
250:             err_abort(status, "lock mutex");
251:         f->value = 1;
252:         status = pthread_cond_signal(&f->cond);
253:         if (status != 0)
254:             err_abort(status, "signal condition");
255:         status = pthread_mutex_unlock(&f->mutex);
256:         if (status != 0)
257:             err_abort(status, "unlock mutex");
258:     }
259: 
260:     return NULL;                        // end thread
261: }
262: 
263: #else
264: 
265: #include <stdio.h>
266: #include <errno.h>
267: 
268: #include "root.h"
269: #include "rmem.h"
270: 
271: struct FileData
272: {
273:     File *file;
274:     int result;
275:     //HANDLE event;
276: };
277: 
278: struct AsyncRead
279: {
280:     static AsyncRead *create(size_t nfiles);
281:     void addFile(File *file);
282:     void start();
283:     int read(size_t i);
284:     static void dispose(AsyncRead *);
285: 
286:     //HANDLE hThread;
287: 
288:     size_t filesdim;
289:     size_t filesmax;
290:     FileData files[1];
291: };
292: 
293: 
294: AsyncRead *AsyncRead::create(size_t nfiles)
295: {
296:     AsyncRead *aw = (AsyncRead *)mem.calloc(1, sizeof(AsyncRead) +
297:                                 (nfiles - 1) * sizeof(FileData));
298:     aw->filesmax = nfiles;
299:     return aw;
300: }
301: 
302: void AsyncRead::addFile(File *file)
303: {
304:     //printf("addFile(file = %p)\n", file);
305:     //printf("filesdim = %d, filesmax = %d\n", filesdim, filesmax);
306:     assert(filesdim < filesmax);
307:     files[filesdim].file = file;
308:     //files[filesdim].event = CreateEvent(NULL, TRUE, FALSE, NULL);
309:     //ResetEvent(files[filesdim].event);
310:     filesdim++;
311: }
312: 
313: void AsyncRead::start()
314: {
315: }
316: 
317: int AsyncRead::read(size_t i)
318: {
319:     FileData *f = &files[i];
320:     f->result = f->file->read();
321:     return f->result;
322: }
323: 
324: void AsyncRead::dispose(AsyncRead *aw)
325: {
326:     delete aw;
327: }
328: 
329: #endif
330: