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