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;
 110             thread->prev->next = thread->next;
 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             dTHXa(thread->interp);
 127             PERL_SET_CONTEXT(thread->interp);
 128             SvREFCNT_dec(thread->params);
 129             thread->params = Nullsv;
 130             perl_destruct(thread->interp);
 131             perl_free(thread->interp);
 132             thread->interp = NULL;
 133         }
 134         PERL_SET_CONTEXT(aTHX);
 135         MUTEX_UNLOCK(&thread->mutex);
 136         MUTEX_DESTROY(&thread->mutex);
 137         PerlMemShared_free(thread);
 138 }
 139
 140 int
 141 Perl_ithread_hook(pTHX)
 142 {
 143     int veto_cleanup = 0;
 144     MUTEX_LOCK(&create_destruct_mutex);
 145     if (aTHX == PL_curinterp && active_threads != 1) {
 146         Perl_warn(aTHX_ "A thread exited while %" IVdf " other threads were still running",
 147                                                 (IV)active_threads);
 148         veto_cleanup = 1;
 149     }
 150     MUTEX_UNLOCK(&create_destruct_mutex);
 151     return veto_cleanup;
 152 }
 153
 154 void
 155 Perl_ithread_detach(pTHX_ ithread *thread)
 156 {
 157     MUTEX_LOCK(&thread->mutex);
 158     if (!(thread->state & (PERL_ITHR_DETACHED|PERL_ITHR_JOINED))) {
 159         thread->state |= PERL_ITHR_DETACHED;
 160 #ifdef WIN32
 161         CloseHandle(thread->handle);
 162         thread->handle = 0;
 163 #else
 164         PERL_THREAD_DETACH(thread->thr);
 165 #endif
 166     }
 167     if ((thread->state & PERL_ITHR_FINISHED) &&
 168         (thread->state & PERL_ITHR_DETACHED)) {
 169         MUTEX_UNLOCK(&thread->mutex);
 170         Perl_ithread_destruct(aTHX_ thread, "detach");
 171     }
 172     else {
 173         MUTEX_UNLOCK(&thread->mutex);
 174     }
 175 }
 176
 177 /* MAGIC (in mg.h sense) hooks */
 178
 179 int
 180 ithread_mg_get(pTHX_ SV *sv, MAGIC *mg)
 181 {
 182     ithread *thread = (ithread *) mg->mg_ptr;
 183     SvIVX(sv) = PTR2IV(thread);
 184     SvIOK_on(sv);
 185     return 0;
 186 }
 187
 188 int
 189 ithread_mg_free(pTHX_ SV *sv, MAGIC *mg)
 190 {
 191     ithread *thread = (ithread *) mg->mg_ptr;
 192     MUTEX_LOCK(&thread->mutex);
 193     thread->count--;
 194     if (thread->count == 0) {
 195        if(thread->state & PERL_ITHR_FINISHED &&
 196           (thread->state & PERL_ITHR_DETACHED ||
 197            thread->state & PERL_ITHR_JOINED))
 198        {
 199             MUTEX_UNLOCK(&thread->mutex);
 200             Perl_ithread_destruct(aTHX_ thread, "no reference");
 201        }
 202        else {
 203             MUTEX_UNLOCK(&thread->mutex);
 204        }
 205     }
 206     else {
 207         MUTEX_UNLOCK(&thread->mutex);
 208     }
 209     return 0;
 210 }
 211
 212 int
 213 ithread_mg_dup(pTHX_ MAGIC *mg, CLONE_PARAMS *param)
 214 {
 215     ithread *thread = (ithread *) mg->mg_ptr;
 216     MUTEX_LOCK(&thread->mutex);
 217     thread->count++;
 218     MUTEX_UNLOCK(&thread->mutex);
 219     return 0;
 220 }
 221
 222 MGVTBL ithread_vtbl = {
 223  ithread_mg_get,        /* get */
 224  0,                     /* set */
 225  0,                     /* len */
 226  0,                     /* clear */
 227  ithread_mg_free,       /* free */
 228  0,                     /* copy */
 229  ithread_mg_dup         /* dup */
 230 };
 231
 232
 233 /*
 234  *      Starts executing the thread. Needs to clean up memory a tad better.
 235  *      Passed as the C level function to run in the new thread
 236  */
 237
 238 #ifdef WIN32
 239 THREAD_RET_TYPE
 240 Perl_ithread_run(LPVOID arg) {
 241 #else
 242 void*
 243 Perl_ithread_run(void * arg) {
 244 #endif
 245         ithread* thread = (ithread*) arg;
 246         dTHXa(thread->interp);
 247         PERL_SET_CONTEXT(thread->interp);
 248         PERL_THREAD_SETSPECIFIC(self_key,thread);
 249
 250 #if 0
 251         /* Far from clear messing with ->thr child-side is a good idea */
 252         MUTEX_LOCK(&thread->mutex);
 253 #ifdef WIN32
 254         thread->thr = GetCurrentThreadId();
 255 #else
 256         thread->thr = pthread_self();
 257 #endif
 258         MUTEX_UNLOCK(&thread->mutex);
 259 #endif
 260
 261         PL_perl_destruct_level = 2;
 262
 263         {
 264                 AV* params = (AV*) SvRV(thread->params);
 265                 I32 len = av_len(params)+1;
 266                 int i;
 267                 dSP;
 268                 ENTER;
 269                 SAVETMPS;
 270                 PUSHMARK(SP);
 271                 for(i = 0; i < len; i++) {
 272                     XPUSHs(av_shift(params));
 273                 }
 274                 PUTBACK;
 275                 len = call_sv(thread->init_function, thread->gimme|G_EVAL);
 276                 SPAGAIN;
 277                 for (i=len-1; i >= 0; i--) {
 278                   SV *sv = POPs;
 279                   av_store(params, i, SvREFCNT_inc(sv));
 280                 }
 281                 PUTBACK;
 282                 if (SvTRUE(ERRSV)) {
 283                     Perl_warn(aTHX_ "thread failed to start: %" SVf, ERRSV);
 284                 }
 285                 FREETMPS;
 286                 LEAVE;
 287                 SvREFCNT_dec(thread->init_function);
 288         }
 289
 290         PerlIO_flush((PerlIO*)NULL);
 291         MUTEX_LOCK(&thread->mutex);
 292         thread->state |= PERL_ITHR_FINISHED;
 293
 294         if (thread->state & PERL_ITHR_DETACHED) {
 295                 MUTEX_UNLOCK(&thread->mutex);
 296                 Perl_ithread_destruct(aTHX_ thread, "detached finish");
 297         } else {
 298                 MUTEX_UNLOCK(&thread->mutex);
 299         }
 300         MUTEX_LOCK(&create_destruct_mutex);
 301         active_threads--;
 302         assert( active_threads >= 0 );
 303         MUTEX_UNLOCK(&create_destruct_mutex);
 304
 305 #ifdef WIN32
 306         return (DWORD)0;
 307 #else
 308         return 0;
 309 #endif
 310 }
 311
 312 SV *
 313 ithread_to_SV(pTHX_ SV *obj, ithread *thread, char *classname, bool inc)
 314 {
 315     SV *sv;
 316     MAGIC *mg;
 317     if (inc) {
 318         MUTEX_LOCK(&thread->mutex);
 319         thread->count++;
 320         MUTEX_UNLOCK(&thread->mutex);
 321     }
 322     if (!obj)
 323      obj = newSV(0);
 324     sv = newSVrv(obj,classname);
 325     sv_setiv(sv,PTR2IV(thread));
 326     mg = sv_magicext(sv,Nullsv,PERL_MAGIC_shared_scalar,&ithread_vtbl,(char *)thread,0);
 327     mg->mg_flags |= MGf_DUP;
 328     SvREADONLY_on(sv);
 329     return obj;
 330 }
 331
 332 ithread *
 333 SV_to_ithread(pTHX_ SV *sv)
 334 {
 335     ithread *thread;
 336     if (SvROK(sv))
 337      {
 338       thread = INT2PTR(ithread*, SvIV(SvRV(sv)));
 339      }
 340     else
 341      {
 342       PERL_THREAD_GETSPECIFIC(self_key,thread);
 343      }
 344     return thread;
 345 }
 346
 347 /*
 348  * iThread->create(); ( aka iThread->new() )
 349  * Called in context of parent thread
 350  */
 351
 352 SV *
 353 Perl_ithread_create(pTHX_ SV *obj, char* classname, SV* init_function, SV* params)
 354 {
 355         ithread*        thread;
 356         CLONE_PARAMS    clone_param;
 357
 358         MUTEX_LOCK(&create_destruct_mutex);
 359         thread = PerlMemShared_malloc(sizeof(ithread));
 360         Zero(thread,1,ithread);
 361         thread->next = threads;
 362         thread->prev = threads->prev;
 363         threads->prev = thread;
 364         thread->prev->next = thread;
 365         /* Set count to 1 immediately in case thread exits before
 366          * we return to caller !
 367          */
 368         thread->count = 1;
 369         MUTEX_INIT(&thread->mutex);
 370         thread->tid = tid_counter++;
 371         thread->gimme = GIMME_V;
 372
 373         /* "Clone" our interpreter into the thread's interpreter
 374          * This gives thread access to "static data" and code.
 375          */
 376
 377         PerlIO_flush((PerlIO*)NULL);
 378
 379 #ifdef WIN32
 380         thread->interp = perl_clone(aTHX, CLONEf_KEEP_PTR_TABLE | CLONEf_CLONE_HOST);
 381 #else
 382         thread->interp = perl_clone(aTHX, CLONEf_KEEP_PTR_TABLE);
 383 #endif
 384         /* perl_clone leaves us in new interpreter's context.
 385            As it is tricky to spot implcit aTHX create a new scope
 386            with aTHX matching the context for the duration of
 387            our work for new interpreter.
 388          */
 389         {
 390             dTHXa(thread->interp);
 391             /* Here we remove END blocks since they should only run
 392                in the thread they are created
 393             */
 394             SvREFCNT_dec(PL_endav);
 395             PL_endav = newAV();
 396             clone_param.flags = 0;
 397             thread->init_function = sv_dup(init_function, &clone_param);
 398             if (SvREFCNT(thread->init_function) == 0) {
 399                 SvREFCNT_inc(thread->init_function);
 400             }
 401
 402             thread->params = sv_dup(params, &clone_param);
 403             SvREFCNT_inc(thread->params);
 404             SvTEMP_off(thread->init_function);
 405             ptr_table_free(PL_ptr_table);
 406             PL_ptr_table = NULL;
 407             PL_exit_flags |= PERL_EXIT_DESTRUCT_END;
 408         }
 409
 410         PERL_SET_CONTEXT(aTHX);
 411
 412         /* Start the thread */
 413
 414 #ifdef WIN32
 415
 416         thread->handle = CreateThread(NULL, 0, Perl_ithread_run,
 417                         (LPVOID)thread, 0, &thread->thr);
 418
 419 #else
 420         {
 421           static pthread_attr_t attr;
 422           static int attr_inited = 0;
 423           static int attr_joinable = PTHREAD_CREATE_JOINABLE;
 424           if (!attr_inited) {
 425             attr_inited = 1;
 426             pthread_attr_init(&attr);
 427           }
 428 #  ifdef PTHREAD_ATTR_SETDETACHSTATE
 429             PTHREAD_ATTR_SETDETACHSTATE(&attr, attr_joinable);
 430 #  endif
 431 #  ifdef THREAD_CREATE_NEEDS_STACK
 432             if(pthread_attr_setstacksize(&attr, THREAD_CREATE_NEEDS_STACK))
 433               croak("panic: pthread_attr_setstacksize failed");
 434 #  endif
 435
 436 #ifdef OLD_PTHREADS_API
 437           pthread_create( &thread->thr, attr, Perl_ithread_run, (void *)thread);
 438 #else
 439           pthread_create( &thread->thr, &attr, Perl_ithread_run, (void *)thread);
 440 #endif
 441         }
 442 #endif
 443         known_threads++;
 444         active_threads++;
 445         MUTEX_UNLOCK(&create_destruct_mutex);
 446         sv_2mortal(params);
 447         return ithread_to_SV(aTHX_ obj, thread, classname, FALSE);
 448 }
 449
 450 SV*
 451 Perl_ithread_self (pTHX_ SV *obj, char* Class)
 452 {
 453     ithread *thread;
 454     PERL_THREAD_GETSPECIFIC(self_key,thread);
 455     return ithread_to_SV(aTHX_ obj, thread, Class, TRUE);
 456 }
 457
 458 /*
 459  * Joins the thread this code needs to take the returnvalue from the
 460  * call_sv and send it back
 461  */
 462
 463 void
 464 Perl_ithread_CLONE(pTHX_ SV *obj)
 465 {
 466  if (SvROK(obj))
 467   {
 468    ithread *thread = SV_to_ithread(aTHX_ obj);
 469   }
 470  else
 471   {
 472    Perl_warn(aTHX_ "CLONE %" SVf,obj);
 473   }
 474 }
 475
 476 AV*
 477 Perl_ithread_join(pTHX_ SV *obj)
 478 {
 479     ithread *thread = SV_to_ithread(aTHX_ obj);
 480     MUTEX_LOCK(&thread->mutex);
 481     if (thread->state & PERL_ITHR_DETACHED) {
 482         MUTEX_UNLOCK(&thread->mutex);
 483         Perl_croak(aTHX_ "Cannot join a detached thread");
 484     }
 485     else if (thread->state & PERL_ITHR_JOINED) {
 486         MUTEX_UNLOCK(&thread->mutex);
 487         Perl_croak(aTHX_ "Thread already joined");
 488     }
 489     else {
 490         AV* retparam;
 491 #ifdef WIN32
 492         DWORD waitcode;
 493 #else
 494         void *retval;
 495 #endif
 496         MUTEX_UNLOCK(&thread->mutex);
 497 #ifdef WIN32
 498         waitcode = WaitForSingleObject(thread->handle, INFINITE);
 499 #else
 500         pthread_join(thread->thr,&retval);
 501 #endif
 502         MUTEX_LOCK(&thread->mutex);
 503
 504         /* sv_dup over the args */
 505         {
 506           AV* params = (AV*) SvRV(thread->params);
 507           CLONE_PARAMS clone_params;
 508           clone_params.stashes = newAV();
 509           PL_ptr_table = ptr_table_new();
 510           retparam = (AV*) sv_dup((SV*)params, &clone_params);
 511           SvREFCNT_dec(clone_params.stashes);
 512           SvREFCNT_inc(retparam);
 513           ptr_table_free(PL_ptr_table);
 514           PL_ptr_table = NULL;
 515
 516         }
 517         /* We have finished with it */
 518         thread->state |= PERL_ITHR_JOINED;
 519         MUTEX_UNLOCK(&thread->mutex);
 520         sv_unmagic(SvRV(obj),PERL_MAGIC_shared_scalar);
 521         return retparam;
 522     }
 523     return (AV*)NULL;
 524 }
 525
 526 void
 527 Perl_ithread_DESTROY(pTHX_ SV *sv)
 528 {
 529     ithread *thread = SV_to_ithread(aTHX_ sv);
 530     sv_unmagic(SvRV(sv),PERL_MAGIC_shared_scalar);
 531 }
 532
 533 #endif /* USE_ITHREADS */
 534
 535 MODULE = threads                PACKAGE = threads       PREFIX = ithread_
 536 PROTOTYPES: DISABLE
 537
 538 #ifdef USE_ITHREADS
 539
 540 void
 541 ithread_new (classname, function_to_call, ...)
 542 char *  classname
 543 SV *    function_to_call
 544 CODE:
 545 {
 546     AV* params = newAV();
 547     if (items > 2) {
 548         int i;
 549         for(i = 2; i < items ; i++) {
 550             av_push(params, SvREFCNT_inc(ST(i)));
 551         }
 552     }
 553     ST(0) = sv_2mortal(Perl_ithread_create(aTHX_ Nullsv, classname, function_to_call, newRV_noinc((SV*) params)));
 554     XSRETURN(1);
 555 }
 556
 557 void
 558 ithread_list(char *classname)
 559 PPCODE:
 560 {
 561   ithread *curr_thread;
 562   MUTEX_LOCK(&create_destruct_mutex);
 563   curr_thread = threads;
 564   PUSHs( sv_2mortal(ithread_to_SV(aTHX_ NULL, curr_thread, classname, TRUE)));
 565   while(curr_thread) {
 566     curr_thread = curr_thread->next;
 567     if(curr_thread == threads)
 568       break;
 569     if(curr_thread->state & PERL_ITHR_DETACHED ||
 570        curr_thread->state & PERL_ITHR_JOINED)
 571       continue;
 572     PUSHs( sv_2mortal(ithread_to_SV(aTHX_ NULL, curr_thread, classname, TRUE)));
 573   }
 574   MUTEX_UNLOCK(&create_destruct_mutex);
 575 }
 576
 577
 578 void
 579 ithread_self(char *classname)
 580 CODE:
 581 {
 582         ST(0) = sv_2mortal(Perl_ithread_self(aTHX_ Nullsv,classname));
 583         XSRETURN(1);
 584 }
 585
 586 int
 587 ithread_tid(ithread *thread)
 588
 589 void
 590 ithread_join(SV *obj)
 591 PPCODE:
 592 {
 593   AV* params = Perl_ithread_join(aTHX_ obj);
 594   int i;
 595   I32 len = AvFILL(params);
 596   for (i = 0; i <= len; i++) {
 597     SV* tmp = av_shift(params);
 598     XPUSHs(tmp);
 599     sv_2mortal(tmp);
 600   }
 601   SvREFCNT_dec(params);
 602 }
 603
 604 void
 605 yield(...)
 606 CODE:
 607 {
 608     YIELD;
 609 }
 610
 611
 612 void
 613 ithread_detach(ithread *thread)
 614
 615 void
 616 ithread_DESTROY(SV *thread)
 617
 618 #endif /* USE_ITHREADS */
 619
 620 BOOT:
 621 {
 622 #ifdef USE_ITHREADS
 623         ithread* thread;
 624         PL_perl_destruct_level = 2;
 625         PERL_THREAD_ALLOC_SPECIFIC(self_key);
 626         MUTEX_INIT(&create_destruct_mutex);
 627         MUTEX_LOCK(&create_destruct_mutex);
 628         PL_threadhook = &Perl_ithread_hook;
 629         thread  = PerlMemShared_malloc(sizeof(ithread));
 630         Zero(thread,1,ithread);
 631         PL_perl_destruct_level = 2;
 632         MUTEX_INIT(&thread->mutex);
 633         threads = thread;
 634         thread->next = thread;
 635         thread->prev = thread;
 636         thread->interp = aTHX;
 637         thread->count  = 1;  /* imortal */
 638         thread->tid = tid_counter++;
 639         known_threads++;
 640         active_threads++;
 641         thread->state = 1;
 642 #ifdef WIN32
 643         thread->thr = GetCurrentThreadId();
 644 #else
 645         thread->thr = pthread_self();
 646 #endif
 647
 648         PERL_THREAD_SETSPECIFIC(self_key,thread);
 649         MUTEX_UNLOCK(&create_destruct_mutex);
 650 #endif /* USE_ITHREADS */
 651 }
 652