Line data Source code
1 : /* Copyright (C) 2013 The PARI group.
2 :
3 : This file is part of the PARI/GP package.
4 :
5 : PARI/GP is free software; you can redistribute it and/or modify it under the
6 : terms of the GNU General Public License as published by the Free Software
7 : Foundation; either version 2 of the License, or (at your option) any later
8 : version. It is distributed in the hope that it will be useful, but WITHOUT
9 : ANY WARRANTY WHATSOEVER.
10 :
11 : Check the License for details. You should have received a copy of it, along
12 : with the package; see the file 'COPYING'. If not, write to the Free Software
13 : Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */
14 : #include <pthread.h>
15 : #include "pari.h"
16 : #include "paripriv.h"
17 : #include "mt.h"
18 : #if defined(_WIN32)
19 : # include "../systems/mingw/mingw.h"
20 : #endif
21 :
22 : #define DEBUGLEVEL DEBUGLEVEL_mt
23 :
24 : struct mt_queue
25 : {
26 : long no;
27 : pari_sp avma;
28 : struct pari_mainstack *mainstack;
29 : GEN input, output;
30 : GEN worker;
31 : long workid;
32 : pthread_cond_t cond;
33 : pthread_mutex_t mut;
34 : pthread_cond_t *pcond;
35 : pthread_mutex_t *pmut;
36 : };
37 :
38 : struct mt_pstate
39 : {
40 : pthread_t *th;
41 : struct pari_thread *pth;
42 : struct mt_queue *mq;
43 : long n, nbint, last;
44 : long pending;
45 : pthread_cond_t pcond;
46 : pthread_mutex_t pmut;
47 : };
48 :
49 : static THREAD long mt_thread_no = -1;
50 : static struct mt_pstate *pari_mt;
51 :
52 : #define LOCK(x) pthread_mutex_lock(x); do
53 : #define UNLOCK(x) while(0); pthread_mutex_unlock(x)
54 :
55 : void
56 200344696 : mt_sigint_block(void)
57 : {
58 200344696 : if (mt_thread_no>=0)
59 28044386 : pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED,NULL);
60 201843567 : }
61 :
62 : void
63 200763753 : mt_sigint_unblock(void)
64 : {
65 200763753 : if (mt_thread_no>=0)
66 28451127 : pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS,NULL);
67 201842538 : }
68 :
69 : void
70 1807 : mt_err_recover(long er)
71 : {
72 1807 : if (mt_thread_no>=0)
73 : {
74 10 : struct mt_pstate *mt = pari_mt;
75 10 : struct mt_queue *mq = mt->mq+mt_thread_no;
76 10 : GEN err = pari_err_last();
77 10 : err = err_get_num(err)==e_STACK ? err_e_STACK: bin_copy(copy_bin(err));
78 10 : LOCK(mq->pmut)
79 : {
80 10 : mq->output = err;
81 10 : pthread_cond_signal(mq->pcond);
82 10 : } UNLOCK(mq->pmut);
83 10 : pthread_exit((void*)1);
84 : }
85 1797 : else mtsingle_err_recover(er);
86 1797 : }
87 :
88 : void
89 0 : mt_break_recover(void)
90 : {
91 0 : if (mt_thread_no<0) mtsingle_err_recover(0);
92 0 : }
93 :
94 : void
95 0 : mt_sigint(void)
96 : {
97 0 : if (pari_mt) pthread_cond_broadcast(&pari_mt->pcond);
98 0 : }
99 :
100 : int
101 216805 : mt_is_parallel(void)
102 : {
103 216805 : return !!pari_mt;
104 : }
105 :
106 : int
107 30903558 : mt_is_thread(void)
108 : {
109 30903558 : return mt_thread_no>=0 ? 1: mtsingle_is_thread();
110 : }
111 :
112 : long
113 383559 : mt_nbthreads(void)
114 : {
115 383559 : return pari_mt ? 1: pari_mt_nbthreads;
116 : }
117 :
118 : void
119 322680 : mt_thread_init(void) { mt_thread_no = 0; }
120 :
121 : void
122 13 : mt_export_add(const char *str, GEN val)
123 : {
124 13 : if (pari_mt)
125 0 : pari_err(e_MISC,"export() not allowed during parallel sections");
126 13 : export_add(str, val);
127 13 : }
128 :
129 : void
130 8 : mt_export_del(const char *str)
131 : {
132 8 : if (pari_mt)
133 0 : pari_err(e_MISC,"unexport() not allowed during parallel sections");
134 8 : export_del(str);
135 8 : }
136 :
137 1 : void mt_broadcast(GEN code) {(void) code;}
138 :
139 260 : void pari_mt_init(void)
140 : {
141 260 : pari_mt = NULL;
142 : #ifdef _SC_NPROCESSORS_CONF
143 260 : if (!pari_mt_nbthreads) pari_mt_nbthreads = sysconf(_SC_NPROCESSORS_CONF);
144 : #elif defined(_WIN32)
145 : if (!pari_mt_nbthreads) pari_mt_nbthreads = win32_nbthreads();
146 : #else
147 : pari_mt_nbthreads = 1;
148 : #endif
149 260 : }
150 :
151 260 : void pari_mt_close(void) { }
152 :
153 : static void
154 323256 : mt_queue_cleanup(void *arg)
155 : {
156 : (void) arg;
157 323256 : pari_thread_close();
158 322142 : }
159 :
160 : static void
161 323031 : mt_queue_unlock(void *arg)
162 323031 : { pthread_mutex_unlock((pthread_mutex_t*) arg); }
163 :
164 : static void*
165 324506 : mt_queue_run(void *arg)
166 : {
167 324506 : GEN args = pari_thread_start((struct pari_thread*) arg);
168 322654 : pari_sp av = avma;
169 322654 : struct mt_queue *mq = (struct mt_queue *) args;
170 322654 : mt_thread_no = mq->no;
171 322654 : pthread_cleanup_push(mt_queue_cleanup,NULL);
172 322839 : LOCK(mq->pmut)
173 : {
174 324508 : mq->mainstack = pari_mainstack;
175 324508 : mq->avma = av;
176 324508 : pthread_cond_signal(mq->pcond);
177 324508 : } UNLOCK(mq->pmut);
178 : for(;;)
179 434722 : {
180 : GEN work, done;
181 759188 : LOCK(&mq->mut)
182 : {
183 759157 : pthread_cleanup_push(mt_queue_unlock, &mq->mut);
184 1170456 : while(!mq->input)
185 735767 : pthread_cond_wait(&mq->cond, &mq->mut);
186 434689 : pthread_cleanup_pop(0);
187 434740 : } UNLOCK(&mq->mut);
188 434713 : pari_mainstack = mq->mainstack;
189 434713 : set_avma(mq->avma);
190 434608 : work = mq->input;
191 434608 : pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS,NULL);
192 434757 : done = closure_callgenvec(mq->worker,work);
193 433689 : pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED,NULL);
194 434586 : LOCK(mq->pmut)
195 : {
196 434736 : mq->mainstack = pari_mainstack;
197 434736 : mq->avma = av;
198 434736 : mq->input = NULL;
199 434736 : mq->output = done;
200 434736 : pthread_cond_signal(mq->pcond);
201 434736 : } UNLOCK(mq->pmut);
202 : }
203 : pthread_cleanup_pop(1);
204 : #ifdef __GNUC__
205 : return NULL; /* LCOV_EXCL_LINE */
206 : #endif
207 : }
208 :
209 : static long
210 605786 : mt_queue_check(struct mt_pstate *mt)
211 : {
212 : long i;
213 4563992 : for(i=0; i<mt->n; i++)
214 : {
215 4392945 : struct mt_queue *mq = mt->mq+i;
216 4392945 : if (mq->output) return i;
217 : }
218 171047 : return -1;
219 : }
220 :
221 : static GEN
222 700343 : mtpthread_queue_get(struct mt_state *junk, long *workid, long *pending)
223 : {
224 700343 : struct mt_pstate *mt = pari_mt;
225 : struct mt_queue *mq;
226 700343 : GEN done = NULL;
227 : long last;
228 : (void) junk;
229 700343 : if (mt->nbint<mt->n)
230 : {
231 265602 : mt->last = mt->nbint;
232 265602 : *pending = mt->pending;
233 265602 : return NULL;
234 : }
235 434741 : BLOCK_SIGINT_START
236 434741 : LOCK(&mt->pmut)
237 : {
238 605786 : while ((last = mt_queue_check(mt)) < 0)
239 : {
240 171047 : pthread_cond_wait(&mt->pcond, &mt->pmut);
241 171047 : if (PARI_SIGINT_pending)
242 : {
243 2 : int sig = PARI_SIGINT_pending;
244 2 : PARI_SIGINT_pending = 0;
245 2 : pthread_mutex_unlock(&mt->pmut);
246 2 : PARI_SIGINT_block = 0;
247 2 : raise(sig);
248 0 : PARI_SIGINT_block = 1;
249 0 : pthread_mutex_lock(&mt->pmut);
250 : }
251 : }
252 434739 : } UNLOCK(&mt->pmut);
253 434739 : BLOCK_SIGINT_END
254 434739 : mq = mt->mq+last;
255 434739 : done = gcopy(mq->output);
256 434739 : mq->output = NULL;
257 434739 : if (workid) *workid = mq->workid;
258 434739 : if (typ(done) == t_ERROR)
259 : {
260 5 : if (err_get_num(done)==e_STACK)
261 0 : pari_err(e_STACKTHREAD);
262 : else
263 5 : pari_err(0,done);
264 : }
265 434734 : mt->last = last;
266 434734 : mt->pending--;
267 434734 : *pending = mt->pending;
268 434734 : return done;
269 : }
270 :
271 : static void
272 700343 : mtpthread_queue_submit(struct mt_state *junk, long workid, GEN work)
273 : {
274 700343 : struct mt_pstate *mt = pari_mt;
275 700343 : struct mt_queue *mq = mt->mq+mt->last;
276 : (void) junk;
277 700343 : if (!work) { mt->nbint=mt->n; return; }
278 434793 : BLOCK_SIGINT_START
279 434793 : if (mt->nbint<mt->n)
280 : {
281 324071 : mt->nbint++;
282 324071 : LOCK(mq->pmut)
283 : {
284 395933 : while(!mq->avma)
285 71862 : pthread_cond_wait(mq->pcond, mq->pmut);
286 324071 : } UNLOCK(mq->pmut);
287 : }
288 434793 : LOCK(&mq->mut)
289 : {
290 434793 : mq->output = NULL;
291 434793 : mq->workid = workid;
292 434793 : BLOCK_SIGINT_START
293 : {
294 434793 : pari_sp av = avma;
295 434793 : struct pari_mainstack *st = pari_mainstack;
296 434793 : pari_mainstack = mq->mainstack;
297 434793 : set_avma(mq->avma);
298 434793 : mq->input = gcopy(work);
299 434793 : mq->avma = avma;
300 434793 : mq->mainstack = pari_mainstack;
301 434793 : pari_mainstack = st;
302 434793 : set_avma(av);
303 : }
304 434793 : BLOCK_SIGINT_END
305 434793 : pthread_cond_signal(&mq->cond);
306 434793 : } UNLOCK(&mq->mut);
307 434793 : mt->pending++;
308 434793 : BLOCK_SIGINT_END
309 : }
310 :
311 : void
312 58814 : mt_queue_reset(void)
313 : {
314 58814 : struct mt_pstate *mt = pari_mt;
315 : long i;
316 58814 : BLOCK_SIGINT_START
317 383322 : for (i=0; i<mt->n; i++)
318 324508 : pthread_cancel(mt->th[i]);
319 383322 : for (i=0; i<mt->n; i++)
320 324508 : pthread_join(mt->th[i],NULL);
321 58814 : pari_mt = NULL;
322 58814 : BLOCK_SIGINT_END
323 58814 : if (DEBUGLEVEL) pari_warn(warner,"stopping %ld threads", mt->n);
324 383322 : for (i=0;i<mt->n;i++)
325 : {
326 324508 : struct mt_queue *mq = mt->mq+i;
327 324508 : pthread_cond_destroy(&mq->cond);
328 324508 : pthread_mutex_destroy(&mq->mut);
329 324508 : pari_thread_free(&mt->pth[i]);
330 : }
331 58814 : pari_free(mt->mq);
332 58814 : pari_free(mt->pth);
333 58814 : pari_free(mt->th);
334 58814 : pari_free(mt);
335 58814 : }
336 :
337 : static long
338 58814 : closure_has_clone(GEN fun)
339 : {
340 58814 : if (isclone(fun)) return 1;
341 58808 : if (lg(fun) >= 8)
342 : {
343 58249 : GEN f = closure_get_frame(fun);
344 58249 : long i, l = lg(f);
345 217701 : for (i = 1; i < l; i++)
346 160965 : if (isclone(gel(f,i))) return 1;
347 : }
348 57295 : return 0;
349 : }
350 :
351 : void
352 131215 : mt_queue_start_lim(struct pari_mt *pt, GEN worker, long lim)
353 : {
354 131215 : if (lim==0) lim = pari_mt_nbthreads;
355 131196 : else lim = minss(pari_mt_nbthreads, lim);
356 131215 : if (mt_thread_no >= 0)
357 42408 : mtsequential_queue_start(pt, worker);
358 88807 : else if (pari_mt || lim <= 1)
359 29993 : mtsingle_queue_start(pt, worker);
360 : else
361 : {
362 : struct mt_pstate *mt =
363 58814 : (struct mt_pstate*) pari_malloc(sizeof(struct mt_pstate));
364 58814 : long mtparisize = GP_DATA->threadsize? GP_DATA->threadsize: pari_mainstack->rsize;
365 58814 : long mtparisizemax = GP_DATA->threadsizemax;
366 : long i;
367 58814 : if (closure_has_clone(worker))
368 1519 : worker = gcopy(worker); /* to avoid clone_lock race */
369 58814 : mt->mq = (struct mt_queue *) pari_malloc(sizeof(*mt->mq)*lim);
370 58814 : mt->th = (pthread_t *) pari_malloc(sizeof(*mt->th)*lim);
371 58814 : mt->pth = (struct pari_thread *) pari_malloc(sizeof(*mt->pth)*lim);
372 58814 : mt->pending = 0;
373 58814 : mt->n = lim;
374 58814 : mt->nbint = 0;
375 58814 : mt->last = 0;
376 58814 : pthread_cond_init(&mt->pcond,NULL);
377 58814 : pthread_mutex_init(&mt->pmut,NULL);
378 383322 : for (i=0;i<lim;i++)
379 : {
380 324508 : struct mt_queue *mq = mt->mq+i;
381 324508 : mq->no = i;
382 324508 : mq->avma = 0;
383 324508 : mq->mainstack = NULL;
384 324508 : mq->worker = worker;
385 324508 : mq->input = NULL;
386 324508 : mq->output = NULL;
387 324508 : mq->pcond = &mt->pcond;
388 324508 : mq->pmut = &mt->pmut;
389 324508 : pthread_cond_init(&mq->cond,NULL);
390 324508 : pthread_mutex_init(&mq->mut,NULL);
391 324508 : if (mtparisizemax)
392 0 : pari_thread_valloc(&mt->pth[i],mtparisize,mtparisizemax,(GEN)mq);
393 : else
394 324508 : pari_thread_alloc(&mt->pth[i],mtparisize,(GEN)mq);
395 : }
396 58814 : if (DEBUGLEVEL) pari_warn(warner,"starting %ld threads", lim);
397 58814 : BLOCK_SIGINT_START
398 383322 : for (i=0;i<lim;i++)
399 324508 : pthread_create(&mt->th[i],NULL, &mt_queue_run, (void*)&mt->pth[i]);
400 58814 : pari_mt = mt;
401 58814 : BLOCK_SIGINT_END
402 58814 : pt->get=&mtpthread_queue_get;
403 58814 : pt->submit=&mtpthread_queue_submit;
404 58814 : pt->end=&mt_queue_reset;
405 : }
406 131215 : }
|
