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