Thread.xs

   1 #include "EXTERN.h"
   2 #include "perl.h"
   3 #include "XSUB.h"
   4
   5 static void *
   6 threadstart(arg)
   7 void *arg;
   8 {
   9 #ifdef FAKE_THREADS
  10     Thread savethread = thr;
  11     LOGOP myop;
  12     dSP;
  13     I32 oldscope = scopestack_ix;
  14     I32 retval;
  15     AV *returnav = newAV();
  16     int i;
  17
  18     DEBUG_L(PerlIO_printf(PerlIO_stderr(), "new thread 0x%lx starting at %s\n",
  19                           (unsigned long) thr, SvPEEK(TOPs)));
  20     thr = (Thread) arg;
  21     savemark = TOPMARK;
  22     thr->prev = thr->prev_run = savethread;
  23     thr->next = savethread->next;
  24     thr->next_run = savethread->next_run;
  25     savethread->next = savethread->next_run = thr;
  26     thr->wait_queue = 0;
  27     thr->private = 0;
  28
  29     /* Now duplicate most of perl_call_sv but with a few twists */
  30     op = (OP*)&myop;
  31     Zero(op, 1, LOGOP);
  32     myop.op_flags = OPf_STACKED;
  33     myop.op_next = Nullop;
  34     myop.op_flags |= OPf_KNOW;
  35     myop.op_flags |= OPf_WANT_LIST;
  36     op = pp_entersub(ARGS);
  37     DEBUG_L(if (!op)
  38             PerlIO_printf(PerlIO_stderr(), "thread starts at Nullop\n"));
  39     /*
  40      * When this thread is next scheduled, we start in the right
  41      * place. When the thread runs off the end of the sub, perl.c
  42      * handles things, using savemark to figure out how much of the
  43      * stack is the return value for any join.
  44      */
  45     thr = savethread;           /* back to the old thread */
  46     return 0;
  47 #else
  48     Thread thr = (Thread) arg;
  49     LOGOP myop;
  50     dSP;
  51     I32 oldmark = TOPMARK;
  52     I32 oldscope = scopestack_ix;
  53     I32 retval;
  54     AV *returnav = newAV();
  55     int i;
  56     dJMPENV;
  57     int ret;
  58
  59     /* Don't call *anything* requiring dTHR until after pthread_setspecific */
  60     /*
  61      * Wait until our creator releases us. If we didn't do this, then
  62      * it would be potentially possible for out thread to carry on and
  63      * do stuff before our creator fills in our "self" field. For example,
  64      * if we went and created another thread which tried to pthread_join
  65      * with us, then we'd be in a mess.
  66      */
  67     MUTEX_LOCK(threadstart_mutexp);
  68     MUTEX_UNLOCK(threadstart_mutexp);
  69     MUTEX_DESTROY(threadstart_mutexp);  /* don't need it any more */
  70     Safefree(threadstart_mutexp);
  71
  72     /*
  73      * It's safe to wait until now to set the thread-specific pointer
  74      * from our pthread_t structure to our struct thread, since we're
  75      * the only thread who can get at it anyway.
  76      */
  77     if (pthread_setspecific(thr_key, (void *) thr))
  78         croak("panic: pthread_setspecific");
  79
  80     /* Only now can we use SvPEEK (which calls sv_newmortal which does dTHR) */
  81     DEBUG_L(PerlIO_printf(PerlIO_stderr(), "new thread 0x%lx starting at %s\n",
  82                           (unsigned long) thr, SvPEEK(TOPs)));
  83
  84     JMPENV_PUSH(ret);
  85     switch (ret) {
  86     case 3:
  87         PerlIO_printf(PerlIO_stderr(), "panic: threadstart\n");
  88         /* fall through */
  89     case 1:
  90         STATUS_ALL_FAILURE;
  91         /* fall through */
  92     case 2:
  93         /* my_exit() was called */
  94         while (scopestack_ix > oldscope)
  95             LEAVE;
  96         JMPENV_POP;
  97         av_store(returnav, 0, newSViv(statusvalue));
  98         goto finishoff;
  99     }
 100
 101     /* Now duplicate most of perl_call_sv but with a few twists */
 102     op = (OP*)&myop;
 103     Zero(op, 1, LOGOP);
 104     myop.op_flags = OPf_STACKED;
 105     myop.op_next = Nullop;
 106     myop.op_flags |= OPf_KNOW;
 107     myop.op_flags |= OPf_WANT_LIST;
 108     op = pp_entersub(ARGS);
 109     if (op)
 110         runops();
 111     SPAGAIN;
 112     retval = sp - (stack_base + oldmark);
 113     sp = stack_base + oldmark + 1;
 114     DEBUG_L(for (i = 1; i <= retval; i++)
 115                 PerlIO_printf(PerlIO_stderr(),
 116                               "%p returnav[%d] = %s\n",
 117                               thr, i, SvPEEK(sp[i - 1]));)
 118     av_store(returnav, 0, newSVpv("", 0));
 119     for (i = 1; i <= retval; i++, sp++)
 120         sv_setsv(*av_fetch(returnav, i, TRUE), SvREFCNT_inc(*sp));
 121
 122   finishoff:
 123 #if 0
 124     /* removed for debug */
 125     SvREFCNT_dec(curstack);
 126 #endif
 127     SvREFCNT_dec(cvcache);
 128     Safefree(markstack);
 129     Safefree(scopestack);
 130     Safefree(savestack);
 131     Safefree(retstack);
 132     Safefree(cxstack);
 133     Safefree(tmps_stack);
 134
 135     if (ThrSTATE(thr) == THR_DETACHED) {
 136         DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 137                               "%p detached...zapping returnav\n", thr));
 138         SvREFCNT_dec(returnav);
 139         ThrSETSTATE(thr, THR_DEAD);
 140     }
 141     DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p returning\n", thr));
 142     return (void *) returnav;   /* Available for anyone to join with us */
 143                                 /* unless we are detached in which case */
 144                                 /* noone will see the value anyway. */
 145 #endif
 146 }
 147
 148 Thread
 149 newthread(startsv, initargs)
 150 SV *startsv;
 151 AV *initargs;
 152 {
 153     dTHR;
 154     dSP;
 155     Thread savethread;
 156     int i;
 157
 158     savethread = thr;
 159     New(53, thr, 1, struct thread);
 160     /* If we don't zero these foostack pointers, init_stacks won't init them */
 161     markstack = 0;
 162     scopestack = 0;
 163     savestack = 0;
 164     retstack = 0;
 165     init_stacks(ARGS);
 166     curcop = savethread->Tcurcop;       /* XXX As good a guess as any? */
 167     SPAGAIN;
 168     defstash = savethread->Tdefstash;   /* XXX maybe these should */
 169     curstash = savethread->Tcurstash;   /* always be set to main? */
 170     /* top_env? */
 171     /* runlevel */
 172     cvcache = newHV();
 173     ThrSETSTATE(thr, THR_NORMAL);
 174
 175     /* The following pushes the arg list and startsv onto the *new* stack */
 176     PUSHMARK(sp);
 177     /* Could easily speed up the following greatly */
 178     for (i = 0; i <= AvFILL(initargs); i++)
 179         XPUSHs(SvREFCNT_inc(*av_fetch(initargs, i, FALSE)));
 180     XPUSHs(SvREFCNT_inc(startsv));
 181     PUTBACK;
 182
 183 #ifdef FAKE_THREADS
 184     threadstart(thr);
 185 #else
 186     New(53, threadstart_mutexp, 1, perl_mutex);
 187     /* On your marks... */
 188     MUTEX_INIT(threadstart_mutexp);
 189     MUTEX_LOCK(threadstart_mutexp);
 190     /* Get set...
 191      * Increment the global thread count. It is decremented
 192      * by the destructor for the thread specific key thr_key.
 193      */
 194     MUTEX_LOCK(&nthreads_mutex);
 195     nthreads++;
 196     MUTEX_UNLOCK(&nthreads_mutex);
 197     if (pthread_create(&self, NULL, threadstart, (void*) thr))
 198         return NULL;    /* XXX should clean up first */
 199     /* Go */
 200     MUTEX_UNLOCK(threadstart_mutexp);
 201 #endif
 202     return thr;
 203 }
 204
 205 MODULE = Thread         PACKAGE = Thread
 206
 207 Thread
 208 new(class, startsv, ...)
 209         SV *            class
 210         SV *            startsv
 211         AV *            av = av_make(items - 2, &ST(2));
 212     CODE:
 213         RETVAL = newthread(startsv, av);
 214     OUTPUT:
 215         RETVAL
 216
 217 void
 218 join(t)
 219         Thread  t
 220         AV *    av = NO_INIT
 221         int     i = NO_INIT
 222     PPCODE:
 223         DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 224                               "0x%lx: joining 0x%lx (state 0x%lx)\n",
 225                               (unsigned long)thr, (unsigned long)t,
 226                               (unsigned long)ThrSTATE(t)););
 227         if (ThrSTATE(t) == THR_DETACHED)
 228             croak("tried to join a detached thread");
 229         else if (ThrSTATE(t) == THR_JOINED)
 230             croak("tried to rejoin an already joined thread");
 231         else if (ThrSTATE(t) == THR_DEAD)
 232             croak("tried to join a dead thread");
 233
 234         if (pthread_join(t->Tself, (void **) &av))
 235             croak("pthread_join failed");
 236         ThrSETSTATE(t, THR_JOINED);
 237         /* Could easily speed up the following if necessary */
 238         for (i = 0; i <= AvFILL(av); i++)
 239             XPUSHs(sv_2mortal(*av_fetch(av, i, FALSE)));
 240
 241 void
 242 detach(t)
 243         Thread  t
 244     CODE:
 245         DEBUG_L(PerlIO_printf(PerlIO_stderr(),
 246                               "0x%lx: detaching 0x%lx (state 0x%lx)\n",
 247                               (unsigned long)thr, (unsigned long)t,
 248                               (unsigned long)ThrSTATE(t)););
 249         if (ThrSTATE(t) == THR_DETACHED)
 250             croak("tried to detach an already detached thread");
 251         else if (ThrSTATE(t) == THR_JOINED)
 252             croak("tried to detach an already joined thread");
 253         else if (ThrSTATE(t) == THR_DEAD)
 254             croak("tried to detach a dead thread");
 255         if (pthread_detach(t->Tself))
 256             croak("pthread_detach failed");
 257         ThrSETSTATE(t, THR_DETACHED);
 258
 259 void
 260 DESTROY(t)
 261         Thread  t
 262     CODE:
 263         if (ThrSTATE(t) == THR_NORMAL) {
 264             if (pthread_detach(t->Tself))
 265                 croak("pthread_detach failed");
 266             ThrSETSTATE(t, THR_DETACHED);
 267         }
 268
 269 void
 270 yield()
 271     CODE:
 272 #ifdef OLD_PTHREADS_API
 273         pthread_yield();
 274 #else
 275 #ifndef NO_SCHED_YIELD
 276         sched_yield();
 277 #endif /* NO_SCHED_YIELD */
 278 #endif /* OLD_PTHREADS_API */
 279
 280 void
 281 cond_wait(sv)
 282         SV *    sv
 283         MAGIC * mg = NO_INIT
 284 CODE:
 285         if (SvROK(sv))
 286             sv = SvRV(sv);
 287
 288         mg = condpair_magic(sv);
 289         DEBUG_L(PerlIO_printf(PerlIO_stderr(), "0x%lx: cond_wait 0x%lx\n",
 290                               (unsigned long)thr, (unsigned long)sv));
 291         MUTEX_LOCK(MgMUTEXP(mg));
 292         if (MgOWNER(mg) != thr) {
 293             MUTEX_UNLOCK(MgMUTEXP(mg));
 294             croak("cond_wait for lock that we don't own\n");
 295         }
 296         MgOWNER(mg) = 0;
 297         COND_WAIT(MgCONDP(mg), MgMUTEXP(mg));
 298         MgOWNER(mg) = thr;
 299         MUTEX_UNLOCK(MgMUTEXP(mg));
 300
 301 void
 302 cond_signal(sv)
 303         SV *    sv
 304         MAGIC * mg = NO_INIT
 305 CODE:
 306         if (SvROK(sv)) {
 307             /*
 308              * Kludge to allow lock of real objects without requiring
 309              * to pass in every type of argument by explicit reference.
 310              */
 311             sv = SvRV(sv);
 312         }
 313         mg = condpair_magic(sv);
 314         DEBUG_L(PerlIO_printf(PerlIO_stderr(), "0x%lx: cond_signal 0x%lx\n",
 315                               (unsigned long)thr, (unsigned long)sv));
 316         MUTEX_LOCK(MgMUTEXP(mg));
 317         if (MgOWNER(mg) != thr) {
 318             MUTEX_UNLOCK(MgMUTEXP(mg));
 319             croak("cond_signal for lock that we don't own\n");
 320         }
 321         COND_SIGNAL(MgCONDP(mg));
 322         MUTEX_UNLOCK(MgMUTEXP(mg));
 323
 324 void
 325 cond_broadcast(sv)
 326         SV *    sv
 327         MAGIC * mg = NO_INIT
 328 CODE:
 329         if (SvROK(sv))
 330             sv = SvRV(sv);
 331
 332         mg = condpair_magic(sv);
 333         DEBUG_L(PerlIO_printf(PerlIO_stderr(), "0x%lx: cond_broadcast 0x%lx\n",
 334                               (unsigned long)thr, (unsigned long)sv));
 335         MUTEX_LOCK(MgMUTEXP(mg));
 336         if (MgOWNER(mg) != thr) {
 337             MUTEX_UNLOCK(MgMUTEXP(mg));
 338             croak("cond_broadcast for lock that we don't own\n");
 339         }
 340         COND_BROADCAST(MgCONDP(mg));
 341         MUTEX_UNLOCK(MgMUTEXP(mg));