[ViewVC] Diff of: cvs/Coro/Coro/State.xs

Comparing Coro/Coro/State.xs (file contents):
Revision 1.292 by root, Tue Nov 18 06:09:35 2008 UTC vs.
Revision 1.299 by root, Wed Nov 19 00:06:55 2008 UTC

 #endif
 /* The next macros try to return the current stack pointer, in an as
  * portable way as possible. */
 #if __GNUC__ >= 4
+# define dSTACKLEVEL int stacklevel_dummy
-# define dSTACKLEVEL void *stacklevel = __builtin_frame_address (0)
+# define STACKLEVEL __builtin_frame_address (0)
 #else
-# define dSTACKLEVEL volatile void *stacklevel = (volatile void *)&stacklevel
+# define dSTACKLEVEL volatile void *stacklevel
+# define STACKLEVEL ((void *)&stacklevel)
 #endif
 #define IN_DESTRUCT (PL_main_cv == Nullcv)
 #if __GNUC__ >= 3
 #define expect_true(expr)  expect ((expr) != 0, 1)
 #define NOINLINE attribute ((noinline))
 #include "CoroAPI.h"
+#define GCoroAPI (&coroapi) /* very sneaky */
 #ifdef USE_ITHREADS
 # if CORO_PTHREAD
 static void *coro_thx;
 # endif
   /* statistics */
   int usecount; /* number of transfers to this coro */
   /* coro process data */
   int prio;
-  SV *throw; /* exception to be thrown */
+  SV *except; /* exception to be thrown */
   /* async_pool */
   SV *saved_deffh;
   /* linked list */
 /* the following variables are effectively part of the perl context */
 /* and get copied between struct coro and these variables */
 /* the mainr easonw e don't support windows process emulation */
 static struct CoroSLF slf_frame; /* the current slf frame */
-static SV *coro_throw;
 /** Coro ********************************************************************/
 #define PRIO_MAX     3
 #define PRIO_HIGH    1
     PUTBACK;
   }
   slf_frame  = c->slf_frame;
