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