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