Thread.xs

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