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