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