-  coro_throw = c->throw;
+  CORO_THROW = c->except;
 }
 static void
 save_perl (pTHX_ Coro__State c)
 {
-  c->throw     = coro_throw;
+  c->except    = CORO_THROW;
   c->slf_frame = slf_frame;
   {
     dSP;
     I32 cxix = cxstack_ix;
   {
     dSP;
     UNOP myop;
     Zero (&myop, 1, UNOP);
-    myop.op_next = Nullop;
+    myop.op_next  = Nullop;
     myop.op_flags = OPf_WANT_VOID;
     PUSHMARK (SP);
     XPUSHs (sv_2mortal (av_shift (GvAV (PL_defgv))));
     PUTBACK;
    */
   slf_frame.prepare = prepare_nop;   /* provide a nop function for an eventual pp_slf */
   slf_frame.check   = slf_check_nop; /* signal pp_slf to not repeat */
   /* and we have to provide the pp_slf op in any case, so pp_slf can skip it */
+  coro_setup_op.op_next   = PL_op;
   coro_setup_op.op_type   = OP_CUSTOM;
   coro_setup_op.op_ppaddr = pp_slf;
-  coro_setup_op.op_next   = PL_op;
+  /* no flags required, as an init function won't be called */
   PL_op = (OP *)&coro_setup_op;
   /* copy throw, in case it was set before coro_setup */
-  coro_throw = coro->throw;
+  CORO_THROW = coro->except;
 }
 static void
 coro_destruct (pTHX_ struct coro *coro)
 {
   SvREFCNT_dec (PL_diehook);
   SvREFCNT_dec (PL_warnhook);
   SvREFCNT_dec (coro->saved_deffh);
-  SvREFCNT_dec (coro_throw);
+  SvREFCNT_dec (CORO_THROW);
   coro_destruct_stacks (aTHX);
 }
 INLINE void
 static int
 slf_check_set_stacklevel (pTHX_ struct CoroSLF *frame)
 {
   *frame = cctx_ssl_frame;
-  return 1; /* execute the restored frame - there must be one */
+  return frame->check (aTHX_ frame); /* execute the restored frame - there must be one */
 }
-/* initialises PL_top_env and injects a pseudo-slf-call to sett he stacklevel */
+/* initialises PL_top_env and injects a pseudo-slf-call to set the stacklevel */
 static void NOINLINE
 cctx_prepare (pTHX_ coro_cctx *cctx)
 {
   PL_top_env = &PL_start_env;
   if (cctx->flags & CC_TRACE)
     PL_runops = runops_trace;
-  /* we already must be in an SLF call, there is no other valid way
+  /* we already must be executing an SLF op, there is no other valid way
    * that can lead to creation of a new cctx */
   assert (("FATAL: can't prepare slf-less cctx in Coro module (please report)",
            slf_frame.prepare && PL_op->op_ppaddr == pp_slf));
+  /* we must emulate leaving pp_slf, which is done inside slf_check_set_stacklevel */
   cctx_ssl_cctx  = cctx;
   cctx_ssl_frame = slf_frame;
   slf_frame.prepare = slf_prepare_set_stacklevel;
   slf_frame.check   = slf_check_set_stacklevel;
   dSTACKLEVEL;
   /* sometimes transfer is only called to set idle_sp */
   if (expect_false (!next))
     {
-      ((coro_cctx *)prev)->idle_sp = (void *)stacklevel;
+      ((coro_cctx *)prev)->idle_sp = STACKLEVEL;
       assert (((coro_cctx *)prev)->idle_te = PL_top_env); /* just for the side-effect when asserts are enabled */
     }
   else if (expect_true (prev != next))
     {
       coro_cctx *prev__cctx;
       prev__cctx = prev->cctx;
       /* possibly untie and reuse the cctx */
       if (expect_true (
-            prev__cctx->idle_sp == (void *)stacklevel
+            prev__cctx->idle_sp == STACKLEVEL
             && !(prev__cctx->flags & CC_TRACE)
             && !force_cctx
          ))
         {
           /* I assume that stacklevel is a stronger indicator than PL_top_env changes */
     }
   while (slf_frame.check (aTHX_ &slf_frame));
   slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */
+  /* exception handling */
+  if (expect_false (CORO_THROW))
+    {
+      SV *exception = sv_2mortal (CORO_THROW);
+      CORO_THROW = 0;
+      sv_setsv (ERRSV, exception);
+      croak (0);
+    }
   /* return value handling - mostly like entersub */
   /* make sure we put something on the stack in scalar context */
   if (GIMME_V == G_SCALAR)
     {
       dSP;
         bot [1] = *sp;
       SP = bot + 1;
       PUTBACK;
-    }
-  /* exception handling */
-  if (expect_false (coro_throw))
-    {
-      SV *exception = sv_2mortal (coro_throw);
-      coro_throw = 0;
-      sv_setsv (ERRSV, exception);
-      croak (0);
     }
   return NORMAL;
 }
 };
 /*****************************************************************************/
 /* Coro::Semaphore */
+static SV *
+coro_semaphore_new (int count)
+{
+  /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */
+  AV *av = newAV ();
+  SV **ary;
+  /* unfortunately, building manually saves memory */
+  Newx (ary, 2, SV *);
+  AvALLOC (av) = ary;
+  AvARRAY (av) = ary;
+  AvMAX   (av) = 1;
+  AvFILLp (av) = 0;
+  ary [0] = newSViv (count);
+  return newRV_noinc ((SV *)av);
+}
 static void
 coro_semaphore_adjust (pTHX_ AV *av, IV adjust)
 {
   SV *count_sv = AvARRAY (av)[0];
   IV count = SvIVX (count_sv);
   /* call $sem->adjust (0) to possibly wake up some other waiters */
   coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);
 }
 static int
-slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
+slf_check_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, int acquire)
 {
   AV *av = (AV *)frame->data;
   SV *count_sv = AvARRAY (av)[0];
+  /* if we are about to throw, don't actually acquire the lock, just throw */
+  if (CORO_THROW)
+    return 0;
-  if (SvIVX (count_sv) > 0)
+  else if (SvIVX (count_sv) > 0)
     {
       SvSTATE_current->on_destroy = 0;
+      if (acquire)
-      SvIVX (count_sv) = SvIVX (count_sv) - 1;
+        SvIVX (count_sv) = SvIVX (count_sv) - 1;
+      else
+        coro_semaphore_adjust (aTHX_ av, 0);
       return 0;
     }
   else
     {
       int i;
       av_push (av, SvREFCNT_inc (SvRV (coro_current)));
       return 1;
     }
 }
