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