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 166730864 : mt_sigint_block(void)
57 : {
58 166730864 : if (mt_thread_no>=0)
59 23449784 : pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED,NULL);
60 168446418 : }
61 :
62 : void
63 167148646 : mt_sigint_unblock(void)
64 : {
65 167148646 : if (mt_thread_no>=0)
66 23833777 : pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS,NULL);
67 168419838 : }
68 :
69 : void
70 1565 : mt_err_recover(long er)
71 : {
72 : (void) er;
73 1565 : if (mt_thread_no>=0)
74 : {
75 8 : struct mt_pstate *mt = pari_mt;
76 8 : struct mt_queue *mq = mt->mq+mt_thread_no;
77 8 : GEN err = pari_err_last();
78 8 : err = err_get_num(err)==e_STACK ? err_e_STACK: bin_copy(copy_bin(err));
79 7 : LOCK(mq->pmut)
80 : {
81 7 : mq->output = err;
82 7 : pthread_cond_signal(mq->pcond);
83 7 : } UNLOCK(mq->pmut);
84 5 : pthread_exit((void*)1);
85 : }
86 1557 : else mtsingle_err_recover(er);
87 1557 : }
88 :
89 : void
90 0 : mt_sigint(void)
91 : {
92 0 : if (pari_mt) pthread_cond_broadcast(&pari_mt->pcond);
93 0 : }
94 :
95 : int
96 213378 : mt_is_parallel(void)
97 : {
98 213378 : return !!pari_mt;
99 : }
100 :
101 : int
102 30444073 : mt_is_thread(void)
103 : {
104 30444073 : return mt_thread_no>=0 ? 1: mtsingle_is_thread();
105 : }
106 :
107 : long
108 95578 : mt_nbthreads(void)
109 : {
110 95578 : return pari_mt ? 1: pari_mt_nbthreads;
111 : }
112 :
113 : void
114 227547 : mt_thread_init(void) { mt_thread_no = 0; }
115 :
116 : void
117 13 : mt_export_add(const char *str, GEN val)
118 : {
119 13 : if (pari_mt)
120 0 : pari_err(e_MISC,"export() not allowed during parallel sections");
121 13 : export_add(str, val);
122 13 : }
123 :
124 : void
125 8 : mt_export_del(const char *str)
126 : {
127 8 : if (pari_mt)
128 0 : pari_err(e_MISC,"unexport() not allowed during parallel sections");
129 8 : export_del(str);
130 8 : }
131 :
132 1 : void mt_broadcast(GEN code) {(void) code;}
133 :
134 252 : void pari_mt_init(void)
135 : {
136 252 : pari_mt = NULL;
137 : #ifdef _SC_NPROCESSORS_CONF
138 252 : if (!pari_mt_nbthreads) pari_mt_nbthreads = sysconf(_SC_NPROCESSORS_CONF);
139 : #elif defined(_WIN32)
140 : if (!pari_mt_nbthreads) pari_mt_nbthreads = win32_nbthreads();
141 : #else
142 : pari_mt_nbthreads = 1;
143 : #endif
144 252 : }
145 :
146 252 : void pari_mt_close(void) { }
147 :
148 : static void
149 227749 : mt_queue_cleanup(void *arg)
150 : {
151 : (void) arg;
152 227749 : pari_thread_close();
153 226922 : }
154 :
155 : static void
156 227664 : mt_queue_unlock(void *arg)
157 227664 : { pthread_mutex_unlock((pthread_mutex_t*) arg); }
158 :
159 : static void*
160 228685 : mt_queue_run(void *arg)
161 : {
162 228685 : GEN args = pari_thread_start((struct pari_thread*) arg);
163 227532 : pari_sp av = avma;
164 227532 : struct mt_queue *mq = (struct mt_queue *) args;
165 227532 : mt_thread_no = mq->no;
166 227532 : pthread_cleanup_push(mt_queue_cleanup,NULL);
167 227643 : LOCK(mq->pmut)
168 : {
169 228786 : mq->mainstack = pari_mainstack;
170 228786 : mq->avma = av;
171 228786 : pthread_cond_signal(mq->pcond);
172 228786 : } UNLOCK(mq->pmut);
173 : for(;;)
174 283079 : {
175 : GEN work, done;
176 511843 : LOCK(&mq->mut)
177 : {
178 511812 : pthread_cleanup_push(mt_queue_unlock, &mq->mut);
179 775166 : while(!mq->input)
180 492216 : pthread_cond_wait(&mq->cond, &mq->mut);
181 282950 : pthread_cleanup_pop(0);
182 282888 : } UNLOCK(&mq->mut);
183 283061 : pari_mainstack = mq->mainstack;
184 283061 : set_avma(mq->avma);
185 282766 : work = mq->input;
186 282766 : pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS,NULL);
187 283070 : done = closure_callgenvec(mq->worker,work);
188 282022 : pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED,NULL);
189 282923 : LOCK(mq->pmut)
190 : {
191 283103 : mq->mainstack = pari_mainstack;
192 283103 : mq->avma = av;
193 283103 : mq->input = NULL;
194 283103 : mq->output = done;
195 283103 : pthread_cond_signal(mq->pcond);
196 283103 : } UNLOCK(mq->pmut);
197 : }
198 : pthread_cleanup_pop(1);
199 : #ifdef __GNUC__
200 : return NULL; /* LCOV_EXCL_LINE */
201 : #endif
202 : }
203 :
204 : static long
205 404966 : mt_queue_check(struct mt_pstate *mt)
206 : {
207 : long i;
208 4685899 : for(i=0; i<mt->n; i++)
209 : {
210 4564039 : struct mt_queue *mq = mt->mq+i;
211 4564039 : if (mq->output) return i;
212 : }
213 121860 : return -1;
214 : }
215 :
216 : static GEN
217 473885 : mtpthread_queue_get(struct mt_state *junk, long *workid, long *pending)
218 : {
219 473885 : struct mt_pstate *mt = pari_mt;
220 : struct mt_queue *mq;
221 473885 : GEN done = NULL;
222 : long last;
223 : (void) junk;
224 473885 : if (mt->nbint<mt->n)
225 : {
226 190777 : mt->last = mt->nbint;
227 190777 : *pending = mt->pending;
228 190777 : return NULL;
229 : }
230 283108 : BLOCK_SIGINT_START
231 283108 : LOCK(&mt->pmut)
232 : {
233 404966 : while ((last = mt_queue_check(mt)) < 0)
234 : {
235 121860 : pthread_cond_wait(&mt->pcond, &mt->pmut);
236 121860 : if (PARI_SIGINT_pending)
237 : {
238 2 : int sig = PARI_SIGINT_pending;
239 2 : PARI_SIGINT_pending = 0;
240 2 : pthread_mutex_unlock(&mt->pmut);
241 2 : PARI_SIGINT_block = 0;
242 2 : raise(sig);
243 0 : PARI_SIGINT_block = 1;
244 0 : pthread_mutex_lock(&mt->pmut);
245 : }
246 : }
247 283106 : } UNLOCK(&mt->pmut);
248 283106 : BLOCK_SIGINT_END
249 283106 : mq = mt->mq+last;
250 283106 : done = gcopy(mq->output);
251 283106 : mq->output = NULL;
252 283106 : if (workid) *workid = mq->workid;
253 283106 : if (typ(done) == t_ERROR)
254 : {
255 5 : if (err_get_num(done)==e_STACK)
256 0 : pari_err(e_STACKTHREAD);
257 : else
258 5 : pari_err(0,done);
259 : }
260 283101 : mt->last = last;
261 283101 : mt->pending--;
262 283101 : *pending = mt->pending;
263 283101 : return done;
264 : }
265 :
266 : static void
267 473885 : mtpthread_queue_submit(struct mt_state *junk, long workid, GEN work)
268 : {
269 473885 : struct mt_pstate *mt = pari_mt;
270 473885 : struct mt_queue *mq = mt->mq+mt->last;
271 : (void) junk;
272 473885 : if (!work) { mt->nbint=mt->n; return; }
273 283194 : BLOCK_SIGINT_START
274 283194 : if (mt->nbint<mt->n)
275 : {
276 228090 : mt->nbint++;
277 228090 : LOCK(mq->pmut)
278 : {
279 294632 : while(!mq->avma)
280 66542 : pthread_cond_wait(mq->pcond, mq->pmut);
281 228090 : } UNLOCK(mq->pmut);
282 : }
283 283194 : LOCK(&mq->mut)
284 : {
285 283194 : mq->output = NULL;
286 283194 : mq->workid = workid;
287 283194 : BLOCK_SIGINT_START
288 : {
289 283194 : pari_sp av = avma;
290 283194 : struct pari_mainstack *st = pari_mainstack;
291 283194 : pari_mainstack = mq->mainstack;
292 283194 : set_avma(mq->avma);
293 283194 : mq->input = gcopy(work);
294 283194 : mq->avma = avma;
295 283194 : mq->mainstack = pari_mainstack;
296 283194 : pari_mainstack = st;
297 283194 : set_avma(av);
298 : }
299 283194 : BLOCK_SIGINT_END
300 283194 : pthread_cond_signal(&mq->cond);
301 283194 : } UNLOCK(&mq->mut);
302 283194 : mt->pending++;
303 283194 : BLOCK_SIGINT_END
304 : }
305 :
306 : void
307 37709 : mt_queue_reset(void)
308 : {
309 37709 : struct mt_pstate *mt = pari_mt;
310 : long i;
311 37709 : BLOCK_SIGINT_START
312 266495 : for (i=0; i<mt->n; i++)
313 228786 : pthread_cancel(mt->th[i]);
314 266495 : for (i=0; i<mt->n; i++)
315 228786 : pthread_join(mt->th[i],NULL);
316 37709 : pari_mt = NULL;
317 37709 : BLOCK_SIGINT_END
318 37709 : if (DEBUGLEVEL) pari_warn(warner,"stopping %ld threads", mt->n);
319 266495 : for (i=0;i<mt->n;i++)
320 : {
321 228786 : struct mt_queue *mq = mt->mq+i;
322 228786 : pthread_cond_destroy(&mq->cond);
323 228786 : pthread_mutex_destroy(&mq->mut);
324 228786 : pari_thread_free(&mt->pth[i]);
325 : }
326 37709 : pari_free(mt->mq);
327 37709 : pari_free(mt->pth);
328 37709 : pari_free(mt->th);
329 37709 : pari_free(mt);
330 37709 : }
331 :
332 : static long
333 37709 : closure_has_clone(GEN fun)
334 : {
335 37709 : if (isclone(fun)) return 1;
336 37707 : if (lg(fun) >= 8)
337 : {
338 37152 : GEN f = closure_get_frame(fun);
339 37152 : long i, l = lg(f);
340 146453 : for (i = 1; i < l; i++)
341 110206 : if (isclone(gel(f,i))) return 1;
342 : }
343 36802 : return 0;
344 : }
345 :
346 : void
347 49003 : mt_queue_start_lim(struct pari_mt *pt, GEN worker, long lim)
348 : {
349 49003 : if (lim==0) lim = pari_mt_nbthreads;
350 48984 : else lim = minss(pari_mt_nbthreads, lim);
351 49003 : if (mt_thread_no >= 0)
352 4225 : mtsequential_queue_start(pt, worker);
353 44778 : else if (pari_mt || lim <= 1)
354 7069 : mtsingle_queue_start(pt, worker);
355 : else
356 : {
357 : struct mt_pstate *mt =
358 37709 : (struct mt_pstate*) pari_malloc(sizeof(struct mt_pstate));
359 37709 : long mtparisize = GP_DATA->threadsize? GP_DATA->threadsize: pari_mainstack->rsize;
360 37709 : long mtparisizemax = GP_DATA->threadsizemax;
361 : long i;
362 37709 : if (closure_has_clone(worker))
363 907 : worker = gcopy(worker); /* to avoid clone_lock race */
364 37709 : mt->mq = (struct mt_queue *) pari_malloc(sizeof(*mt->mq)*lim);
365 37709 : mt->th = (pthread_t *) pari_malloc(sizeof(*mt->th)*lim);
366 37709 : mt->pth = (struct pari_thread *) pari_malloc(sizeof(*mt->pth)*lim);
367 37709 : mt->pending = 0;
368 37709 : mt->n = lim;
369 37709 : mt->nbint = 0;
370 37709 : mt->last = 0;
371 37709 : pthread_cond_init(&mt->pcond,NULL);
372 37709 : pthread_mutex_init(&mt->pmut,NULL);
373 266495 : for (i=0;i<lim;i++)
374 : {
375 228786 : struct mt_queue *mq = mt->mq+i;
376 228786 : mq->no = i;
377 228786 : mq->avma = 0;
378 228786 : mq->mainstack = NULL;
379 228786 : mq->worker = worker;
380 228786 : mq->input = NULL;
381 228786 : mq->output = NULL;
382 228786 : mq->pcond = &mt->pcond;
383 228786 : mq->pmut = &mt->pmut;
384 228786 : pthread_cond_init(&mq->cond,NULL);
385 228786 : pthread_mutex_init(&mq->mut,NULL);
386 228786 : if (mtparisizemax)
387 0 : pari_thread_valloc(&mt->pth[i],mtparisize,mtparisizemax,(GEN)mq);
388 : else
389 228786 : pari_thread_alloc(&mt->pth[i],mtparisize,(GEN)mq);
390 : }
391 37709 : if (DEBUGLEVEL) pari_warn(warner,"starting %ld threads", lim);
392 37709 : BLOCK_SIGINT_START
393 266495 : for (i=0;i<lim;i++)
394 228786 : pthread_create(&mt->th[i],NULL, &mt_queue_run, (void*)&mt->pth[i]);
395 37709 : pari_mt = mt;
396 37709 : BLOCK_SIGINT_END
397 37709 : pt->get=&mtpthread_queue_get;
398 37709 : pt->submit=&mtpthread_queue_submit;
399 37709 : pt->end=&mt_queue_reset;
400 : }
401 49003 : }
|
