ext/threads/threads.xs

   1 #define PERL_NO_GET_CONTEXT
   2 #include "EXTERN.h"
   3 #include "perl.h"
   4 #include "XSUB.h"
   5
   6 #ifdef USE_ITHREADS
   7
   8 #ifdef WIN32
   9 #include <windows.h>
  10 #include <win32thread.h>
  11 #define PERL_THREAD_SETSPECIFIC(k,v) TlsSetValue(k,v)
  12 #define PERL_THREAD_GETSPECIFIC(k,v) v = TlsGetValue(k)
  13 #define PERL_THREAD_ALLOC_SPECIFIC(k) \
  14 STMT_START {\
  15   if((k = TlsAlloc()) == TLS_OUT_OF_INDEXES) {\
  16     PerlIO_printf(PerlIO_stderr(),"panic threads.h: TlsAlloc");\
  17     exit(1);\
  18   }\
  19 } STMT_END
  20 #else
  21 #include <pthread.h>
  22 #include <thread.h>
  23
  24 #define PERL_THREAD_SETSPECIFIC(k,v) pthread_setspecific(k,v)
  25 #ifdef OLD_PTHREADS_API
  26 #define PERL_THREAD_DETACH(t) pthread_detach(&(t))
  27 #define PERL_THREAD_GETSPECIFIC(k,v) pthread_getspecific(k,&v)
  28 #define PERL_THREAD_ALLOC_SPECIFIC(k) STMT_START {\
  29   if(pthread_keycreate(&(k),0)) {\
  30     PerlIO_printf(PerlIO_stderr(), "panic threads.h: pthread_key_create");\
  31     exit(1);\
  32   }\
  33 } STMT_END
  34 #else
  35 #define PERL_THREAD_DETACH(t) pthread_detach((t))
  36 #define PERL_THREAD_GETSPECIFIC(k,v) v = pthread_getspecific(k)
  37 #define PERL_THREAD_ALLOC_SPECIFIC(k) STMT_START {\
  38   if(pthread_key_create(&(k),0)) {\
  39     PerlIO_printf(PerlIO_stderr(), "panic threads.h: pthread_key_create");\
  40     exit(1);\
  41   }\
  42 } STMT_END
  43 #endif
  44 #endif
  45
  46 /* Values for 'state' member */
  47 #define PERL_ITHR_JOINABLE              0
  48 #define PERL_ITHR_DETACHED              1
  49 #define PERL_ITHR_FINISHED              4
  50 #define PERL_ITHR_JOINED                2
  51
  52 typedef struct ithread_s {
  53     struct ithread_s *next;     /* next thread in the list */
  54     struct ithread_s *prev;     /* prev thread in the list */
  55     PerlInterpreter *interp;    /* The threads interpreter */
  56     I32 tid;                    /* threads module's thread id */
  57     perl_mutex mutex;           /* mutex for updating things in this struct */
  58     I32 count;                  /* how many SVs have a reference to us */
  59     signed char state;          /* are we detached ? */
  60     int gimme;                  /* Context of create */
  61     SV* init_function;          /* Code to run */
  62     SV* params;                 /* args to pass function */
  63 #ifdef WIN32
  64         DWORD   thr;            /* OS's idea if thread id */
  65         HANDLE handle;          /* OS's waitable handle */
  66 #else
  67         pthread_t thr;          /* OS's handle for the thread */
  68 #endif
  69 } ithread;
  70
  71 ithread *threads;
  72
  73 /* Macros to supply the aTHX_ in an embed.h like manner */
  74 #define ithread_join(thread)            Perl_ithread_join(aTHX_ thread)
  75 #define ithread_DESTROY(thread)         Perl_ithread_DESTROY(aTHX_ thread)
  76 #define ithread_CLONE(thread)           Perl_ithread_CLONE(aTHX_ thread)
  77 #define ithread_detach(thread)          Perl_ithread_detach(aTHX_ thread)
  78 #define ithread_tid(thread)             ((thread)->tid)
  79 #define ithread_yield(thread)           (YIELD);
  80
  81 static perl_mutex create_destruct_mutex;  /* protects the creation and destruction of threads*/
  82
  83 I32 tid_counter = 0;
  84 I32 known_threads = 0;
  85 I32 active_threads = 0;
  86 perl_key self_key;
  87
  88 /*
  89  *  Clear up after thread is done with
  90  */
  91 void
  92 Perl_ithread_destruct (pTHX_ ithread* thread, const char *why)
  93 {
  94         MUTEX_LOCK(&thread->mutex);
  95         if (!thread->next) {
  96             Perl_croak(aTHX_ "panic: destruct destroyed thread %p (%s)",thread, why);
  97         }
  98         if (thread->count != 0) {
  99                 MUTEX_UNLOCK(&thread->mutex);
 100                 return;
 101         }
 102         MUTEX_LOCK(&create_destruct_mutex);
 103         /* Remove from circular list of threads */
 104         if (thread->next == thread) {
 105             /* last one should never get here ? */
 106             threads = NULL;
 107         }
 108         else {
 109             thread->next->prev = thread->prev->next;
 110             thread->prev->next = thread->next->prev;
 111             if (threads == thread) {
 112                 threads = thread->next;
 113             }
 114             thread->next = NULL;
 115             thread->prev = NULL;
 116         }
 117         known_threads--;
 118         assert( known_threads >= 0 );
 119 #if 0
 120         Perl_warn(aTHX_ "destruct %d @ %p by %p now %d",
 121                   thread->tid,thread->interp,aTHX, known_threads);
 122 #endif
 123         MUTEX_UNLOCK(&create_destruct_mutex);
 124         /* Thread is now disowned */
 125         if (thread->interp) {
 126             PERL_SET_CONTEXT(thread->interp);
 127             perl_destruct(thread->interp);
 128             perl_free(thread->interp);
 129             thread->interp = NULL;
 130         }
 131         PERL_SET_CONTEXT(aTHX);
 132         MUTEX_UNLOCK(&thread->mutex);
 133 }
 134
 135 int
 136 Perl_ithread_hook(pTHX)
 137 {
 138     int veto_cleanup = 0;
 139     MUTEX_LOCK(&create_destruct_mutex);
 140     if (aTHX == PL_curinterp && active_threads != 1) {
 141         Perl_warn(aTHX_ "Cleanup skipped %d active threads", active_threads);
 142         veto_cleanup = 1;
 143     }
 144     MUTEX_UNLOCK(&create_destruct_mutex);
 145     return veto_cleanup;
 146 }
 147
 148 void
 149 Perl_ithread_detach(pTHX_ ithread *thread)
 150 {
 151     MUTEX_LOCK(&thread->mutex);
 152     if (!(thread->state & (PERL_ITHR_DETACHED|PERL_ITHR_JOINED))) {
 153         thread->state |= PERL_ITHR_DETACHED;
 154 #ifdef WIN32
 155         CloseHandle(thread->handle);
 156         thread->handle = 0;
 157 #else
 158         PERL_THREAD_DETACH(thread->thr);
 159 #endif
 160     }
 161     if ((thread->state & PERL_ITHR_FINISHED) &&
 162         (thread->state & PERL_ITHR_DETACHED)) {
 163         MUTEX_UNLOCK(&thread->mutex);
 164         Perl_ithread_destruct(aTHX_ thread, "detach");
 165     }
 166     else {
 167         MUTEX_UNLOCK(&thread->mutex);
 168     }
 169 }
 170
 171 /* MAGIC (in mg.h sense) hooks */
 172
 173 int
 174 ithread_mg_get(pTHX_ SV *sv, MAGIC *mg)
 175 {
 176     ithread *thread = (ithread *) mg->mg_ptr;
 177     SvIVX(sv) = PTR2IV(thread);
 178     SvIOK_on(sv);
 179     return 0;
 180 }
 181
 182 int
 183 ithread_mg_free(pTHX_ SV *sv, MAGIC *mg)
 184 {
 185     ithread *thread = (ithread *) mg->mg_ptr;
 186     MUTEX_LOCK(&thread->mutex);
 187     thread->count--;
 188     if (thread->count == 0) {
 189         if (!(thread->state & (PERL_ITHR_DETACHED|PERL_ITHR_JOINED))) {
 190             Perl_warn(aTHX_ "Implicit detach");
 191         }
 192         MUTEX_UNLOCK(&thread->mutex);
 193         Perl_ithread_detach(aTHX_ thread);
 194     }
 195     else {
 196         MUTEX_UNLOCK(&thread->mutex);
 197     }
 198     return 0;
 199 }
 200
 201 int
 202 ithread_mg_dup(pTHX_ MAGIC *mg, CLONE_PARAMS *param)
 203 {
 204     ithread *thread = (ithread *) mg->mg_ptr;
 205     MUTEX_LOCK(&thread->mutex);
 206     thread->count++;
 207     MUTEX_UNLOCK(&thread->mutex);
 208     return 0;
 209 }
 210
 211 MGVTBL ithread_vtbl = {
 212  ithread_mg_get,        /* get */
 213  0,                     /* set */
 214  0,                     /* len */
 215  0,                     /* clear */
 216  ithread_mg_free,       /* free */
 217  0,                     /* copy */
 218  ithread_mg_dup         /* dup */
 219 };
 220
 221
 222 /*
 223  *      Starts executing the thread. Needs to clean up memory a tad better.
 224  *      Passed as the C level function to run in the new thread
 225  */
 226
 227 #ifdef WIN32
 228 THREAD_RET_TYPE
 229 Perl_ithread_run(LPVOID arg) {
 230 #else
 231 void*
 232 Perl_ithread_run(void * arg) {
 233 #endif
 234         ithread* thread = (ithread*) arg;
 235         dTHXa(thread->interp);
 236         PERL_SET_CONTEXT(thread->interp);
 237         PERL_THREAD_SETSPECIFIC(self_key,thread);
 238
 239 #if 0
 240         /* Far from clear messing with ->thr child-side is a good idea */
 241         MUTEX_LOCK(&thread->mutex);
 242 #ifdef WIN32
 243         thread->thr = GetCurrentThreadId();
 244 #else
 245         thread->thr = pthread_self();
 246 #endif
 247         MUTEX_UNLOCK(&thread->mutex);
 248 #endif
 249
 250         PL_perl_destruct_level = 2;
 251
 252         {
 253                 AV* params = (AV*) SvRV(thread->params);
 254                 I32 len = av_len(params)+1;
 255                 int i;
 256                 dSP;
 257                 ENTER;
 258                 SAVETMPS;
 259                 PUSHMARK(SP);
 260                 for(i = 0; i < len; i++) {
 261                     XPUSHs(av_shift(params));
 262                 }
 263                 PUTBACK;
 264                 len = call_sv(thread->init_function, thread->gimme|G_EVAL);
 265                 SPAGAIN;
 266                 for (i=len-1; i >= 0; i--) {
 267                   SV *sv = POPs;
 268                   av_store(params, i, SvREFCNT_inc(sv));
 269                 }
 270                 PUTBACK;
 271                 if (SvTRUE(ERRSV)) {
 272                     Perl_warn(aTHX_ "Died:%_",ERRSV);
 273                 }
 274                 FREETMPS;
 275                 LEAVE;
 276                 SvREFCNT_dec(thread->init_function);
 277         }
 278
 279         PerlIO_flush((PerlIO*)NULL);
 280         MUTEX_LOCK(&create_destruct_mutex);
 281         active_threads--;
 282         assert( active_threads >= 0 );
 283         MUTEX_UNLOCK(&create_destruct_mutex);
 284         MUTEX_LOCK(&thread->mutex);
 285         thread->state |= PERL_ITHR_FINISHED;
 286
 287         if (thread->state & PERL_ITHR_DETACHED) {
 288                 MUTEX_UNLOCK(&thread->mutex);
 289                 SvREFCNT_dec(thread->params);
 290                 thread->params = Nullsv;
 291                 Perl_ithread_destruct(aTHX_ thread, "detached finish");
 292         } else {
 293                 MUTEX_UNLOCK(&thread->mutex);
 294         }
 295 #ifdef WIN32
 296         return (DWORD)0;
 297 #else
 298         return 0;
 299 #endif
 300 }
 301
 302 SV *
 303 ithread_to_SV(pTHX_ SV *obj, ithread *thread, char *classname, bool inc)
 304 {
 305     SV *sv;
 306     MAGIC *mg;
 307     if (inc) {
 308         MUTEX_LOCK(&thread->mutex);
 309         thread->count++;
 310         MUTEX_UNLOCK(&thread->mutex);
 311     }
 312     if (!obj)
 313      obj = newSV(0);
 314     sv = newSVrv(obj,classname);
 315     sv_setiv(sv,PTR2IV(thread));
 316     mg = sv_magicext(sv,Nullsv,PERL_MAGIC_shared_scalar,&ithread_vtbl,(char *)thread,0);
 317     mg->mg_flags |= MGf_DUP;
 318     SvREADONLY_on(sv);
 319     return obj;
 320 }
 321
 322 ithread *
 323 SV_to_ithread(pTHX_ SV *sv)
 324 {
 325     ithread *thread;
 326     if (SvROK(sv))
 327      {
 328       thread = INT2PTR(ithread*, SvIV(SvRV(sv)));
 329      }
 330     else
 331      {
 332       PERL_THREAD_GETSPECIFIC(self_key,thread);
 333      }
 334     return thread;
 335 }
 336
 337 /*
 338  * iThread->create(); ( aka iThread->new() )
 339  * Called in context of parent thread
 340  */
 341
 342 SV *
 343 Perl_ithread_create(pTHX_ SV *obj, char* classname, SV* init_function, SV* params)
 344 {
 345         ithread*        thread;
 346         CLONE_PARAMS    clone_param;
 347
 348         MUTEX_LOCK(&create_destruct_mutex);
 349         thread = PerlMemShared_malloc(sizeof(ithread));
 350         Zero(thread,1,ithread);
 351         thread->next = threads;
 352         thread->prev = threads->prev;
 353         thread->prev->next = thread;
 354         /* Set count to 1 immediately in case thread exits before
 355          * we return to caller !
 356          */
 357         thread->count = 1;
 358         MUTEX_INIT(&thread->mutex);
 359         thread->tid = tid_counter++;
 360         thread->gimme = GIMME_V;
 361         thread->state = (thread->gimme == G_VOID) ? 1 : 0;
 362
 363         /* "Clone" our interpreter into the thread's interpreter
 364          * This gives thread access to "static data" and code.
 365          */
 366
 367         PerlIO_flush((PerlIO*)NULL);
 368
 369 #ifdef WIN32
 370         thread->interp = perl_clone(aTHX, CLONEf_KEEP_PTR_TABLE | CLONEf_CLONE_HOST);
 371 #else
 372         thread->interp = perl_clone(aTHX, CLONEf_KEEP_PTR_TABLE);
 373 #endif
 374         /* perl_clone leaves us in new interpreter's context.
 375            As it is tricky to spot implcit aTHX create a new scope
 376            with aTHX matching the context for the duration of
 377            our work for new interpreter.
 378          */
 379         {
 380             dTHXa(thread->interp);
 381             /* Here we remove END blocks since they should only run
 382                in the thread they are created
 383             */
 384             SvREFCNT_dec(PL_endav);
 385             PL_endav = newAV();
 386             clone_param.flags = 0;
 387             thread->init_function = sv_dup(init_function, &clone_param);
 388             if (SvREFCNT(thread->init_function) == 0) {
 389                 SvREFCNT_inc(thread->init_function);
 390             }
 391
 392             thread->params = sv_dup(params, &clone_param);
 393             SvREFCNT_inc(thread->params);
 394             SvTEMP_off(thread->init_function);
 395             ptr_table_free(PL_ptr_table);
 396             PL_ptr_table = NULL;
 397         }
 398
 399         PERL_SET_CONTEXT(aTHX);
 400
 401         /* Start the thread */
 402
 403 #ifdef WIN32
 404
 405         thread->handle = CreateThread(NULL, 0, Perl_ithread_run,
 406                         (LPVOID)thread, 0, &thread->thr);
 407
 408 #else
 409         {
 410           static pthread_attr_t attr;
 411           static int attr_inited = 0;
 412           static int attr_joinable = PTHREAD_CREATE_JOINABLE;
 413           if (!attr_inited) {
 414             attr_inited = 1;
 415             pthread_attr_init(&attr);
 416           }
 417 #  ifdef PTHREAD_ATTR_SETDETACHSTATE
 418             PTHREAD_ATTR_SETDETACHSTATE(&attr, attr_joinable);
 419 #  endif
 420 #  ifdef THREAD_CREATE_NEEDS_STACK
 421             if(pthread_attr_setstacksize(&attr, THREAD_CREATE_NEEDS_STACK))
 422               croak("panic: pthread_attr_setstacksize failed");
 423 #  endif
 424
 425 #ifdef OLD_PTHREADS_API
 426           pthread_create( &thread->thr, attr, Perl_ithread_run, (void *)thread);
 427 #else
 428           pthread_create( &thread->thr, &attr, Perl_ithread_run, (void *)thread);
 429 #endif
 430         }
 431 #endif
 432         known_threads++;
 433         active_threads++;
 434         MUTEX_UNLOCK(&create_destruct_mutex);
 435         return ithread_to_SV(aTHX_ obj, thread, classname, FALSE);
 436 }
 437
 438 SV*
 439 Perl_ithread_self (pTHX_ SV *obj, char* Class)
 440 {
 441     ithread *thread;
 442     PERL_THREAD_GETSPECIFIC(self_key,thread);
 443     return ithread_to_SV(aTHX_ obj, thread, Class, TRUE);
 444 }
 445
 446 /*
 447  * Joins the thread this code needs to take the returnvalue from the
 448  * call_sv and send it back
 449  */
 450
 451 void
 452 Perl_ithread_CLONE(pTHX_ SV *obj)
 453 {
 454  if (SvROK(obj))
 455   {
 456    ithread *thread = SV_to_ithread(aTHX_ obj);
 457   }
 458  else
 459   {
 460    Perl_warn(aTHX_ "CLONE %_",obj);
 461   }
 462 }
 463
 464 AV*
 465 Perl_ithread_join(pTHX_ SV *obj)
 466 {
 467     ithread *thread = SV_to_ithread(aTHX_ obj);
 468     MUTEX_LOCK(&thread->mutex);
 469     if (thread->state & PERL_ITHR_DETACHED) {
 470         MUTEX_UNLOCK(&thread->mutex);
 471         Perl_croak(aTHX_ "Cannot join a detached thread");
 472     }
 473     else if (thread->state & PERL_ITHR_JOINED) {
 474         MUTEX_UNLOCK(&thread->mutex);
 475         Perl_croak(aTHX_ "Thread already joined");
 476     }
 477     else {
 478         AV* retparam;
 479 #ifdef WIN32
 480         DWORD waitcode;
 481 #else
 482         void *retval;
 483 #endif
 484         MUTEX_UNLOCK(&thread->mutex);
 485 #ifdef WIN32
 486         waitcode = WaitForSingleObject(thread->handle, INFINITE);
 487 #else
 488         pthread_join(thread->thr,&retval);
 489 #endif
 490         MUTEX_LOCK(&thread->mutex);
 491
 492         /* sv_dup over the args */
 493         {
 494           AV* params = (AV*) SvRV(thread->params);
 495           CLONE_PARAMS clone_params;
 496           clone_params.stashes = newAV();
 497           PL_ptr_table = ptr_table_new();
 498           retparam = (AV*) sv_dup((SV*)params, &clone_params);
 499           SvREFCNT_dec(clone_params.stashes);
 500           SvREFCNT_inc(retparam);
 501           ptr_table_free(PL_ptr_table);
 502           PL_ptr_table = NULL;
 503
 504         }
 505         /* We have finished with it */
 506         thread->state |= PERL_ITHR_JOINED;
 507         MUTEX_UNLOCK(&thread->mutex);
 508         sv_unmagic(SvRV(obj),PERL_MAGIC_shared_scalar);
 509         Perl_ithread_destruct(aTHX_ thread, "joined");
 510         return retparam;
 511     }
 512     return (AV*)NULL;
 513 }
 514
 515 void
 516 Perl_ithread_DESTROY(pTHX_ SV *sv)
 517 {
 518     ithread *thread = SV_to_ithread(aTHX_ sv);
 519     sv_unmagic(SvRV(sv),PERL_MAGIC_shared_scalar);
 520 }
 521
 522 #endif /* USE_ITHREADS */
 523
 524 MODULE = threads                PACKAGE = threads       PREFIX = ithread_
 525 PROTOTYPES: DISABLE
 526
 527 #ifdef USE_ITHREADS
 528
 529 void
 530 ithread_new (classname, function_to_call, ...)
 531 char *  classname
 532 SV *    function_to_call
 533 CODE:
 534 {
 535     AV* params = newAV();
 536     if (items > 2) {
 537         int i;
 538         for(i = 2; i < items ; i++) {
 539             av_push(params, ST(i));
 540         }
 541     }
 542     ST(0) = sv_2mortal(Perl_ithread_create(aTHX_ Nullsv, classname, function_to_call, newRV_noinc((SV*) params)));
 543     XSRETURN(1);
 544 }
 545
 546 void
 547 ithread_self(char *classname)
 548 CODE:
 549 {
 550         ST(0) = sv_2mortal(Perl_ithread_self(aTHX_ Nullsv,classname));
 551         XSRETURN(1);
 552 }
 553
 554 int
 555 ithread_tid(ithread *thread)
 556
 557 void
 558 ithread_join(SV *obj)
 559 PPCODE:
 560 {
 561   AV* params = Perl_ithread_join(aTHX_ obj);
 562   int i;
 563   I32 len = AvFILL(params);
 564   for (i = 0; i <= len; i++) {
 565     XPUSHs(av_shift(params));
 566   }
 567   SvREFCNT_dec(params);
 568 }
 569
 570 void
 571 ithread_yield(ithread *thread)
 572
 573 void
 574 ithread_detach(ithread *thread)
 575
 576 void
 577 ithread_DESTROY(SV *thread)
 578
 579 #endif /* USE_ITHREADS */
 580
 581 BOOT:
 582 {
 583 #ifdef USE_ITHREADS
 584         ithread* thread;
 585         PL_perl_destruct_level = 2;
 586         PERL_THREAD_ALLOC_SPECIFIC(self_key);
 587         MUTEX_INIT(&create_destruct_mutex);
 588         MUTEX_LOCK(&create_destruct_mutex);
 589         PL_threadhook = &Perl_ithread_hook;
 590         thread  = PerlMemShared_malloc(sizeof(ithread));
 591         Zero(thread,1,ithread);
 592         PL_perl_destruct_level = 2;
 593         MUTEX_INIT(&thread->mutex);
 594         threads = thread;
 595         thread->next = thread;
 596         thread->prev = thread;
 597         thread->interp = aTHX;
 598         thread->count  = 1;  /* imortal */
 599         thread->tid = tid_counter++;
 600         known_threads++;
 601         active_threads++;
 602         thread->state = 1;
 603 #ifdef WIN32
 604         thread->thr = GetCurrentThreadId();
 605 #else
 606         thread->thr = pthread_self();
 607 #endif
 608
 609         PERL_THREAD_SETSPECIFIC(self_key,thread);
 610         MUTEX_UNLOCK(&create_destruct_mutex);
 611 #endif /* USE_ITHREADS */
 612 }
 613