-static void
+static int
+slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
+{
+  return slf_check_semaphore_down_or_wait (aTHX_ frame, 1);
+}
+static int
+slf_check_semaphore_wait (pTHX_ struct CoroSLF *frame)
+{
+  return slf_check_semaphore_down_or_wait (aTHX_ frame, 0);
+}
+static void
-slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
+slf_init_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
 {
   AV *av = (AV *)SvRV (arg [0]);
   if (SvIVX (AvARRAY (av)[0]) > 0)
     {
       frame->data    = (void *)av;
       frame->prepare = prepare_nop;
-      SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
     }
   else
     {
       av_push (av, SvREFCNT_inc (SvRV (coro_current)));
       frame->data    = (void *)sv_2mortal (SvREFCNT_inc ((SV *)av));
       frame->prepare = prepare_schedule;
       /* to avoid race conditions when a woken-up coro gets terminated */
       /* we arrange for a temporary on_destroy that calls adjust (0) */
-      assert (!SvSTATE_current->on_destroy);//D
       SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
     }
+}
+static void
+slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
+{
+  slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
   frame->check = slf_check_semaphore_down;
+}
+static void
+slf_init_semaphore_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
+{
+  slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
+  frame->check = slf_check_semaphore_wait;
 }
 /*****************************************************************************/
 /* gensub: simple closure generation utility */
 static int
 slf_check_aio_req (pTHX_ struct CoroSLF *frame)
 {
   AV *state = (AV *)frame->data;
+  /* if we are about to throw, return early */
+  /* this does not cancel the aio request, but at least */
+  /* it quickly returns */
+  if (CORO_THROW)
+    return 0;
   /* one element that is an RV? repeat! */
   if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0]))
     return 1;
 throw (Coro::State self, SV *throw = &PL_sv_undef)
 	PROTOTYPE: $;$
         CODE:
 {
 	struct coro *current = SvSTATE_current;
-	SV **throwp = self == current ? &coro_throw : &self->throw;
+	SV **throwp = self == current ? &CORO_THROW : &self->except;
         SvREFCNT_dec (*throwp);
         *throwp = SvOK (throw) ? newSVsv (throw) : 0;
 }
 void
           coroapi.ready        = api_ready;
           coroapi.is_ready     = api_is_ready;
           coroapi.nready       = coro_nready;
           coroapi.current      = coro_current;
-          GCoroAPI = &coroapi;
+          /*GCoroAPI = &coroapi;*/
           sv_setiv (sv, (IV)&coroapi);
           SvREADONLY_on (sv);
         }
 }
 MODULE = Coro::State                PACKAGE = Coro::Semaphore
 SV *
-new (SV *klass, SV *count_ = 0)
+new (SV *klass, SV *count = 0)
 	CODE:
-{
+        RETVAL = sv_bless (
-        /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */
+                   coro_semaphore_new (count && SvOK (count) ? SvIV (count) : 1),
-        AV *av = newAV ();
+                   GvSTASH (CvGV (cv))
-        SV **ary;
+                 );
+	OUTPUT:
+        RETVAL
-        /* unfortunately, building manually saves memory */
+# helper for Coro::Channel
-        Newx (ary, 2, SV *);
+SV *
-        AvALLOC (av) = ary;
+_alloc (int count)
-        AvARRAY (av) = ary;
+	CODE:
-        AvMAX   (av) = 1;
+        RETVAL = coro_semaphore_new (count);
-        AvFILLp (av) = 0;
-        ary [0] = newSViv (count_ && SvOK (count_) ? SvIV (count_) : 1);
-        RETVAL = sv_bless (newRV_noinc ((SV *)av), GvSTASH (CvGV (cv)));
-}
 	OUTPUT:
         RETVAL
 SV *
 count (SV *self)
         CODE:
         coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1);
 void
 down (SV *self)
+	ALIAS:
+        Coro::Signal::wait = 0
         CODE:
         CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
+void
+wait (SV *self)
+        CODE:
+        CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
 void
 try (SV *self)
         PPCODE:
 {
           XSRETURN_NO;
 }
 void
 waiters (SV *self)
-    	CODE:
+    	PPCODE:
 {
         AV *av = (AV *)SvRV (self);
+        int wcount = AvFILLp (av) + 1 - 1;
         if (GIMME_V == G_SCALAR)
-          XPUSHs (sv_2mortal (newSVsv (AvARRAY (av)[0])));
+          XPUSHs (sv_2mortal (newSViv (wcount)));
         else
           {
             int i;
-            EXTEND (SP, AvFILLp (av) + 1 - 1);
+            EXTEND (SP, wcount);
-            for (i = 1; i <= AvFILLp (av); ++i)
+            for (i = 1; i <= wcount; ++i)
               PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
           }
+}
+MODULE = Coro::State                PACKAGE = Coro::Signal
+SV *
+new (SV *klass)
+	CODE:
+        RETVAL = sv_bless (
+                   coro_semaphore_new (0),
+                   GvSTASH (CvGV (cv))
+                 );
+        OUTPUT:
+        RETVAL
+void
+broadcast (SV *self, int adjust = 1)
+        CODE:
+{
+	AV *av = (AV *)SvRV (self);
+        SvIVX (AvARRAY (av)[0]) = 0; /* not necessary, but gives me fuzzy warm feelings */
+        coro_semaphore_adjust (aTHX_ av, AvFILLp (av) + 1 - 1);
+        SvIVX (AvARRAY (av)[0]) = 0; /* necessary */
 }
 MODULE = Coro::State                PACKAGE = Coro::AnyEvent

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing Coro/Coro/State.xs (file contents): Revision 1.292 by root, Tue Nov 18 06:09:35 2008 UTC vs. Revision 1.299 by root, Wed Nov 19 00:06:55 2008 UTC

Diff Legend

Comparing Coro/Coro/State.xs (file contents):
Revision 1.292 by root, Tue Nov 18 06:09:35 2008 UTC vs.
Revision 1.299 by root, Wed Nov 19 00:06:55 2008 UTC