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

Comparing Coro/Coro/State.xs (file contents):
Revision 1.376 by root, Fri Oct 2 20:48:04 2009 UTC vs.
Revision 1.396 by root, Fri May 6 18:01:13 2011 UTC

+/* this works around a bug in mingw32 providing a non-working setjmp */
+#define USE_NO_MINGW_SETJMP_TWO_ARGS
 #define NDEBUG 1
 #include "libcoro/coro.c"
 #define PERL_NO_GET_CONTEXT
 #ifndef SVs_PADSTALE
 # define SVs_PADSTALE 0
 #endif
-#ifdef WIN32
+#if defined(_WIN32)
+# undef HAS_GETTIMEOFDAY
 # undef setjmp
 # undef longjmp
 # undef _exit
 # define setjmp _setjmp /* deep magic */
 #else
   int valgrind_id;
 #endif
   unsigned char flags;
 } coro_cctx;
-coro_cctx *cctx_current; /* the currently running cctx */
+static coro_cctx *cctx_current; /* the currently running cctx */
 /*****************************************************************************/
 enum {
   CF_RUNNING   = 0x0001, /* coroutine is running */
   CF_READY     = 0x0002, /* coroutine is ready */
   CF_NEW       = 0x0004, /* has never been switched to */
   CF_DESTROYED = 0x0008, /* coroutine data has been freed */
   CF_SUSPENDED = 0x0010, /* coroutine can't be scheduled */
+  CF_NOCANCEL  = 0x0020, /* cannot cancel, set slf_frame.data to 1 (hackish) */
 };
 /* the structure where most of the perl state is stored, overlaid on the cxstack */
 typedef struct
 {
   CV *startcv; /* the CV to execute */
   AV *args;    /* data associated with this coroutine (initial args) */
   int refcnt;  /* coroutines are refcounted, yes */
   int flags;   /* CF_ flags */
   HV *hv;      /* the perl hash associated with this coro, if any */
-  void (*on_destroy)(pTHX_ struct coro *coro);
+  void (*on_destroy)(pTHX_ struct coro *coro); /* for temporary use by xs in critical sections */
   /* statistics */
   int usecount; /* number of transfers to this coro */
   /* coro process data */
 typedef struct coro *Coro__State;
 typedef struct coro *Coro__State_or_hashref;
 /* 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 */
+/* the main reason we don't support windows process emulation */
 static struct CoroSLF slf_frame; /* the current slf frame */
 /** Coro ********************************************************************/
 #define CORO_PRIO_MAX     3
   /* entersub is an UNOP, select a LISTOP... keep your fingers crossed */
   PL_op->op_flags |= OPf_STACKED;
   PL_op->op_private = 0;
   return PL_ppaddr [OP_ENTERSUB](aTHX);
 }
