ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/Coro/Coro/State.xs
(Generate patch)

Comparing Coro/Coro/State.xs (file contents):
Revision 1.307 by root, Wed Nov 19 14:34:46 2008 UTC vs.
Revision 1.317 by root, Thu Nov 20 06:28:52 2008 UTC

55 55
56#define PERL_VERSION_ATLEAST(a,b,c) \ 56#define PERL_VERSION_ATLEAST(a,b,c) \
57 (PERL_REVISION > (a) \ 57 (PERL_REVISION > (a) \
58 || (PERL_REVISION == (a) \ 58 || (PERL_REVISION == (a) \
59 && (PERL_VERSION > (b) \ 59 && (PERL_VERSION > (b) \
60 || (PERL_VERSION == (b) && PERLSUBVERSION >= (c))))) 60 || (PERL_VERSION == (b) && PERL_SUBVERSION >= (c)))))
61 61
62#if !PERL_VERSION_ATLEAST (5,6,0) 62#if !PERL_VERSION_ATLEAST (5,6,0)
63# ifndef PL_ppaddr 63# ifndef PL_ppaddr
64# define PL_ppaddr ppaddr 64# define PL_ppaddr ppaddr
65# endif 65# endif
98# define newSV(l) NEWSV(0,l) 98# define newSV(l) NEWSV(0,l)
99#endif 99#endif
100#ifndef CvISXSUB_on 100#ifndef CvISXSUB_on
101# define CvISXSUB_on(cv) (void)cv 101# define CvISXSUB_on(cv) (void)cv
102#endif 102#endif
103#ifndef CvISXSUB
104# define CvISXSUB(cv) (CvXSUB (cv) ? TRUE : FALSE)
105#endif
103 106
104/* 5.8.7 */ 107/* 5.8.7 */
105#ifndef SvRV_set 108#ifndef SvRV_set
106# define SvRV_set(s,v) SvRV(s) = (v) 109# define SvRV_set(s,v) SvRV(s) = (v)
107#endif 110#endif
176static HV *hv_sig; /* %SIG */ 179static HV *hv_sig; /* %SIG */
177 180
178/* async_pool helper stuff */ 181/* async_pool helper stuff */
179static SV *sv_pool_rss; 182static SV *sv_pool_rss;
180static SV *sv_pool_size; 183static SV *sv_pool_size;
184static SV *sv_async_pool_idle;
181static AV *av_async_pool; 185static AV *av_async_pool;
186static SV *sv_Coro;
187static CV *cv_pool_handler;
188static CV *cv_coro_state_new;
182 189
183/* Coro::AnyEvent */ 190/* Coro::AnyEvent */
184static SV *sv_activity; 191static SV *sv_activity;
185 192
186static struct coro_cctx *cctx_first; 193static struct coro_cctx *cctx_first;
263 SV *except; /* exception to be thrown */ 270 SV *except; /* exception to be thrown */
264 SV *rouse_cb; 271 SV *rouse_cb;
265 272
266 /* async_pool */ 273 /* async_pool */
267 SV *saved_deffh; 274 SV *saved_deffh;
275 SV *invoke_cb;
276 AV *invoke_av;
268 277
269 /* linked list */ 278 /* linked list */
270 struct coro *next, *prev; 279 struct coro *next, *prev;
271}; 280};
272 281
289 298
290/* for Coro.pm */ 299/* for Coro.pm */
291static SV *coro_current; 300static SV *coro_current;
292static SV *coro_readyhook; 301static SV *coro_readyhook;
293static AV *coro_ready [PRIO_MAX - PRIO_MIN + 1]; 302static AV *coro_ready [PRIO_MAX - PRIO_MIN + 1];
303static CV *cv_coro_run, *cv_coro_terminate;
294static struct coro *coro_first; 304static struct coro *coro_first;
295#define coro_nready coroapi.nready 305#define coro_nready coroapi.nready
296 306
297/** lowlevel stuff **********************************************************/ 307/** lowlevel stuff **********************************************************/
298 308
326 return get_hv (name, create); 336 return get_hv (name, create);
327} 337}
328 338
329/* may croak */ 339/* may croak */
330INLINE CV * 340INLINE CV *
331coro_sv_2cv (SV *sv) 341coro_sv_2cv (pTHX_ SV *sv)
332{ 342{
333 HV *st; 343 HV *st;
334 GV *gvp; 344 GV *gvp;
335 return sv_2cv (sv, &st, &gvp, 0); 345 return sv_2cv (sv, &st, &gvp, 0);
336} 346}
686#if !PERL_VERSION_ATLEAST (5,10,0) 696#if !PERL_VERSION_ATLEAST (5,10,0)
687 Safefree (PL_retstack); 697 Safefree (PL_retstack);
688#endif 698#endif
689} 699}
690 700
701#define CORO_RSS \
702 rss += sizeof (SYM (curstackinfo)); \
703 rss += (SYM (curstackinfo->si_cxmax) + 1) * sizeof (PERL_CONTEXT); \
704 rss += sizeof (SV) + sizeof (struct xpvav) + (1 + AvMAX (SYM (curstack))) * sizeof (SV *); \
705 rss += SYM (tmps_max) * sizeof (SV *); \
706 rss += (SYM (markstack_max) - SYM (markstack_ptr)) * sizeof (I32); \
707 rss += SYM (scopestack_max) * sizeof (I32); \
708 rss += SYM (savestack_max) * sizeof (ANY);
709
691static size_t 710static size_t
692coro_rss (pTHX_ struct coro *coro) 711coro_rss (pTHX_ struct coro *coro)
693{ 712{
694 size_t rss = sizeof (*coro); 713 size_t rss = sizeof (*coro);
695 714
696 if (coro->mainstack) 715 if (coro->mainstack)
697 { 716 {
698 perl_slots tmp_slot;
699 perl_slots *slot;
700
701 if (coro->flags & CF_RUNNING) 717 if (coro->flags & CF_RUNNING)
702 { 718 {
703 slot = &tmp_slot; 719 #define SYM(sym) PL_ ## sym
704 720 CORO_RSS;
705 #define VAR(name,type) slot->name = PL_ ## name;
706 # include "state.h"
707 #undef VAR 721 #undef SYM
708 } 722 }
709 else 723 else
710 slot = coro->slot;
711
712 if (slot)
713 { 724 {
714 rss += sizeof (slot->curstackinfo); 725 #define SYM(sym) coro->slot->sym
715 rss += (slot->curstackinfo->si_cxmax + 1) * sizeof (PERL_CONTEXT); 726 CORO_RSS;
716 rss += sizeof (SV) + sizeof (struct xpvav) + (1 + AvMAX (slot->curstack)) * sizeof (SV *); 727 #undef SYM
717 rss += slot->tmps_max * sizeof (SV *);
718 rss += (slot->markstack_max - slot->markstack_ptr) * sizeof (I32);
719 rss += slot->scopestack_max * sizeof (I32);
720 rss += slot->savestack_max * sizeof (ANY);
721
722#if !PERL_VERSION_ATLEAST (5,10,0)
723 rss += slot->retstack_max * sizeof (OP *);
724#endif
725 } 728 }
726 } 729 }
727 730
728 return rss; 731 return rss;
729} 732}
871 dSP; 874 dSP;
872 UNOP myop; 875 UNOP myop;
873 876
874 Zero (&myop, 1, UNOP); 877 Zero (&myop, 1, UNOP);
875 myop.op_next = Nullop; 878 myop.op_next = Nullop;
879 myop.op_type = OP_ENTERSUB;
876 myop.op_flags = OPf_WANT_VOID; 880 myop.op_flags = OPf_WANT_VOID;
877 881
878 PUSHMARK (SP); 882 PUSHMARK (SP);
879 XPUSHs (sv_2mortal (av_shift (GvAV (PL_defgv)))); 883 PUSHs ((SV *)coro->startcv);
880 PUTBACK; 884 PUTBACK;
881 PL_op = (OP *)&myop; 885 PL_op = (OP *)&myop;
882 PL_op = PL_ppaddr[OP_ENTERSUB](aTHX); 886 PL_op = PL_ppaddr[OP_ENTERSUB](aTHX);
883 SPAGAIN;
884 } 887 }
885 888
886 /* this newly created coroutine might be run on an existing cctx which most 889 /* this newly created coroutine might be run on an existing cctx which most
887 * likely was suspended in pp_slf, so we have to emulate entering pp_slf here. 890 * likely was suspended in pp_slf, so we have to emulate entering pp_slf here.
888 */ 891 */
891 894
892 /* and we have to provide the pp_slf op in any case, so pp_slf can skip it */ 895 /* and we have to provide the pp_slf op in any case, so pp_slf can skip it */
893 coro_setup_op.op_next = PL_op; 896 coro_setup_op.op_next = PL_op;
894 coro_setup_op.op_type = OP_CUSTOM; 897 coro_setup_op.op_type = OP_CUSTOM;
895 coro_setup_op.op_ppaddr = pp_slf; 898 coro_setup_op.op_ppaddr = pp_slf;
896 /* no flags required, as an init function won't be called */ 899 /* no flags etc. required, as an init function won't be called */
897 900
898 PL_op = (OP *)&coro_setup_op; 901 PL_op = (OP *)&coro_setup_op;
899 902
900 /* copy throw, in case it was set before coro_setup */ 903 /* copy throw, in case it was set before coro_setup */
901 CORO_THROW = coro->except; 904 CORO_THROW = coro->except;
929 SvREFCNT_dec (GvSV (irsgv)); 932 SvREFCNT_dec (GvSV (irsgv));
930 933
931 SvREFCNT_dec (PL_diehook); 934 SvREFCNT_dec (PL_diehook);
932 SvREFCNT_dec (PL_warnhook); 935 SvREFCNT_dec (PL_warnhook);
933 936
934 SvREFCNT_dec (CORO_THROW);
935 SvREFCNT_dec (coro->saved_deffh); 937 SvREFCNT_dec (coro->saved_deffh);
936 SvREFCNT_dec (coro->rouse_cb); 938 SvREFCNT_dec (coro->rouse_cb);
939 SvREFCNT_dec (coro->invoke_cb);
940 SvREFCNT_dec (coro->invoke_av);
937 941
938 coro_destruct_stacks (aTHX); 942 coro_destruct_stacks (aTHX);
939} 943}
940 944
941INLINE void 945INLINE void
1144 transfer_tail (aTHX); 1148 transfer_tail (aTHX);
1145 1149
1146 /* somebody or something will hit me for both perl_run and PL_restartop */ 1150 /* somebody or something will hit me for both perl_run and PL_restartop */
1147 PL_restartop = PL_op; 1151 PL_restartop = PL_op;
1148 perl_run (PL_curinterp); 1152 perl_run (PL_curinterp);
1153 /*
1154 * Unfortunately, there is no way to get at the return values of the
1155 * coro body here, as perl_run destroys these
1156 */
1149 1157
1150 /* 1158 /*
1151 * If perl-run returns we assume exit() was being called or the coro 1159 * If perl-run returns we assume exit() was being called or the coro
1152 * fell off the end, which seems to be the only valid (non-bug) 1160 * fell off the end, which seems to be the only valid (non-bug)
1153 * reason for perl_run to return. We try to exit by jumping to the 1161 * reason for perl_run to return. We try to exit by jumping to the
1447 } 1455 }
1448 1456
1449 cctx_destroy (coro->cctx); 1457 cctx_destroy (coro->cctx);
1450 SvREFCNT_dec (coro->startcv); 1458 SvREFCNT_dec (coro->startcv);
1451 SvREFCNT_dec (coro->args); 1459 SvREFCNT_dec (coro->args);
1460 SvREFCNT_dec (CORO_THROW);
1452 1461
1453 if (coro->next) coro->next->prev = coro->prev; 1462 if (coro->next) coro->next->prev = coro->prev;
1454 if (coro->prev) coro->prev->next = coro->next; 1463 if (coro->prev) coro->prev->next = coro->next;
1455 if (coro == coro_first) coro_first = coro->next; 1464 if (coro == coro_first) coro_first = coro->next;
1456 1465
1603api_is_ready (pTHX_ SV *coro_sv) 1612api_is_ready (pTHX_ SV *coro_sv)
1604{ 1613{
1605 return !!(SvSTATE (coro_sv)->flags & CF_READY); 1614 return !!(SvSTATE (coro_sv)->flags & CF_READY);
1606} 1615}
1607 1616
1617/* expects to own a reference to next->hv */
1608INLINE void 1618INLINE void
1619prepare_schedule_to (pTHX_ struct coro_transfer_args *ta, struct coro *next)
1620{
1621 SV *prev_sv = SvRV (coro_current);
1622
1623 ta->prev = SvSTATE_hv (prev_sv);
1624 ta->next = next;
1625
1626 TRANSFER_CHECK (*ta);
1627
1628 SvRV_set (coro_current, (SV *)next->hv);
1629
1630 free_coro_mortal (aTHX);
1631 coro_mortal = prev_sv;
1632}
1633
1634static void
1609prepare_schedule (pTHX_ struct coro_transfer_args *ta) 1635prepare_schedule (pTHX_ struct coro_transfer_args *ta)
1610{ 1636{
1611 SV *prev_sv, *next_sv;
1612
1613 for (;;) 1637 for (;;)
1614 { 1638 {
1615 next_sv = coro_deq (aTHX); 1639 SV *next_sv = coro_deq (aTHX);
1616 1640
1617 /* nothing to schedule: call the idle handler */
1618 if (expect_false (!next_sv)) 1641 if (expect_true (next_sv))
1619 { 1642 {
1643 struct coro *next = SvSTATE_hv (next_sv);
1644
1645 /* cannot transfer to destroyed coros, skip and look for next */
1646 if (expect_false (next->flags & CF_DESTROYED))
1647 SvREFCNT_dec (next_sv); /* coro_nready has already been taken care of by destroy */
1648 else
1649 {
1650 next->flags &= ~CF_READY;
1651 --coro_nready;
1652
1653 return prepare_schedule_to (aTHX_ ta, next);
1654 }
1655 }
1656 else
1657 {
1658 /* nothing to schedule: call the idle handler */
1620 dSP; 1659 dSP;
1621 1660
1622 ENTER; 1661 ENTER;
1623 SAVETMPS; 1662 SAVETMPS;
1624 1663
1626 PUTBACK; 1665 PUTBACK;
1627 call_sv (get_sv ("Coro::idle", FALSE), G_VOID | G_DISCARD); 1666 call_sv (get_sv ("Coro::idle", FALSE), G_VOID | G_DISCARD);
1628 1667
1629 FREETMPS; 1668 FREETMPS;
1630 LEAVE; 1669 LEAVE;
1631 continue;
1632 } 1670 }
1633
1634 ta->next = SvSTATE_hv (next_sv);
1635
1636 /* cannot transfer to destroyed coros, skip and look for next */
1637 if (expect_false (ta->next->flags & CF_DESTROYED))
1638 {
1639 SvREFCNT_dec (next_sv);
1640 /* coro_nready has already been taken care of by destroy */
1641 continue;
1642 }
1643
1644 --coro_nready;
1645 break;
1646 } 1671 }
1647
1648 /* free this only after the transfer */
1649 prev_sv = SvRV (coro_current);
1650 ta->prev = SvSTATE_hv (prev_sv);
1651 TRANSFER_CHECK (*ta);
1652 assert (("FATAL: next coroutine isn't marked as ready in Coro (please report)", ta->next->flags & CF_READY));
1653 ta->next->flags &= ~CF_READY;
1654 SvRV_set (coro_current, next_sv);
1655
1656 free_coro_mortal (aTHX);
1657 coro_mortal = prev_sv;
1658} 1672}
1659 1673
1660INLINE void 1674INLINE void
1661prepare_cede (pTHX_ struct coro_transfer_args *ta) 1675prepare_cede (pTHX_ struct coro_transfer_args *ta)
1662{ 1676{
1683{ 1697{
1684 struct coro_transfer_args ta; 1698 struct coro_transfer_args ta;
1685 1699
1686 prepare_schedule (aTHX_ &ta); 1700 prepare_schedule (aTHX_ &ta);
1687 TRANSFER (ta, 1); 1701 TRANSFER (ta, 1);
1702}
1703
1704static void
1705api_schedule_to (pTHX_ SV *coro_sv)
1706{
1707 struct coro_transfer_args ta;
1708 struct coro *next = SvSTATE (coro_sv);
1709
1710 SvREFCNT_inc_NN (coro_sv);
1711 prepare_schedule_to (aTHX_ &ta, next);
1688} 1712}
1689 1713
1690static int 1714static int
1691api_cede (pTHX) 1715api_cede (pTHX)
1692{ 1716{
1742 coro->slot->runops = RUNOPS_DEFAULT; 1766 coro->slot->runops = RUNOPS_DEFAULT;
1743 } 1767 }
1744} 1768}
1745 1769
1746/*****************************************************************************/ 1770/*****************************************************************************/
1771/* async pool handler */
1772
1773static int
1774slf_check_pool_handler (pTHX_ struct CoroSLF *frame)
1775{
1776 HV *hv = (HV *)SvRV (coro_current);
1777 struct coro *coro = (struct coro *)frame->data;
1778
1779 if (!coro->invoke_cb)
1780 return 1; /* loop till we have invoke */
1781 else
1782 {
1783 hv_store (hv, "desc", sizeof ("desc") - 1,
1784 newSVpvn ("[async_pool]", sizeof ("[async_pool]") - 1), 0);
1785
1786 coro->saved_deffh = SvREFCNT_inc_NN ((SV *)PL_defoutgv);
1787
1788 {
1789 dSP;
1790 XPUSHs (sv_2mortal (coro->invoke_cb)); coro->invoke_cb = 0;
1791 PUTBACK;
1792 }
1793
1794 SvREFCNT_dec (GvAV (PL_defgv));
1795 GvAV (PL_defgv) = coro->invoke_av;
1796 coro->invoke_av = 0;
1797
1798 return 0;
1799 }
1800}
1801
1802static void
1803slf_init_pool_handler (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1804{
1805 HV *hv = (HV *)SvRV (coro_current);
1806 struct coro *coro = SvSTATE_hv ((SV *)hv);
1807
1808 if (expect_true (coro->saved_deffh))
1809 {
1810 /* subsequent iteration */
1811 SvREFCNT_dec ((SV *)PL_defoutgv); PL_defoutgv = (GV *)coro->saved_deffh;
1812 coro->saved_deffh = 0;
1813
1814 if (coro_rss (aTHX_ coro) > SvUV (sv_pool_rss)
1815 || av_len (av_async_pool) + 1 >= SvIV (sv_pool_size))
1816 {
1817 coro->invoke_cb = SvREFCNT_inc_NN ((SV *)cv_coro_terminate);
1818 coro->invoke_av = newAV ();
1819
1820 frame->prepare = prepare_nop;
1821 }
1822 else
1823 {
1824 av_clear (GvAV (PL_defgv));
1825 hv_store (hv, "desc", sizeof ("desc") - 1, SvREFCNT_inc_NN (sv_async_pool_idle), 0);
1826
1827 coro->prio = 0;
1828
1829 if (coro->cctx && (coro->cctx->flags & CC_TRACE))
1830 api_trace (aTHX_ coro_current, 0);
1831
1832 frame->prepare = prepare_schedule;
1833 av_push (av_async_pool, SvREFCNT_inc (hv));
1834 }
1835 }
1836 else
1837 {
1838 /* first iteration, simply fall through */
1839 frame->prepare = prepare_nop;
1840 }
1841
1842 frame->check = slf_check_pool_handler;
1843 frame->data = (void *)coro;
1844}
1845
1846/*****************************************************************************/
1747/* rouse callback */ 1847/* rouse callback */
1748 1848
1749#define CORO_MAGIC_type_rouse PERL_MAGIC_ext 1849#define CORO_MAGIC_type_rouse PERL_MAGIC_ext
1750 1850
1751static void 1851static void
1755 SV *data = (SV *)GENSUB_ARG; 1855 SV *data = (SV *)GENSUB_ARG;
1756 1856
1757 if (SvTYPE (SvRV (data)) != SVt_PVAV) 1857 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1758 { 1858 {
1759 /* first call, set args */ 1859 /* first call, set args */
1760 int i;
1761 AV *av = newAV (); 1860 AV *av = newAV ();
1762 SV *coro = SvRV (data); 1861 SV *coro = SvRV (data);
1763 1862
1764 SvRV_set (data, (SV *)av); 1863 SvRV_set (data, (SV *)av);
1765 api_ready (aTHX_ coro); 1864 api_ready (aTHX_ coro);
1906static void 2005static void
1907slf_init_schedule (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2006slf_init_schedule (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1908{ 2007{
1909 frame->prepare = prepare_schedule; 2008 frame->prepare = prepare_schedule;
1910 frame->check = slf_check_nop; 2009 frame->check = slf_check_nop;
2010}
2011
2012static void
2013slf_prepare_schedule_to (pTHX_ struct coro_transfer_args *ta)
2014{
2015 struct coro *next = (struct coro *)slf_frame.data;
2016
2017 SvREFCNT_inc_NN (next->hv);
2018 prepare_schedule_to (aTHX_ ta, next);
2019}
2020
2021static void
2022slf_init_schedule_to (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2023{
2024 if (!items)
2025 croak ("Coro::schedule_to expects a coroutine argument, caught");
2026
2027 frame->data = (void *)SvSTATE (arg [0]);
2028 frame->prepare = slf_prepare_schedule_to;
2029 frame->check = slf_check_nop;
2030}
2031
2032static void
2033slf_init_cede_to (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2034{
2035 api_ready (aTHX_ SvRV (coro_current));
2036
2037 slf_init_schedule_to (aTHX_ frame, cv, arg, items);
1911} 2038}
1912 2039
1913static void 2040static void
1914slf_init_cede (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2041slf_init_cede (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1915{ 2042{
2294{ 2421{
2295 if (items >= 2) 2422 if (items >= 2)
2296 { 2423 {
2297 /* callback form */ 2424 /* callback form */
2298 AV *av = (AV *)SvRV (arg [0]); 2425 AV *av = (AV *)SvRV (arg [0]);
2299 CV *cb_cv = coro_sv_2cv (arg [1]); 2426 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]);
2300 2427
2301 av_push (av, (SV *)SvREFCNT_inc_NN (cb_cv)); 2428 av_push (av, (SV *)SvREFCNT_inc_NN (cb_cv));
2302 2429
2303 if (SvIVX (AvARRAY (av)[0]) > 0) 2430 if (SvIVX (AvARRAY (av)[0]) > 0)
2304 coro_semaphore_adjust (aTHX_ av, 0); 2431 coro_semaphore_adjust (aTHX_ av, 0);
2390 dXSARGS; 2517 dXSARGS;
2391 AV *state = (AV *)GENSUB_ARG; 2518 AV *state = (AV *)GENSUB_ARG;
2392 SV *coro = av_pop (state); 2519 SV *coro = av_pop (state);
2393 SV *data_sv = newSV (sizeof (struct io_state)); 2520 SV *data_sv = newSV (sizeof (struct io_state));
2394 2521
2395 av_extend (state, items); 2522 av_extend (state, items - 1);
2396 2523
2397 sv_upgrade (data_sv, SVt_PV); 2524 sv_upgrade (data_sv, SVt_PV);
2398 SvCUR_set (data_sv, sizeof (struct io_state)); 2525 SvCUR_set (data_sv, sizeof (struct io_state));
2399 SvPOK_only (data_sv); 2526 SvPOK_only (data_sv);
2400 2527
2610 assert (("PRIO_NORMAL must be 0", !PRIO_NORMAL)); 2737 assert (("PRIO_NORMAL must be 0", !PRIO_NORMAL));
2611} 2738}
2612 2739
2613SV * 2740SV *
2614new (char *klass, ...) 2741new (char *klass, ...)
2742 ALIAS:
2743 Coro::new = 1
2615 CODE: 2744 CODE:
2616{ 2745{
2617 struct coro *coro; 2746 struct coro *coro;
2618 MAGIC *mg; 2747 MAGIC *mg;
2619 HV *hv; 2748 HV *hv;
2620 CV *cb; 2749 CV *cb;
2621 int i; 2750 int i;
2622 2751
2623 if (items > 1) 2752 if (items > 1)
2624 { 2753 {
2625 cb = coro_sv_2cv (ST (1)); 2754 cb = coro_sv_2cv (aTHX_ ST (1));
2626 2755
2756 if (!ix)
2757 {
2627 if (CvISXSUB (cb)) 2758 if (CvISXSUB (cb))
2628 croak ("Coro::State doesn't support XS functions as coroutine start, caught"); 2759 croak ("Coro::State doesn't support XS functions as coroutine start, caught");
2629 2760
2630 if (!CvROOT (cb)) 2761 if (!CvROOT (cb))
2631 croak ("Coro::State doesn't support autoloaded or undefined functions as coroutine start, caught"); 2762 croak ("Coro::State doesn't support autoloaded or undefined functions as coroutine start, caught");
2763 }
2632 } 2764 }
2633 2765
2634 Newz (0, coro, 1, struct coro); 2766 Newz (0, coro, 1, struct coro);
2635 coro->args = newAV (); 2767 coro->args = newAV ();
2636 coro->flags = CF_NEW; 2768 coro->flags = CF_NEW;
2644 mg->mg_flags |= MGf_DUP; 2776 mg->mg_flags |= MGf_DUP;
2645 RETVAL = sv_bless (newRV_noinc ((SV *)hv), gv_stashpv (klass, 1)); 2777 RETVAL = sv_bless (newRV_noinc ((SV *)hv), gv_stashpv (klass, 1));
2646 2778
2647 if (items > 1) 2779 if (items > 1)
2648 { 2780 {
2649 coro->startcv = SvREFCNT_inc_NN (cb);
2650
2651 av_extend (coro->args, items - 1); 2781 av_extend (coro->args, items - 1 + ix - 1);
2782
2783 if (ix)
2784 {
2652 av_push (coro->args, SvREFCNT_inc (cb)); 2785 av_push (coro->args, SvREFCNT_inc_NN ((SV *)cb));
2786 cb = cv_coro_run;
2787 }
2788
2789 coro->startcv = (CV *)SvREFCNT_inc_NN ((SV *)cb);
2790
2653 for (i = 2; i < items; i++) 2791 for (i = 2; i < items; i++)
2654 av_push (coro->args, newSVsv (ST (i))); 2792 av_push (coro->args, newSVsv (ST (i)));
2655 } 2793 }
2656} 2794}
2657 OUTPUT: 2795 OUTPUT:
2858 2996
2859BOOT: 2997BOOT:
2860{ 2998{
2861 int i; 2999 int i;
2862 3000
2863 av_async_pool = coro_get_av (aTHX_ "Coro::async_pool", TRUE); 3001 av_async_pool = coro_get_av (aTHX_ "Coro::async_pool", TRUE);
2864 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE); 3002 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE);
2865 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE); 3003 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE);
2866 3004 cv_coro_run = get_cv ( "Coro::_terminate", GV_ADD);
3005 cv_coro_terminate = get_cv ( "Coro::terminate" , GV_ADD);
2867 coro_current = coro_get_sv (aTHX_ "Coro::current", FALSE); 3006 coro_current = coro_get_sv (aTHX_ "Coro::current" , FALSE); SvREADONLY_on (coro_current);
2868 SvREADONLY_on (coro_current); 3007
3008 sv_async_pool_idle = newSVpv ("[async pool idle]", 0); SvREADONLY_on (sv_async_pool_idle);
3009 sv_Coro = newSVpv ("Coro", 0); SvREADONLY_on (sv_Coro);
3010 cv_pool_handler = get_cv ("Coro::pool_handler", GV_ADD); SvREADONLY_on (cv_pool_handler);
3011 cv_coro_state_new = get_cv ("Coro::State::new", 0); SvREADONLY_on (cv_coro_state_new);
2869 3012
2870 coro_stash = gv_stashpv ("Coro", TRUE); 3013 coro_stash = gv_stashpv ("Coro", TRUE);
2871 3014
2872 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (PRIO_MAX)); 3015 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (PRIO_MAX));
2873 newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (PRIO_HIGH)); 3016 newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (PRIO_HIGH));
2881 3024
2882 { 3025 {
2883 SV *sv = coro_get_sv (aTHX_ "Coro::API", TRUE); 3026 SV *sv = coro_get_sv (aTHX_ "Coro::API", TRUE);
2884 3027
2885 coroapi.schedule = api_schedule; 3028 coroapi.schedule = api_schedule;
3029 coroapi.schedule_to = api_schedule_to;
2886 coroapi.cede = api_cede; 3030 coroapi.cede = api_cede;
2887 coroapi.cede_notself = api_cede_notself; 3031 coroapi.cede_notself = api_cede_notself;
2888 coroapi.ready = api_ready; 3032 coroapi.ready = api_ready;
2889 coroapi.is_ready = api_is_ready; 3033 coroapi.is_ready = api_is_ready;
2890 coroapi.nready = coro_nready; 3034 coroapi.nready = coro_nready;
2898 3042
2899void 3043void
2900schedule (...) 3044schedule (...)
2901 CODE: 3045 CODE:
2902 CORO_EXECUTE_SLF_XS (slf_init_schedule); 3046 CORO_EXECUTE_SLF_XS (slf_init_schedule);
3047
3048void
3049schedule_to (...)
3050 CODE:
3051 CORO_EXECUTE_SLF_XS (slf_init_schedule_to);
3052
3053void
3054cede_to (...)
3055 CODE:
3056 CORO_EXECUTE_SLF_XS (slf_init_cede_to);
2903 3057
2904void 3058void
2905cede (...) 3059cede (...)
2906 CODE: 3060 CODE:
2907 CORO_EXECUTE_SLF_XS (slf_init_cede); 3061 CORO_EXECUTE_SLF_XS (slf_init_cede);
2962 CODE: 3116 CODE:
2963 RETVAL = coro_nready; 3117 RETVAL = coro_nready;
2964 OUTPUT: 3118 OUTPUT:
2965 RETVAL 3119 RETVAL
2966 3120
2967# for async_pool speedup
2968void 3121void
2969_pool_1 (SV *cb) 3122_pool_handler (...)
2970 CODE: 3123 CODE:
2971{ 3124 CORO_EXECUTE_SLF_XS (slf_init_pool_handler);
2972 HV *hv = (HV *)SvRV (coro_current);
2973 struct coro *coro = SvSTATE_hv ((SV *)hv);
2974 AV *defav = GvAV (PL_defgv);
2975 SV *invoke = hv_delete (hv, "_invoke", sizeof ("_invoke") - 1, 0);
2976 AV *invoke_av;
2977 int i, len;
2978 3125
2979 if (!invoke) 3126void
3127async_pool (SV *cv, ...)
3128 PROTOTYPE: &@
3129 PPCODE:
3130{
3131 HV *hv = (HV *)av_pop (av_async_pool);
3132 AV *av = newAV ();
3133 SV *cb = ST (0);
3134 int i;
3135
3136 av_extend (av, items - 2);
3137 for (i = 1; i < items; ++i)
3138 av_push (av, SvREFCNT_inc_NN (ST (i)));
3139
3140 if ((SV *)hv == &PL_sv_undef)
2980 { 3141 {
2981 SV *old = PL_diehook; 3142 PUSHMARK (SP);
2982 PL_diehook = 0; 3143 EXTEND (SP, 2);
2983 SvREFCNT_dec (old); 3144 PUSHs (sv_Coro);
2984 croak ("\3async_pool terminate\2\n"); 3145 PUSHs ((SV *)cv_pool_handler);
3146 PUTBACK;
3147 call_sv ((SV *)cv_coro_state_new, G_SCALAR);
3148 SPAGAIN;
3149
3150 hv = (HV *)SvREFCNT_inc_NN (SvRV (POPs));
2985 } 3151 }
2986 3152
2987 SvREFCNT_dec (coro->saved_deffh);
2988 coro->saved_deffh = SvREFCNT_inc_NN ((SV *)PL_defoutgv);
2989
2990 hv_store (hv, "desc", sizeof ("desc") - 1,
2991 newSVpvn ("[async_pool]", sizeof ("[async_pool]") - 1), 0);
2992
2993 invoke_av = (AV *)SvRV (invoke);
2994 len = av_len (invoke_av);
2995
2996 sv_setsv (cb, AvARRAY (invoke_av)[0]);
2997
2998 if (len > 0)
2999 { 3153 {
3000 av_fill (defav, len - 1); 3154 struct coro *coro = SvSTATE_hv (hv);
3001 for (i = 0; i < len; ++i) 3155
3002 av_store (defav, i, SvREFCNT_inc_NN (AvARRAY (invoke_av)[i + 1])); 3156 assert (!coro->invoke_cb);
3157 assert (!coro->invoke_av);
3158 coro->invoke_cb = SvREFCNT_inc (cb);
3159 coro->invoke_av = av;
3003 } 3160 }
3004}
3005 3161
3006void 3162 api_ready (aTHX_ (SV *)hv);
3007_pool_2 (SV *cb)
3008 CODE:
3009{
3010 HV *hv = (HV *)SvRV (coro_current);
3011 struct coro *coro = SvSTATE_hv ((SV *)hv);
3012 3163
3013 sv_setsv (cb, &PL_sv_undef); 3164 if (GIMME_V != G_VOID)
3014 3165 XPUSHs (sv_2mortal (newRV_noinc ((SV *)hv)));
3015 SvREFCNT_dec ((SV *)PL_defoutgv); PL_defoutgv = (GV *)coro->saved_deffh; 3166 else
3016 coro->saved_deffh = 0;
3017
3018 if (coro_rss (aTHX_ coro) > SvUV (sv_pool_rss)
3019 || av_len (av_async_pool) + 1 >= SvIV (sv_pool_size))
3020 {
3021 SV *old = PL_diehook;
3022 PL_diehook = 0;
3023 SvREFCNT_dec (old); 3167 SvREFCNT_dec (hv);
3024 croak ("\3async_pool terminate\2\n");
3025 }
3026
3027 av_clear (GvAV (PL_defgv));
3028 hv_store (hv, "desc", sizeof ("desc") - 1,
3029 newSVpvn ("[async_pool idle]", sizeof ("[async_pool idle]") - 1), 0);
3030
3031 coro->prio = 0;
3032
3033 if (coro->cctx && (coro->cctx->flags & CC_TRACE))
3034 api_trace (aTHX_ coro_current, 0);
3035
3036 av_push (av_async_pool, newSVsv (coro_current));
3037} 3168}
3038 3169
3039SV * 3170SV *
3040rouse_cb () 3171rouse_cb ()
3041 PROTOTYPE: 3172 PROTOTYPE:
3215 3346
3216void 3347void
3217_register (char *target, char *proto, SV *req) 3348_register (char *target, char *proto, SV *req)
3218 CODE: 3349 CODE:
3219{ 3350{
3220 CV *req_cv = coro_sv_2cv (req); 3351 CV *req_cv = coro_sv_2cv (aTHX_ req);
3221 /* newXSproto doesn't return the CV on 5.8 */ 3352 /* newXSproto doesn't return the CV on 5.8 */
3222 CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__); 3353 CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__);
3223 sv_setpv ((SV *)slf_cv, proto); 3354 sv_setpv ((SV *)slf_cv, proto);
3224 sv_magicext ((SV *)slf_cv, (SV *)req_cv, CORO_MAGIC_type_aio, 0, 0, 0); 3355 sv_magicext ((SV *)slf_cv, (SV *)req_cv, CORO_MAGIC_type_aio, 0, 0, 0);
3225} 3356}

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines