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 215096804 : mt_sigint_block(void)
57 : {
58 215096804 : if (mt_thread_no>=0)
59 35641027 : pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED,NULL);
60 215096804 : }
61 :
62 : void
63 215096801 : mt_sigint_unblock(void)
64 : {
65 215096801 : if (mt_thread_no>=0)
66 35641027 : pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS,NULL);
67 215096791 : }
68 :
69 : void
70 1869 : mt_err_recover(long er)
71 : {
72 1869 : 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 : pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED,NULL);
79 10 : LOCK(mq->pmut)
80 : {
81 10 : mq->output = err;
82 10 : pthread_cond_signal(mq->pcond);
83 10 : } UNLOCK(mq->pmut);
84 10 : pthread_exit((void*)1);
85 : }
86 1859 : else mtsingle_err_recover(er);
87 1859 : }
88 :
89 : void
90 0 : mt_break_recover(void)
91 : {
92 0 : if (mt_thread_no<0) mtsingle_err_recover(0);
93 0 : }
94 :
95 : void
96 0 : mt_sigint(void)
97 : {
98 0 : if (pari_mt) pthread_cond_broadcast(&pari_mt->pcond);
99 0 : }
100 :
101 : int
102 220910 : mt_is_parallel(void)
103 : {
104 220910 : return !!pari_mt;
105 : }
106 :
107 : int
108 31106809 : mt_is_thread(void)
109 : {
110 31106809 : return mt_thread_no>=0 ? 1: mtsingle_is_thread();
111 : }
112 :
113 : long
114 389769 : mt_nbthreads(void)
115 : {
116 389769 : return pari_mt ? 1: pari_mt_nbthreads;
117 : }
118 :
119 : void
120 376736 : mt_thread_init(void) { mt_thread_no = 0; }
121 :
122 : void
123 13 : mt_export_add(const char *str, GEN val)
124 : {
125 13 : if (pari_mt)
126 0 : pari_err(e_MISC,"export() not allowed during parallel sections");
127 13 : export_add(str, val);
128 13 : }
129 :
130 : void
131 8 : mt_export_del(const char *str)
132 : {
133 8 : if (pari_mt)
134 0 : pari_err(e_MISC,"unexport() not allowed during parallel sections");
135 8 : export_del(str);
136 8 : }
137 :
138 1 : void mt_broadcast(GEN code) {(void) code;}
139 :
140 269 : void pari_mt_init(void)
141 : {
142 269 : pari_mt = NULL;
143 : #ifdef _SC_NPROCESSORS_CONF
144 269 : if (!pari_mt_nbthreads) pari_mt_nbthreads = sysconf(_SC_NPROCESSORS_CONF);
145 : #elif defined(_WIN32)
146 : if (!pari_mt_nbthreads) pari_mt_nbthreads = win32_nbthreads();
147 : #else
148 : pari_mt_nbthreads = 1;
149 : #endif
150 269 : }
151 :
152 269 : void pari_mt_close(void) { }
153 :
154 : static void
155 376736 : mt_queue_cleanup(void *arg)
156 : {
157 : (void) arg;
158 376736 : pari_thread_close();
159 376736 : }
160 :
161 : static void
162 2 : mt_queue_unlock(void *arg)
163 2 : { pthread_mutex_unlock((pthread_mutex_t*) arg); }
164 :
165 : static void*
166 376736 : mt_queue_run(void *arg)
167 : {
168 376736 : GEN args = pari_thread_start((struct pari_thread*) arg);
169 376736 : pari_sp av = avma;
170 376736 : struct mt_queue *mq = (struct mt_queue *) args;
171 376736 : mt_thread_no = mq->no;
172 376736 : pthread_cleanup_push(mt_queue_cleanup,NULL);
173 376736 : LOCK(mq->pmut)
174 : {
175 376736 : mq->mainstack = pari_mainstack;
176 376736 : mq->avma = av;
177 376736 : pthread_cond_signal(mq->pcond);
178 376736 : } UNLOCK(mq->pmut);
179 : for(;;)
180 511089 : {
181 : GEN work, done;
182 887825 : LOCK(&mq->mut)
183 : {
184 887825 : pthread_cleanup_push(mt_queue_unlock, &mq->mut);
185 1751757 : while(!mq->input)
186 863934 : pthread_cond_wait(&mq->cond, &mq->mut);
187 887823 : pthread_cleanup_pop(0);
188 887823 : } UNLOCK(&mq->mut);
189 887823 : pari_mainstack = mq->mainstack;
190 887823 : set_avma(mq->avma);
191 887823 : work = mq->input;
192 887823 : if (typ(work)==t_ERROR && err_get_num(work)==e_STOP) break;
193 511148 : pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS,NULL);
194 511148 : done = closure_callgenvec(mq->worker,work);
195 511089 : pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED,NULL);
196 511089 : LOCK(mq->pmut)
197 : {
198 511089 : mq->mainstack = pari_mainstack;
199 511089 : mq->avma = av;
200 511089 : mq->input = NULL;
201 511089 : mq->output = done;
202 511089 : pthread_cond_signal(mq->pcond);
203 511089 : } UNLOCK(mq->pmut);
204 : }
205 376675 : pthread_cleanup_pop(1);
206 : #ifdef __GNUC__
207 : return NULL; /* LCOV_EXCL_LINE */
208 : #endif
209 : }
210 :
211 : static long
212 728081 : mt_queue_check(struct mt_pstate *mt)
213 : {
214 : long i;
215 5819403 : for(i=0; i<mt->n; i++)
216 : {
217 5602414 : struct mt_queue *mq = mt->mq+i;
218 5602414 : if (mq->output) return i;
219 : }
220 216989 : return -1;
221 : }
222 :
223 : static GEN
224 815211 : mtpthread_queue_get(struct mt_state *junk, long *workid, long *pending)
225 : {
226 815211 : struct mt_pstate *mt = pari_mt;
227 : struct mt_queue *mq;
228 815211 : GEN done = NULL;
229 : long last;
230 : (void) junk;
231 815211 : if (mt->nbint<mt->n)
232 : {
233 304117 : mt->last = mt->nbint;
234 304117 : *pending = mt->pending;
235 304117 : return NULL;
236 : }
237 511094 : BLOCK_SIGINT_START
238 511094 : LOCK(&mt->pmut)
239 : {
240 728081 : while ((last = mt_queue_check(mt)) < 0)
241 : {
242 216989 : pthread_cond_wait(&mt->pcond, &mt->pmut);
243 216989 : if (PARI_SIGINT_pending)
244 : {
245 2 : int sig = PARI_SIGINT_pending;
246 2 : PARI_SIGINT_pending = 0;
247 2 : pthread_mutex_unlock(&mt->pmut);
248 2 : PARI_SIGINT_block = 0;
249 2 : raise(sig);
250 0 : PARI_SIGINT_block = 1;
251 0 : pthread_mutex_lock(&mt->pmut);
252 : }
253 : }
254 511092 : } UNLOCK(&mt->pmut);
255 511092 : BLOCK_SIGINT_END
256 511092 : mq = mt->mq+last;
257 511092 : done = gcopy(mq->output);
258 511092 : mq->output = NULL;
259 511092 : if (workid) *workid = mq->workid;
260 511092 : if (typ(done) == t_ERROR)
261 : {
262 5 : if (err_get_num(done)==e_STACK)
263 0 : pari_err(e_STACKTHREAD);
264 : else
265 5 : pari_err(0,done);
266 : }
267 511087 : mt->last = last;
268 511087 : mt->pending--;
269 511087 : *pending = mt->pending;
270 511087 : return done;
271 : }
272 :
273 : static void
274 815211 : mtpthread_queue_submit(struct mt_state *junk, long workid, GEN work)
275 : {
276 815211 : struct mt_pstate *mt = pari_mt;
277 815211 : struct mt_queue *mq = mt->mq+mt->last;
278 : (void) junk;
279 815211 : if (!work) { mt->nbint=mt->n; return; }
280 511148 : BLOCK_SIGINT_START
281 511148 : if (mt->nbint<mt->n)
282 : {
283 369210 : mt->nbint++;
284 369210 : LOCK(mq->pmut)
285 : {
286 446835 : while(!mq->avma)
287 77625 : pthread_cond_wait(mq->pcond, mq->pmut);
288 369210 : } UNLOCK(mq->pmut);
289 : }
290 511148 : LOCK(&mq->mut)
291 : {
292 511148 : pari_sp av = avma;
293 511148 : struct pari_mainstack *st = pari_mainstack;
294 511148 : mq->output = NULL;
295 511148 : mq->workid = workid;
296 511148 : pari_mainstack = mq->mainstack;
297 511148 : set_avma(mq->avma);
298 511148 : mq->input = gcopy(work);
299 511148 : mq->avma = avma;
300 511148 : mq->mainstack = pari_mainstack;
301 511148 : pari_mainstack = st;
302 511148 : set_avma(av);
303 511148 : pthread_cond_signal(&mq->cond);
304 511148 : } UNLOCK(&mq->mut);
305 511148 : mt->pending++;
306 511148 : BLOCK_SIGINT_END
307 : }
308 :
309 : static void
310 65882 : mtpthread_queue_cleanup(struct mt_pstate *mt)
311 : {
312 : long i;
313 65882 : if (DEBUGLEVEL) pari_warn(warner,"stopping %ld threads", mt->n);
314 65882 : BLOCK_SIGINT_START
315 65882 : pari_mt = NULL;
316 442618 : for (i=0;i<mt->n;i++)
317 : {
318 376736 : struct mt_queue *mq = mt->mq+i;
319 376736 : pthread_cond_destroy(&mq->cond);
320 376736 : pthread_mutex_destroy(&mq->mut);
321 376736 : pari_thread_free(&mt->pth[i]);
322 : }
323 65882 : pthread_cond_destroy(&mt->pcond);
324 65882 : pthread_mutex_destroy(&mt->pmut);
325 65882 : BLOCK_SIGINT_END
326 65882 : pari_free(mt->mq);
327 65882 : pari_free(mt->pth);
328 65882 : pari_free(mt->th);
329 65882 : pari_free(mt);
330 65882 : }
331 :
332 : static void
333 65875 : mtpthread_queue_end(void)
334 : {
335 65875 : const long err_e_STOP[] = { evaltyp(t_ERROR) | _evallg(2), e_STOP};
336 65875 : struct mt_pstate *mt = pari_mt;
337 : long i;
338 65875 : BLOCK_SIGINT_START
339 442550 : for (i=0; i<mt->n; i++)
340 : {
341 376675 : struct mt_queue *mq = mt->mq+i;
342 376675 : LOCK(&mq->mut)
343 : {
344 376675 : mq->output = NULL;
345 376675 : mq->workid = 0;
346 376675 : mq->input = (GEN) err_e_STOP;
347 376675 : pthread_cond_signal(&mq->cond);
348 376675 : } UNLOCK(&mq->mut);
349 : }
350 442550 : for (i=0; i<mt->n; i++)
351 376675 : pthread_join(mt->th[i],NULL);
352 65875 : mtpthread_queue_cleanup(mt);
353 65875 : BLOCK_SIGINT_END
354 65875 : }
355 :
356 : void
357 7 : mt_queue_reset(void)
358 : {
359 7 : struct mt_pstate *mt = pari_mt;
360 : long i;
361 7 : BLOCK_SIGINT_START
362 68 : for (i=0; i<mt->n; i++)
363 61 : pthread_cancel(mt->th[i]);
364 68 : for (i=0; i<mt->n; i++)
365 61 : pthread_join(mt->th[i],NULL);
366 7 : mtpthread_queue_cleanup(mt);
367 7 : BLOCK_SIGINT_END
368 7 : }
369 :
370 : static long
371 65882 : closure_has_clone(GEN fun)
372 : {
373 65882 : if (isclone(fun)) return 1;
374 65876 : if (lg(fun) >= 8)
375 : {
376 65370 : GEN f = closure_get_frame(fun);
377 65370 : long i, l = lg(f);
378 242007 : for (i = 1; i < l; i++)
379 178297 : if (isclone(gel(f,i))) return 1;
380 : }
381 64216 : return 0;
382 : }
383 :
384 : void
385 152873 : mt_queue_start_lim(struct pari_mt *pt, GEN worker, long lim)
386 : {
387 152873 : if (lim==0) lim = pari_mt_nbthreads;
388 152411 : else lim = minss(pari_mt_nbthreads, lim);
389 152873 : if (mt_thread_no >= 0)
390 50428 : mtsequential_queue_start(pt, worker);
391 102445 : else if (pari_mt || lim <= 1)
392 36563 : mtsingle_queue_start(pt, worker);
393 : else
394 : {
395 : struct mt_pstate *mt =
396 65882 : (struct mt_pstate*) pari_malloc(sizeof(struct mt_pstate));
397 65882 : long mtparisize = GP_DATA->threadsize? GP_DATA->threadsize: pari_mainstack->rsize;
398 65882 : long mtparisizemax = GP_DATA->threadsizemax;
399 : long i;
400 65882 : if (closure_has_clone(worker))
401 1666 : worker = gcopy(worker); /* to avoid clone_lock race */
402 65882 : mt->mq = (struct mt_queue *) pari_malloc(sizeof(*mt->mq)*lim);
403 65882 : mt->th = (pthread_t *) pari_malloc(sizeof(*mt->th)*lim);
404 65882 : mt->pth = (struct pari_thread *) pari_malloc(sizeof(*mt->pth)*lim);
405 65882 : mt->pending = 0;
406 65882 : mt->n = lim;
407 65882 : mt->nbint = 0;
408 65882 : mt->last = 0;
409 65882 : pthread_cond_init(&mt->pcond,NULL);
410 65882 : pthread_mutex_init(&mt->pmut,NULL);
411 442618 : for (i=0;i<lim;i++)
412 : {
413 376736 : struct mt_queue *mq = mt->mq+i;
414 376736 : mq->no = i;
415 376736 : mq->avma = 0;
416 376736 : mq->mainstack = NULL;
417 376736 : mq->worker = worker;
418 376736 : mq->input = NULL;
419 376736 : mq->output = NULL;
420 376736 : mq->pcond = &mt->pcond;
421 376736 : mq->pmut = &mt->pmut;
422 376736 : pthread_cond_init(&mq->cond,NULL);
423 376736 : pthread_mutex_init(&mq->mut,NULL);
424 376736 : if (mtparisizemax)
425 0 : pari_thread_valloc(&mt->pth[i],mtparisize,mtparisizemax,(GEN)mq);
426 : else
427 376736 : pari_thread_alloc(&mt->pth[i],mtparisize,(GEN)mq);
428 : }
429 65882 : if (DEBUGLEVEL) pari_warn(warner,"starting %ld threads", lim);
430 65882 : BLOCK_SIGINT_START
431 : {
432 : #ifdef HAS_PTHREAD_SIGMASK
433 : sigset_t set, oldset;
434 65882 : sigfillset(&set);
435 65882 : pthread_sigmask(SIG_SETMASK, &set, &oldset);
436 : #endif
437 442618 : for (i = 0; i < lim; i++)
438 376736 : pthread_create(&mt->th[i], NULL, &mt_queue_run, (void*)&mt->pth[i]);
439 : #ifdef HAS_PTHREAD_SIGMASK
440 65882 : pthread_sigmask(SIG_SETMASK, &oldset, NULL);
441 : #endif
442 65882 : pari_mt = mt;
443 : }
444 65882 : BLOCK_SIGINT_END
445 65882 : pt->get=&mtpthread_queue_get;
446 65882 : pt->submit=&mtpthread_queue_submit;
447 65882 : pt->end=&mtpthread_queue_end;
448 : }
449 152873 : }
|