+/** time stuff **************************************************************/
+#ifdef HAS_GETTIMEOFDAY
+static void
+coro_u2time (pTHX_ UV ret[2])
+{
+  struct timeval tv;
+  gettimeofday (&tv, 0);
+  ret [0] = tv.tv_sec;
+  ret [1] = tv.tv_usec;
+}
+static double
+coro_nvtime ()
+{
+  struct timeval tv;
+  gettimeofday (&tv, 0);
+  return tv.tv_sec + tv.tv_usec * 1e-6;
+}
+static void
+time_init (pTHX)
+{
+  nvtime = coro_nvtime;
+  u2time = coro_u2time;
+}
+#else
+static void
+time_init (pTHX)
+{
+  SV **svp;
+  require_pv ("Time/HiRes.pm");
+  svp = hv_fetch (PL_modglobal, "Time::NVtime", 12, 0);
+  if (!svp)          croak ("Time::HiRes is required, but missing. Caught");
+  if (!SvIOK (*svp)) croak ("Time::NVtime isn't a function pointer. Caught");
+  nvtime = INT2PTR (double (*)(), SvIV (*svp));
+  svp = hv_fetch (PL_modglobal, "Time::U2time", 12, 0);
+  u2time = INT2PTR (void (*)(pTHX_ UV ret[2]), SvIV (*svp));
+}
+#endif
 /** lowlevel stuff **********************************************************/
 static SV *
 coro_get_sv (pTHX_ const char *name, int create)
       SvFLAGS (a)    = (SvFLAGS (a) & keep) | (SvFLAGS (b   ) & ~keep);
       SvFLAGS (b)    = (SvFLAGS (b) & keep) | (SvFLAGS (&tmp) & ~keep);
 #if PERL_VERSION_ATLEAST (5,10,0)
       /* perl 5.10 complicates this _quite_ a bit, but it also is
-       * is much faster, so no quarrels here. alternatively, we could
+       * much faster, so no quarrels here. alternatively, we could
        * sv_upgrade to avoid this.
        */
       {
         /* swap sv_u */
         tmp.sv_u = a->sv_u; a->sv_u = b->sv_u; b->sv_u = tmp.sv_u;
               {
                 CV *cv = cx->blk_sub.cv;
                 if (expect_true (CvDEPTH (cv)))
                   {
+                    EXTEND (SP, 3);
                     PUSHs ((SV *)CvPADLIST (cv));
                     PUSHs (INT2PTR (SV *, (IV)CvDEPTH (cv)));
                     PUSHs ((SV *)cv);
                     CvDEPTH (cv) = 0;
   PL_comppad_name_fill  = 0;
   PL_comppad_name_floor = 0;
   PL_curpm      = 0;
   PL_curpad     = 0;
   PL_localizing = 0;
-  PL_dirty      = 0;
   PL_restartop  = 0;
 #if PERL_VERSION_ATLEAST (5,10,0)
   PL_parser     = 0;
 #endif
   PL_hints      = 0;
     struct coro *current = SvSTATE_current;
     assert (("FATAL: tried to destroy currently running coroutine", coro->mainstack != PL_mainstack));
     save_perl (aTHX_ current);
+    SvRV_set (coro_current, (SV *)coro->hv); /* this will cause acroak in transfer_check */
     load_perl (aTHX_ coro);
     coro_unwind_stacks (aTHX);
     coro_destruct_stacks (aTHX);
     svf    [6] = (SV *)GvHV (PL_hintgv);
     svf    [7] =       PL_diehook;
     svf    [8] =       PL_warnhook;
     assert (9 == sizeof (svf) / sizeof (*svf));
+    SvRV_set (coro_current, (SV *)current->hv);
     load_perl (aTHX_ current);
   }
   {
     unsigned int i;
 cctx_destroy (coro_cctx *cctx)
 {
   if (!cctx)
     return;
-  assert (("FATAL: tried to destroy current cctx", cctx != cctx_current));//D temporary?
+  assert (("FATAL: tried to destroy current cctx", cctx != cctx_current));
   --cctx_count;
   coro_destroy (&cctx->cctx);
   /* coro_transfer creates new, empty cctx's */
 #define TRANSFER(ta, force_cctx) transfer (aTHX_ (ta).prev, (ta).next, (force_cctx))
 #define TRANSFER_CHECK(ta) transfer_check (aTHX_ (ta).prev, (ta).next)
 /** high level stuff ********************************************************/
-static int
+static void
 coro_state_destroy (pTHX_ struct coro *coro)
 {
   if (coro->flags & CF_DESTROYED)
-    return 0;
+    return;
+  /* this callback is reserved for slf functions needing to do cleanup */
   if (coro->on_destroy && !PL_dirty)
     coro->on_destroy (aTHX_ coro);
+  /*
+   * The on_destroy above most likely is from an SLF call.
+   * Since by definition the SLF call will not finish when we destroy
+   * the coro, we will have to force-finish it here, otherwise
+   * cleanup functions cannot call SLF functions.
+   */
+  coro->slf_frame.prepare = 0;
   coro->flags |= CF_DESTROYED;
   if (coro->flags & CF_READY)
     {
   cctx_destroy (coro->cctx);
   SvREFCNT_dec (coro->startcv);
   SvREFCNT_dec (coro->args);
   SvREFCNT_dec (coro->swap_sv);
   SvREFCNT_dec (CORO_THROW);
-  return 1;
 }
 static int
 coro_state_free (pTHX_ SV *sv, MAGIC *mg)
 {
         {
           /* nothing to schedule: call the idle handler */
           if (SvROK (sv_idle)
               && SvOBJECT (SvRV (sv_idle)))
             {
+              if (SvRV (sv_idle) == SvRV (coro_current))
+                croak ("FATAL: $Coro::IDLE blocked itself - did you try to block inside an event loop callback? Caught");
               ++coro_nready; /* hack so that api_ready doesn't invoke ready hook */
               api_ready (aTHX_ SvRV (sv_idle));
               --coro_nready;
             }
           else
 static void
 coro_call_on_destroy (pTHX_ struct coro *coro)
 {
   SV **on_destroyp = hv_fetch (coro->hv, "_on_destroy", sizeof ("_on_destroy") - 1, 0);
-  SV **statusp     = hv_fetch (coro->hv, "_status", sizeof ("_status") - 1, 0);
   if (on_destroyp)
     {
-      AV *on_destroy = (AV *)SvRV (*on_destroyp);
+      SV **statusp = hv_fetch (coro->hv, "_status", sizeof ("_status") - 1, 0);
+      AV *on_destroy =           sv_2mortal (SvREFCNT_inc ((AV *)SvRV (*on_destroyp)));
+      AV *status     = statusp ? sv_2mortal (SvREFCNT_inc ((AV *)SvRV (*statusp))) : 0;
       while (AvFILLp (on_destroy) >= 0)
         {
           dSP; /* don't disturb outer sp */
           SV *cb = av_pop (on_destroy);
           PUSHMARK (SP);
           if (statusp)
             {
               int i;
-              AV *status = (AV *)SvRV (*statusp);
               EXTEND (SP, AvFILLp (status) + 1);
               for (i = 0; i <= AvFILLp (status); ++i)
                 PUSHs (AvARRAY (status)[i]);
             }
         }
     }
 }
 static void
-slf_init_terminate (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
+coro_set_status (HV *coro_hv, SV **arg, int items)
 {
-  int i;
-  HV *hv = (HV *)SvRV (coro_current);
   AV *av = newAV ();
   /* items are actually not so common, so optimise for this case */
   if (items)
     {
+      int i;
       av_extend (av, items - 1);
       for (i = 0; i < items; ++i)
         av_push (av, SvREFCNT_inc_NN (arg [i]));
     }
-  hv_store (hv, "_status", sizeof ("_status") - 1, newRV_noinc ((SV *)av), 0);
+  hv_store (coro_hv, "_status", sizeof ("_status") - 1, newRV_noinc ((SV *)av), 0);
+}
+static void
+slf_init_terminate_cancel_common (pTHX_ struct CoroSLF *frame, HV *coro_hv)
+{
-  av_push (av_destroy, (SV *)newRV_inc ((SV *)hv)); /* RVinc for perl */
+  av_push (av_destroy, (SV *)newRV_inc ((SV *)coro_hv)); /* RVinc for perl */
   api_ready (aTHX_ sv_manager);
   frame->prepare = prepare_schedule;
   frame->check   = slf_check_repeat;
   /* as a minor optimisation, we could unwind all stacks here */
   /* but that puts extra pressure on pp_slf, and is not worth much */
   /*coro_unwind_stacks (aTHX);*/
+}
+static void
+slf_init_terminate (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
+{
+  HV *coro_hv = (HV *)SvRV (coro_current);
+  coro_set_status (coro_hv, arg, items);
+  slf_init_terminate_cancel_common (frame, coro_hv);
+}
+static void
+slf_init_cancel (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
+{
+  HV *coro_hv;
+  struct coro *coro;
+  if (items <= 0)
+    croak ("Coro::cancel called without coro object,");
+  coro = SvSTATE (arg [0]);
+  coro_hv = coro->hv;
+  coro_set_status (coro_hv, arg + 1, items - 1);
+  if (expect_false (coro->flags & CF_NOCANCEL))
+    {
+      /* coro currently busy cancelling something, so just notify it */
+      coro->slf_frame.data = (void *)coro;
+      frame->prepare = prepare_nop;
+      frame->check   = slf_check_nop;
+    }
+  else if (coro_hv == (HV *)SvRV (coro_current))
+    {
+      /* cancelling the current coro is allowed, and equals terminate */
+      slf_init_terminate_cancel_common (frame, coro_hv);
+    }
+  else
+    {
+      struct coro *self = SvSTATE_current;
+      /* otherwise we cancel directly, purely for speed reasons
+       * unfortunately, this requires some magic trickery, as
+       * somebody else could cancel us, so we have to fight the cancellation.
+       * this is ugly, and hopefully fully worth the extra speed.
+       * besides, I can't get the slow-but-safe version working...
+       */
+      slf_frame.data = 0;
+      self->flags |= CF_NOCANCEL;
+      coro_state_destroy (aTHX_ coro);
+      coro_call_on_destroy (aTHX_ coro);
+      self->flags &= ~CF_NOCANCEL;
+      if (slf_frame.data)
+        {
+          /* while we were busy we have been cancelled, so terminate */
+          slf_init_terminate_cancel_common (frame, self->hv);
+        }
+      else
+        {
+          frame->prepare = prepare_nop;
+          frame->check   = slf_check_nop;
+        }
+    }
 }
 /*****************************************************************************/
 /* async pool handler */
   if (PL_op->op_flags & OPf_STACKED)
     {
       if (items > slf_arga)
         {
           slf_arga = items;
-          free (slf_argv);
+          Safefree (slf_argv);
-          slf_argv = malloc (slf_arga * sizeof (SV *));
+          New (0, slf_argv, slf_arga, SV *);
         }
       slf_argc = items;
       for (i = 0; i < items; ++i)
     {
       SV *cb_cv = s_get_cv_croak (arg [1]);
       av_push (av, SvREFCNT_inc_NN (cb_cv));
       if (SvIVX (AvARRAY (av)[0]))
-        coro_signal_wake (aTHX_ av, 1); /* ust be the only waiter */
+        coro_signal_wake (aTHX_ av, 1); /* must be the only waiter */
       frame->prepare = prepare_nop;
       frame->check   = slf_check_nop;
     }
   else if (SvIVX (AvARRAY (av)[0]))
     /* now call the AIO function - we assume our request is uncancelable */
     PUTBACK;
     call_sv ((SV *)req, G_VOID | G_DISCARD);
   }
-  /* now that the requets is going, we loop toll we have a result */
+  /* now that the request is going, we loop till we have a result */
   frame->data    = (void *)state;
   frame->prepare = prepare_schedule;
   frame->check   = slf_check_aio_req;
 }
         coroapi.prepare_nop          = prepare_nop;
         coroapi.prepare_schedule     = prepare_schedule;
         coroapi.prepare_cede         = prepare_cede;
         coroapi.prepare_cede_notself = prepare_cede_notself;
-        {
+        time_init (aTHX);
-          SV **svp = hv_fetch (PL_modglobal, "Time::NVtime", 12, 0);
-          if (!svp)          croak ("Time::HiRes is required");
-          if (!SvIOK (*svp)) croak ("Time::NVtime isn't a function pointer");
-          nvtime = INT2PTR (double (*)(), SvIV (*svp));
-          svp = hv_fetch (PL_modglobal, "Time::U2time", 12, 0);
-          u2time = INT2PTR (void (*)(pTHX_ UV ret[2]), SvIV (*svp));
-        }
         assert (("PRIO_NORMAL must be 0", !CORO_PRIO_NORMAL));
 }
 SV *
 void
 transfer (...)
         PROTOTYPE: $$
 	CODE:
         CORO_EXECUTE_SLF_XS (slf_init_transfer);
-bool
-_destroy (SV *coro_sv)
-	CODE:
-	RETVAL = coro_state_destroy (aTHX_ SvSTATE (coro_sv));
-	OUTPUT:
-        RETVAL
 void
 _exit (int code)
         PROTOTYPE: $
 	CODE:
 	CODE:
 {
         if (coro->mainstack && ((coro->flags & CF_RUNNING) || coro->slot))
           {
             struct coro *current = SvSTATE_current;
+            struct CoroSLF slf_save;
             if (current != coro)
               {
                 PUTBACK;
                 save_perl (aTHX_ current);
                 load_perl (aTHX_ coro);
+                /* the coro is most likely in an active SLF call.
+                 * while not strictly required (the code we execute is
+                 * not allowed to call any SLF functions), it's cleaner
+                 * to reinitialise the slf_frame and restore it later.
+                 * This might one day allow us to actually do SLF calls
+                 * from code executed here.
+                 */
+                slf_save = slf_frame;
+                slf_frame.prepare = 0;
                 SPAGAIN;
               }
             PUSHSTACK;
             SPAGAIN;
             if (current != coro)
               {
                 PUTBACK;
+                slf_frame = slf_save;
                 save_perl (aTHX_ coro);
                 load_perl (aTHX_ current);
                 SPAGAIN;
               }
           }
         RETVAL = boolSV (coro->flags & ix);
 	OUTPUT:
         RETVAL
 void
-throw (Coro::State self, SV *throw = &PL_sv_undef)
+throw (Coro::State self, SV *exception = &PL_sv_undef)
 	PROTOTYPE: $;$
         CODE:
 {
 	struct coro *current = SvSTATE_current;
-	SV **throwp = self == current ? &CORO_THROW : &self->except;
+	SV **exceptionp = self == current ? &CORO_THROW : &self->except;
-        SvREFCNT_dec (*throwp);
+        SvREFCNT_dec (*exceptionp);
-        SvGETMAGIC (throw);
+        SvGETMAGIC (exception);
-        *throwp = SvOK (throw) ? newSVsv (throw) : 0;
+        *exceptionp = SvOK (exception) ? newSVsv (exception) : 0;
 }
 void
 api_trace (SV *coro, int flags = CC_TRACE | CC_TRACE_SUB)
 	PROTOTYPE: $;$
 void
 cancel (Coro::State self)
 	CODE:
 	coro_state_destroy (aTHX_ self);
-        coro_call_on_destroy (aTHX_ self); /* actually only for Coro objects */
 SV *
 enable_times (int enabled = enable_times)
 	CODE:
 {
         api_ready (aTHX_ RETVAL);
 	OUTPUT:
         RETVAL
 void
+_destroy (Coro::State coro)
+	CODE:
+	/* used by the manager thread */
+	coro_state_destroy (aTHX_ coro);
+        coro_call_on_destroy (aTHX_ coro);
+void
 terminate (...)
 	CODE:
         CORO_EXECUTE_SLF_XS (slf_init_terminate);
+void
+cancel (...)
+	CODE:
+        CORO_EXECUTE_SLF_XS (slf_init_cancel);
 void
 schedule (...)
 	CODE:
         CORO_EXECUTE_SLF_XS (slf_init_schedule);
           {
             PL_ppaddr [OP_SSELECT] = coro_old_pp_sselect;
             coro_old_pp_sselect = 0;
           }

Diff Legend

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

Comparing Coro/Coro/State.xs (file contents): Revision 1.376 by root, Fri Oct 2 20:48:04 2009 UTC vs. Revision 1.396 by root, Fri May 6 18:01:13 2011 UTC

Diff Legend

Comparing Coro/Coro/State.xs (file contents):
Revision 1.376 by root, Fri Oct 2 20:48:04 2009 UTC vs.
Revision 1.396 by root, Fri May 6 18:01:13 2011 UTC