ext/Thread/Thread.xs

   1 #include "EXTERN.h"
   2 #include "perl.h"
   3 #include "XSUB.h"
   4
   5 /* Magic signature for Thread's mg_private is "Th" */
   6 #define Thread_MAGIC_SIGNATURE 0x5468
   7
   8 #ifdef __cplusplus
   9 #ifdef I_UNISTD
  10 #include <unistd.h>
  11 #endif
  12 #endif
  13 #include <fcntl.h>
  14
  15 static int sig_pipe[2];
  16
  17 #ifndef THREAD_RET_TYPE
  18 #define THREAD_RET_TYPE void *
  19 #define THREAD_RET_CAST(x) ((THREAD_RET_TYPE) x)
  20 #endif
  21
  22 static void
  23 remove_thread(struct perl_thread *t)
  24 {
  25 #ifdef USE_THREADS
  26     DEBUG_L(WITH_THR(PerlIO_printf(PerlIO_stderr(),
  27                                    "%p: remove_thread %p\n", thr, t)));
  28     MUTEX_LOCK(&threads_mutex);
  29     MUTEX_DESTROY(&t->mutex);
  30     nthreads--;
  31     t->prev->next = t->next;
  32     t->next->prev = t->prev;
  33     COND_BROADCAST(&nthreads_cond);
  34     MUTEX_UNLOCK(&threads_mutex);
  35 #endif
  36 }
  37
  38 static THREAD_RET_TYPE
  39 threadstart(void *arg)
  40 {
  41 #ifdef USE_THREADS
  42 #ifdef FAKE_THREADS
  43     Thread savethread = thr;
  44     LOGOP myop;
  45     dSP;
  46     I32 oldscope = scopestack_ix;
  47     I32 retval;
  48     AV *av;
  49     int i;
  50
  51     DEBUG_L(PerlIO_printf(PerlIO_stderr(), "new thread %p starting at %s\n",
  52                           thr, SvPEEK(TOPs)));
  53     thr = (Thread) arg;
  54     savemark = TOPMARK;
  55     thr->prev = thr->prev_run = savethread;
  56     thr->next = savethread->next;
  57     thr->next_run = savethread->next_run;
  58     savethread->next = savethread->next_run = thr;
  59     thr->wait_queue = 0;
  60     thr->private = 0;
  61
  62     /* Now duplicate most of perl_call_sv but with a few twists */
  63     op = (OP*)&myop;
  64     Zero(op, 1, LOGOP);
  65     myop.op_flags = OPf_STACKED;
  66     myop.op_next = Nullop;
  67     myop.op_flags |= OPf_KNOW;
  68     myop.op_flags |= OPf_WANT_LIST;
  69     op = pp_entersub(ARGS);
  70     DEBUG_L(if (!op)
  71             PerlIO_printf(PerlIO_stderr(), "thread starts at Nullop\n"));
  72     /*
  73      * When this thread is next scheduled, we start in the right
  74      * place. When the thread runs off the end of the sub, perl.c
  75      * handles things, using savemark to figure out how much of the
  76      * stack is the return value for any join.
  77      */
  78     thr = savethread;           /* back to the old thread */
  79     return 0;
  80 #else
  81     Thread thr = (Thread) arg;
  82     LOGOP myop;
  83     djSP;
  84     I32 oldmark = TOPMARK;
  85     I32 oldscope = scopestack_ix;
  86     I32 retval;
  87     SV *sv;
  88     AV *av = newAV();
  89     int i, ret;
  90     dJMPENV;
  91     DEBUG_L(PerlIO_printf(PerlIO_stderr(), "new thread %p waiting to start\n",
  92                           thr));
  93
  94     /* Don't call *anything* requiring dTHR until after SET_THR() */
  95     /*
  96      * Wait until our creator releases us. If we didn't do this, then
  97      * it would be potentially possible for out thread to carry on and
  98      * do stuff before our creator fills in our "self" field. For example,
  99      * if we went and created another thread which tried to JOIN with us,
 100      * then we'd be in a mess.
 101      */
 102     MUTEX_LOCK(&thr->mutex);
 103     MUTEX_UNLOCK(&thr->mutex);
 104
 105     /*
 106      * It's safe to wait until now to set the thread-specific pointer
 107      * from our pthread_t structure to our struct perl_thread, since
 108      * we're the only thread who can get at it anyway.
 109      */
 110     SET_THR(thr);
 111
 112     /* Only now can we use SvPEEK (which calls sv_newmortal which does dTHR) */
 113     DEBUG_L(PerlIO_printf(PerlIO_stderr(), "new thread %p starting at %s\n",
 114                           thr, SvPEEK(TOPs)));
 115
 116     sv = POPs;
 117     PUTBACK;
 118     perl_call_sv(sv, G_ARRAY|G_EVAL);
 119     SPAGAIN;
 120     retval = SP - (stack_base + oldmark);
 121     SP = stack_base + oldmark + 1;
 122     if (SvCUR(thr->errsv)) {
 123         MUTEX_LOCK(&thr->mutex);
 124         thr->flags |= THRf_DID_DIE;
 125         MUTEX_UNLOCK(&thr->mutex);
 126         av_store(av, 0, &sv_no);
 127         av_store(av, 1, newSVsv(thr->errsv));
 128         DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p died: %s\n",
 129                               thr, SvPV(thr->errsv, na)));
 130     } else {
 131         DEBUG_L(STMT_START {
 132             for (i = 1; i <= retval; i++) {
 133                 PerlIO_printf(PerlIO_stderr(), "%p return[%d] = %s\n",
 134                                 thr, i, SvPEEK(SP[i - 1]));
 135             }
 136         } STMT_END);
 137         av_store(av, 0, &sv_yes);
 138         for (i = 1; i <= retval; i++, SP++)
 139             sv_setsv(*av_fetch(av, i, TRUE), SvREFCNT_inc(*SP));
 140     }
 141
 142   finishoff:
 143 #if 0
 144     /* removed for debug */
 145     SvREFCNT_dec(curstack);
 146 #endif
 147     SvREFCNT_dec(thr->cvcache);
 148     SvREFCNT_dec(thr->threadsv);
 149     SvREFCNT_dec(thr->specific);
 150     SvREFCNT_dec(thr->errsv);
 151     SvREFCNT_dec(thr->errhv);
 152
 153     /*Safefree(cxstack);*/
 154     while (curstackinfo->si_next)
 155         curstackinfo = curstackinfo->si_next;
 156     while (curstackinfo) {
 157         PERL_SI *p = curstackinfo->si_prev;
 158         SvREFCNT_dec(curstackinfo->si_stack);
 159         Safefree(curstackinfo->si_cxstack);
 160         Safefree(curstackinfo);
 161         curstackinfo = p;
 162     }
 163     Safefree(markstack);
 164     Safefree(scopestack);
 165     Safefree(savestack);
 166     Safefree(retstack);
 167     Safefree(tmps_stack);
 168     Safefree(ofs);
 169
 170     SvREFCNT_dec(rs);
 171     SvREFCNT_dec(nrs);
 172     SvREFCNT_dec(statname);
 173     Safefree(screamfirst);
 174     Safefree(screamnext);
 175     Safefree(reg_start_tmp);
 176     SvREFCNT_dec(lastscream);
 177     /*SvREFCNT_dec(defoutgv);*/
 178
 179     MUTEX_LOCK(&thr->mutex);
 180     DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 181                           "%p: threadstart finishing: state is %u\n",
 182                           thr, ThrSTATE(thr)));
 183     switch (ThrSTATE(thr)) {
 184     case THRf_R_JOINABLE:
 185         ThrSETSTATE(thr, THRf_ZOMBIE);
 186         MUTEX_UNLOCK(&thr->mutex);
 187         DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 188                               "%p: R_JOINABLE thread finished\n", thr));
 189         break;
 190     case THRf_R_JOINED:
 191         ThrSETSTATE(thr, THRf_DEAD);
 192         MUTEX_UNLOCK(&thr->mutex);
 193         remove_thread(thr);
 194         DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 195                               "%p: R_JOINED thread finished\n", thr));
 196         break;
 197     case THRf_R_DETACHED:
 198         ThrSETSTATE(thr, THRf_DEAD);
 199         MUTEX_UNLOCK(&thr->mutex);
 200         SvREFCNT_dec(av);
 201         DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 202                               "%p: DETACHED thread finished\n", thr));
 203         remove_thread(thr);     /* This might trigger main thread to finish */
 204         break;
 205     default:
 206         MUTEX_UNLOCK(&thr->mutex);
 207         croak("panic: illegal state %u at end of threadstart", ThrSTATE(thr));
 208         /* NOTREACHED */
 209     }
 210     return THREAD_RET_CAST(av); /* Available for anyone to join with */
 211                                         /* us unless we're detached, in which */
 212                                         /* case noone sees the value anyway. */
 213 #endif
 214 #else
 215     return THREAD_RET_CAST(NULL);
 216 #endif
 217 }
 218
 219 static SV *
 220 newthread (SV *startsv, AV *initargs, char *classname)
 221 {
 222 #ifdef USE_THREADS
 223     dSP;
 224     Thread savethread;
 225     int i;
 226     SV *sv;
 227     int err;
 228 #ifndef THREAD_CREATE
 229     static pthread_attr_t attr;
 230     static int attr_inited = 0;
 231     sigset_t fullmask, oldmask;
 232 #endif
 233
 234     savethread = thr;
 235     thr = new_struct_thread(thr);
 236     SPAGAIN;
 237     DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 238                           "%p: newthread (%p), tid is %u, preparing stack\n",
 239                           savethread, thr, thr->tid));
 240     /* The following pushes the arg list and startsv onto the *new* stack */
 241     PUSHMARK(SP);
 242     /* Could easily speed up the following greatly */
 243     for (i = 0; i <= AvFILL(initargs); i++)
 244         XPUSHs(SvREFCNT_inc(*av_fetch(initargs, i, FALSE)));
 245     XPUSHs(SvREFCNT_inc(startsv));
 246     PUTBACK;
 247 #ifdef THREAD_CREATE
 248     err = THREAD_CREATE(thr, threadstart);
 249 #else
 250     /* On your marks... */
 251     MUTEX_LOCK(&thr->mutex);
 252     /* Get set...  */
 253     sigfillset(&fullmask);
 254     if (sigprocmask(SIG_SETMASK, &fullmask, &oldmask) == -1)
 255         croak("panic: sigprocmask");
 256     err = 0;
 257     if (!attr_inited) {
 258         attr_inited = 1;
 259         err = pthread_attr_init(&attr);
 260         if (err == 0)
 261             err = pthread_attr_setdetachstate(&attr, ATTR_JOINABLE);
 262     }
 263     if (err == 0)
 264         err = pthread_create(&thr->self, &attr, threadstart, (void*) thr);
 265     /* Go */
 266     MUTEX_UNLOCK(&thr->mutex);
 267 #endif
 268     if (err) {
 269         DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 270                               "%p: create of %p failed %d\n",
 271                               savethread, thr, err));
 272         /* Thread creation failed--clean up */
 273         SvREFCNT_dec(thr->cvcache);
 274         remove_thread(thr);
 275         MUTEX_DESTROY(&thr->mutex);
 276         for (i = 0; i <= AvFILL(initargs); i++)
 277             SvREFCNT_dec(*av_fetch(initargs, i, FALSE));
 278         SvREFCNT_dec(startsv);
 279         return NULL;
 280     }
 281 #ifdef THREAD_POST_CREATE
 282     THREAD_POST_CREATE(thr);
 283 #else
 284     if (sigprocmask(SIG_SETMASK, &oldmask, 0))
 285         croak("panic: sigprocmask");
 286 #endif
 287     sv = newSViv(thr->tid);
 288     sv_magic(sv, thr->oursv, '~', 0, 0);
 289     SvMAGIC(sv)->mg_private = Thread_MAGIC_SIGNATURE;
 290     return sv_bless(newRV_noinc(sv), gv_stashpv(classname, TRUE));
 291 #else
 292     croak("No threads in this perl");
 293     return &sv_undef;
 294 #endif
 295 }
 296
 297 static Signal_t handle_thread_signal _((int sig));
 298
 299 static Signal_t
 300 handle_thread_signal(int sig)
 301 {
 302     unsigned char c = (unsigned char) sig;
 303     /*
 304      * We're not really allowed to call fprintf in a signal handler
 305      * so don't be surprised if this isn't robust while debugging
 306      * with -DL.
 307      */
 308     DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 309             "handle_thread_signal: got signal %d\n", sig););
 310     write(sig_pipe[1], &c, 1);
 311 }
 312
 313 MODULE = Thread         PACKAGE = Thread
 314 PROTOTYPES: DISABLE
 315
 316 void
 317 new(classname, startsv, ...)
 318         char *          classname
 319         SV *            startsv
 320         AV *            av = av_make(items - 2, &ST(2));
 321     PPCODE:
 322         XPUSHs(sv_2mortal(newthread(startsv, av, classname)));
 323
 324 void
 325 join(t)
 326         Thread  t
 327         AV *    av = NO_INIT
 328         int     i = NO_INIT
 329     PPCODE:
 330 #ifdef USE_THREADS
 331         DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: joining %p (state %u)\n",
 332                               thr, t, ThrSTATE(t)););
 333         MUTEX_LOCK(&t->mutex);
 334         switch (ThrSTATE(t)) {
 335         case THRf_R_JOINABLE:
 336         case THRf_R_JOINED:
 337             ThrSETSTATE(t, THRf_R_JOINED);
 338             MUTEX_UNLOCK(&t->mutex);
 339             break;
 340         case THRf_ZOMBIE:
 341             ThrSETSTATE(t, THRf_DEAD);
 342             MUTEX_UNLOCK(&t->mutex);
 343             remove_thread(t);
 344             break;
 345         default:
 346             MUTEX_UNLOCK(&t->mutex);
 347             croak("can't join with thread");
 348             /* NOTREACHED */
 349         }
 350         JOIN(t, &av);
 351
 352         if (SvTRUE(*av_fetch(av, 0, FALSE))) {
 353             /* Could easily speed up the following if necessary */
 354             for (i = 1; i <= AvFILL(av); i++)
 355                 XPUSHs(sv_2mortal(*av_fetch(av, i, FALSE)));
 356         } else {
 357             char *mess = SvPV(*av_fetch(av, 1, FALSE), na);
 358             DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 359                                   "%p: join propagating die message: %s\n",
 360                                   thr, mess));
 361             croak(mess);
 362         }
 363 #endif
 364
 365 void
 366 detach(t)
 367         Thread  t
 368     CODE:
 369 #ifdef USE_THREADS
 370         DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: detaching %p (state %u)\n",
 371                               thr, t, ThrSTATE(t)););
 372         MUTEX_LOCK(&t->mutex);
 373         switch (ThrSTATE(t)) {
 374         case THRf_R_JOINABLE:
 375             ThrSETSTATE(t, THRf_R_DETACHED);
 376             /* fall through */
 377         case THRf_R_DETACHED:
 378             DETACH(t);
 379             MUTEX_UNLOCK(&t->mutex);
 380             break;
 381         case THRf_ZOMBIE:
 382             ThrSETSTATE(t, THRf_DEAD);
 383             DETACH(t);
 384             MUTEX_UNLOCK(&t->mutex);
 385             remove_thread(t);
 386             break;
 387         default:
 388             MUTEX_UNLOCK(&t->mutex);
 389             croak("can't detach thread");
 390             /* NOTREACHED */
 391         }
 392 #endif
 393
 394 void
 395 equal(t1, t2)
 396         Thread  t1
 397         Thread  t2
 398     PPCODE:
 399         PUSHs((t1 == t2) ? &sv_yes : &sv_no);
 400
 401 void
 402 flags(t)
 403         Thread  t
 404     PPCODE:
 405 #ifdef USE_THREADS
 406         PUSHs(sv_2mortal(newSViv(t->flags)));
 407 #endif
 408
 409 void
 410 self(classname)
 411         char *  classname
 412     PREINIT:
 413         SV *sv;
 414     PPCODE:
 415 #ifdef USE_THREADS
 416         sv = newSViv(thr->tid);
 417         sv_magic(sv, thr->oursv, '~', 0, 0);
 418         SvMAGIC(sv)->mg_private = Thread_MAGIC_SIGNATURE;
 419         PUSHs(sv_2mortal(sv_bless(newRV_noinc(sv),
 420                                   gv_stashpv(classname, TRUE))));
 421 #endif
 422
 423 U32
 424 tid(t)
 425         Thread  t
 426     CODE:
 427 #ifdef USE_THREADS
 428         MUTEX_LOCK(&t->mutex);
 429         RETVAL = t->tid;
 430         MUTEX_UNLOCK(&t->mutex);
 431 #else
 432         RETVAL = 0;
 433 #endif
 434     OUTPUT:
 435         RETVAL
 436
 437 void
 438 DESTROY(t)
 439         SV *    t
 440     PPCODE:
 441         PUSHs(&sv_yes);
 442
 443 void
 444 yield()
 445     CODE:
 446 {
 447 #ifdef USE_THREADS
 448         YIELD;
 449 #endif
 450 }
 451
 452 void
 453 cond_wait(sv)
 454         SV *    sv
 455         MAGIC * mg = NO_INIT
 456 CODE:
 457 #ifdef USE_THREADS
 458         if (SvROK(sv))
 459             sv = SvRV(sv);
 460
 461         mg = condpair_magic(sv);
 462         DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: cond_wait %p\n", thr, sv));
 463         MUTEX_LOCK(MgMUTEXP(mg));
 464         if (MgOWNER(mg) != thr) {
 465             MUTEX_UNLOCK(MgMUTEXP(mg));
 466             croak("cond_wait for lock that we don't own\n");
 467         }
 468         MgOWNER(mg) = 0;
 469         COND_WAIT(MgCONDP(mg), MgMUTEXP(mg));
 470         while (MgOWNER(mg))
 471             COND_WAIT(MgOWNERCONDP(mg), MgMUTEXP(mg));
 472         MgOWNER(mg) = thr;
 473         MUTEX_UNLOCK(MgMUTEXP(mg));
 474 #endif
 475
 476 void
 477 cond_signal(sv)
 478         SV *    sv
 479         MAGIC * mg = NO_INIT
 480 CODE:
 481 #ifdef USE_THREADS
 482         if (SvROK(sv))
 483             sv = SvRV(sv);
 484
 485         mg = condpair_magic(sv);
 486         DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: cond_signal %p\n",thr,sv));
 487         MUTEX_LOCK(MgMUTEXP(mg));
 488         if (MgOWNER(mg) != thr) {
 489             MUTEX_UNLOCK(MgMUTEXP(mg));
 490             croak("cond_signal for lock that we don't own\n");
 491         }
 492         COND_SIGNAL(MgCONDP(mg));
 493         MUTEX_UNLOCK(MgMUTEXP(mg));
 494 #endif
 495
 496 void
 497 cond_broadcast(sv)
 498         SV *    sv
 499         MAGIC * mg = NO_INIT
 500 CODE:
 501 #ifdef USE_THREADS
 502         if (SvROK(sv))
 503             sv = SvRV(sv);
 504
 505         mg = condpair_magic(sv);
 506         DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: cond_broadcast %p\n",
 507                               thr, sv));
 508         MUTEX_LOCK(MgMUTEXP(mg));
 509         if (MgOWNER(mg) != thr) {
 510             MUTEX_UNLOCK(MgMUTEXP(mg));
 511             croak("cond_broadcast for lock that we don't own\n");
 512         }
 513         COND_BROADCAST(MgCONDP(mg));
 514         MUTEX_UNLOCK(MgMUTEXP(mg));
 515 #endif
 516
 517 void
 518 list(classname)
 519         char *  classname
 520     PREINIT:
 521         Thread  t;
 522         AV *    av;
 523         SV **   svp;
 524         int     n = 0;
 525     PPCODE:
 526 #ifdef USE_THREADS
 527         av = newAV();
 528         /*
 529          * Iterate until we have enough dynamic storage for all threads.
 530          * We mustn't do any allocation while holding threads_mutex though.
 531          */
 532         MUTEX_LOCK(&threads_mutex);
 533         do {
 534             n = nthreads;
 535             MUTEX_UNLOCK(&threads_mutex);
 536             if (AvFILL(av) < n - 1) {
 537                 int i = AvFILL(av);
 538                 for (i = AvFILL(av); i < n - 1; i++) {
 539                     SV *sv = newSViv(0);        /* fill in tid later */
 540                     sv_magic(sv, 0, '~', 0, 0); /* fill in other magic later */
 541                     av_push(av, sv_bless(newRV_noinc(sv),
 542                                          gv_stashpv(classname, TRUE)));
 543
 544                 }
 545             }
 546             MUTEX_LOCK(&threads_mutex);
 547         } while (n < nthreads);
 548         n = nthreads;   /* Get the final correct value */
 549
 550         /*
 551          * At this point, there's enough room to fill in av.
 552          * Note that we are holding threads_mutex so the list
 553          * won't change out from under us but all the remaining
 554          * processing is "fast" (no blocking, malloc etc.)
 555          */
 556         t = thr;
 557         svp = AvARRAY(av);
 558         do {
 559             SV *sv = (SV*)SvRV(*svp);
 560             sv_setiv(sv, t->tid);
 561             SvMAGIC(sv)->mg_obj = SvREFCNT_inc(t->oursv);
 562             SvMAGIC(sv)->mg_flags |= MGf_REFCOUNTED;
 563             SvMAGIC(sv)->mg_private = Thread_MAGIC_SIGNATURE;
 564             t = t->next;
 565             svp++;
 566         } while (t != thr);
 567         /*  */
 568         MUTEX_UNLOCK(&threads_mutex);
 569         /* Truncate any unneeded slots in av */
 570         av_fill(av, n - 1);
 571         /* Finally, push all the new objects onto the stack and drop av */
 572         EXTEND(SP, n);
 573         for (svp = AvARRAY(av); n > 0; n--, svp++)
 574             PUSHs(*svp);
 575         (void)sv_2mortal((SV*)av);
 576 #endif
 577
 578
 579 MODULE = Thread         PACKAGE = Thread::Signal
 580
 581 void
 582 kill_sighandler_thread()
 583     PPCODE:
 584         write(sig_pipe[1], "\0", 1);
 585         PUSHs(&sv_yes);
 586
 587 void
 588 init_thread_signals()
 589     PPCODE:
 590         sighandlerp = handle_thread_signal;
 591         if (pipe(sig_pipe) == -1)
 592             XSRETURN_UNDEF;
 593         PUSHs(&sv_yes);
 594
 595 void
 596 await_signal()
 597     PREINIT:
 598         unsigned char c;
 599         SSize_t ret;
 600     CODE:
 601         do {
 602             ret = read(sig_pipe[0], &c, 1);
 603         } while (ret == -1 && errno == EINTR);
 604         if (ret == -1)
 605             croak("panic: await_signal");
 606         ST(0) = sv_newmortal();
 607         if (ret)
 608             sv_setsv(ST(0), c ? psig_ptr[c] : &sv_no);
 609         DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 610                               "await_signal returning %s\n", SvPEEK(ST(0))););
 611
 612 MODULE = Thread         PACKAGE = Thread::Specific
 613
 614 void
 615 data(classname = "Thread::Specific")
 616         char *  classname
 617     PPCODE:
 618 #ifdef USE_THREADS
 619         if (AvFILL(thr->specific) == -1) {
 620             GV *gv = gv_fetchpv("Thread::Specific::FIELDS", TRUE, SVt_PVHV);
 621             av_store(thr->specific, 0, newRV((SV*)GvHV(gv)));
 622         }
 623         XPUSHs(sv_bless(newRV((SV*)thr->specific),gv_stashpv(classname,TRUE)));
 624 #endif