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