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