From 6de4d43db1a18e33f90ef8d6b484ecf6e49b0597 Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Tue, 14 Oct 2025 08:05:16 -0600 Subject: [PATCH] WIP: switch to announcement, remove extract input mess --- .beads/dagx.db | Bin 0 -> 131072 bytes CLAUDE.md | 1 + Cargo.lock | 10 + Cargo.toml | 1 + benches/basic/creation.rs | 2 +- benches/basic/execution.rs | 2 +- dagx-macros/src/lib.rs | 307 +++++++++------------ examples/01_basic.rs | 2 +- examples/02_fan_out.rs | 2 +- examples/03_fan_in.rs | 2 +- examples/04_parallel_computation.rs | 32 ++- examples/circuit_breaker.rs | 2 +- examples/complex_dag.rs | 2 +- examples/conditional_workflow.rs | 2 +- examples/data_pipeline.rs | 2 +- examples/debug_tracing.rs | 2 +- examples/error_handling.rs | 2 +- examples/large_dag.rs | 2 +- examples/parallelism_proof.rs | 2 +- examples/retry_strategies.rs | 2 +- examples/timeout.rs | 2 +- examples/tracing_example.rs | 2 +- src/builder.rs | 6 +- src/builder/tests.rs | 4 +- src/deps.rs | 8 +- src/deps/tests.rs | 4 +- src/extract.rs | 291 -------------------- src/extract/tests.rs | 410 ---------------------------- src/lib.rs | 3 +- src/node.rs | 158 ++++++----- src/node/tests.rs | 157 ++++++----- src/runner.rs | 84 +++--- src/runner/tests.rs | 4 +- src/task.rs | 324 ++++++++++++++++------ src/types.rs | 6 +- src/types/tests.rs | 4 +- tests/arc_tests.rs | 2 +- tests/boundaries/zero_tasks.rs | 2 +- tests/common/mod.rs | 2 +- tests/cycle_prevention.rs | 2 +- tests/execution/integration.rs | 2 +- tests/parallelism/timing.rs | 2 +- tests/runtimes/async_std.rs | 2 +- tests/runtimes/compatibility.rs | 2 +- tests/runtimes/quirky.rs | 4 +- tests/runtimes/smol.rs | 2 +- tests/tracing/with_tracing.rs | 2 +- tests/tracing/without_tracing.rs | 2 +- 48 files changed, 685 insertions(+), 1187 deletions(-) create mode 100644 .beads/dagx.db create mode 100644 CLAUDE.md delete mode 100644 src/extract.rs delete mode 100644 src/extract/tests.rs diff --git a/.beads/dagx.db b/.beads/dagx.db new file mode 100644 index 0000000000000000000000000000000000000000..606320d37cb92403df04829389e7968b506baaf2 GIT binary patch literal 131072 zcmeI5e{3Ateb{%o+@ksNvT6 zBZkw~DIC{<`$xa;y_wyao!KRIC*AojUqV{WzIpFEZ$97e`@Z+S^Ehwy+9l0W$%bw; z6^rCTM?xbbp`Rxt6bg;PfBM_`IR*zb&G`vC2Sa|3hBBWyHU=5O!cS?olfqA@wx^z) z{CDv`n%s-OIR5R}cgO!_?4LwWkNw%$KNveX`j1C{82OvgPmX*u@^43~;olgU5q>g6 z`up?L?o=u^olb{$S1o0|p=zetQL8n*)3Q{<^nQC{X{EBbRv~MPAH7r|-g9Jjg1}GP zP&c$YWUcbz8o9c1;qv0j4RWz^V~(9{D5h1_>g2-mTIGex3R%9gMwYK%xJf8M_F#Iey3X3 z3~PrMGpkL-xT)5wisepC**A4z`F!O?GPn%6vP|6NX1mhTyVb{H)2B~|_rnaKJD2-- z+Jn!1#)il51^@-)>t2wY2Q92!xLolQT)MupQdwTB(zB~;i2^705d8S)AlWRB zOKjJOt_soVvuDFM7)86mn(D2N+N!C2UypdW_MMzKR$3Le$s^e*n>R6xS_&ujdb3QC%5OTMYyDn{xS0~NOzFn_(c4}Db<+o53 z`R~N56S3(f=zC6dY1Y-Y3cXmXrm1F?e=(~%r1X*}eR$nxU4;QJm^u=jd*Oj z3>S|DUz}OIyZFTL>vkKvyG$2ydcPcF{oOs8Y1y60eLU{bxckf@^v<(WUUbC3Iz+Ew z_}gfA7;$F3q1SGjXWo7bb6V%R-4%JOFu(*oRevC+7+}uci`+*~K0dT3ciATNaMNi{g9eV4W;OcyGqUut%F@NzhT7V+wq{v|H2Jybh@8#Z z7F$=%nxVBVO>cRdU$2?j4GeOEf)2=})M{$mQeaR3;~5y#8=7)JiI#4u<^g$ZfFAeI ztDg>z)~=(YufcN<$wt!6`QhhCmMkuxhdB+m63QoU(_zX87$@{DZYkzX`{IrrvYX*; zSv4&Urob?&1Iy{q8?`4zZ{KHi+Azds!^Qg_QPdi`=`Ctk_YP>Rr8KHAy<>MVkv=;z ztaIeXYERJ22%kDXIhh#!N@y~4DHQ(~$^Q`O1HoB#@OzpNx0I{XDHi_2SFkl*LbZS<| zP#4HX=ap9=eGas&w`rz1((FLnrDkuJp~CEJWLvYgNKG*|^?91ai>WZjoL^UfoKuA7 z>xooab)Q(bEP6>8>SxHR*3=q`K`fnQGUt?53nt<R!TBtxh{bZG3bC zCJ1mhQdW}Z#q7M8C)v`1Tw2JBnL@sp%a)7z(tNhCkj)+<%bqd;T%X>%>2SSv&EcB8 z&}=t&ZC>t})`kXd1!6fcGP$ib^fs&Jwr-WybRNSfj|`%mlQUvL%9e_9aR6jrj{QvJkyK^3nn>&|rnv2{ za@)Jsm!zw|jF&mzm*wq8Y z)WCRoSug+~EgjU@+EfkFp>1d0z7MBbrmI>`)ic~f z1?wVPUdZM$g;G`ogO>s=J|{0^^O<}(m(Lf)VxYbPvrHZdeJN$`skBXuKkD)|rRO2s zrL=Y^5k+fJ7en0!^FGj$sUJCKKrl^fLQ_$+1~gN1aRa<5fmD{Ac@tJ>j%vlA>f8Zi zbxGvjlsgP6-Gqz~KZQXNcpRPPg`k^S&ycmP-Dl%RQ@e4fCpg99u8NkI8miLjvWLtAHkk+o&$oy{#B%%`T;RqEQPpNY`+=kp*^F;gze za)BZpVE_Isdt2j2ZJsI4)3*PgHJ)Im(8L-Y800V=+32A&fffd4OJM5nHSacngwE0= z;KQ~QXs$Zdsy*Pf?RUMi>E)QqE=W=)FXhT&IV(#+9P_z_Tp?4=mI~!!Hk&_mmi^;n z)Yrto*ECbyBy865N~Z;u1O2*vYqVvncNFU6%}>A(ZDWTy5^i#B7{;)+@8lrV)f0so z^VKb|5h8&fR-i$fRB+F(sf~hL_KVAqzKrwpUE#el-6YzhLzj~wOVr!jZ9sQ7HKH1Z z4(^gS#tgZntsB(NYaIg|Idy%u43n507?GOhMyJu(alDG(zNt$oXQZ55&Wmz6HyC=c zDCUZCwkRGt%kHt*kreG@6REh{qZFw}0T&N8L~7JM6H{$$5a^HS;D-8c>fD)=p&hWv zJ)+N|BzYlM%!r~?DrC#qz-me51yRh1Ww9ugrJ{XvdoPENrnEhc*2~i~iPzb{(st6I z9>-6^;0P+z--TReeUb@I1R4nR)?g_X^?A_8(CGpjuu#LOLvLrmAXw%7m3a=l4)j~J zFXv-~hGtr{qk>TZ?NZ#6QpiJDaH(tn!TLNXOD)+MJ%KftoraYm%Q|#a;9vc{(W#-` z3XG-1Ot~P-qFj^%Ns8h^wvdsF1u35^W&@jVV3w(`hi7=(?DN(SuW3?8KJu0fJ+LiI zeLeI`DQ))}>u=nH@VspSm`2j{A~rLW7%%R0ai$`BG8J1vYZ2NS&`#kV<9n z(6)l%Ipbq|6xTNaVS~8cPUNV4_8HIgvY0JP;MB8$ zV@d8lC9x>y3MJ8TpC?9+ru5y4&wWaEd!V*T_z)PHgguVt17ql5^_s>8nvv1!kFm8UCbj zFC_d}_;ul)@B`ty!k-Io3V$HHA$&>r+FpD?sJKq2{5*}t1o$Bf z1n|R=7N7rrEd=xZ?+AY({8ac`;eSIM;CsS<5WX&a?S5|p^CJNyfCP{L5t`6#fe4{xlNs4dFitZ$6wFqf{h-1dsp{Kmter2_OL^fCP{L5X8^kag4KoB57g8&H{1W1IV^ka;EjM9$?{TPWv5^*;FPkuZEGyX3NpB5CM zEF4XKKl#<ur=U8yQT&C|uhNFY>^}Reh z5|A(G?SLa6h^)rF>Y#6+_q{bW0a}bFz8DgIUwBn0CI2@0uakOmF7bUx!XG4n1dsp{ zKmter2_OL^fCP{L5_rc6EXJd$RBAQF-q}`SpEBQ4(>uj;roQ=NS<2;Q{uX$18+Lax z7sp*9WF{gXBm!o!lAMwAB|jct`?*J8{kLL4HZn9s{%*4ZS)7BT7)kvoV*K3^*2p&niq5toQO6EW;YU{3<6 z3}QStGU5|4)Bz&w{R21(`usmGR73Dr|6dXQ zoA4*X?+U*u>f^A01`j~NB{{S0VIF~kN^@u0!RP}JTL;?E&EEn{r}|N{(o|B|35i!|G!*s|G#{1 z|G#{1|3Aq!|NjB(0r*|m0q_lo1bj*O+ylD_6odqj01`j~NB{{S0VIF~kN^@u0!RP} z3?W&-Ld2Qg8k*vc3NJUp4gR|59)MFZSmDBK`aS(2=2)z>`P-2_OL^ zfCP{L5E2iMzm#Zv2J zO)+m0%YYp*h|;Q)rcyI>V(G+CZ*??7B})yxr4qeGu5_$+$0F8FTXpTRA?6oEX(1== zpE?3pOQ-IN&xP!(nU-R8%xY~*X>B@JeRW2w&n(RF?IC97W-QHWsF1*GZtkGRZBIR8koLD&CfiiGxIm+2-j-&?oDo7brtNQuC{8L zO7l=d;!+tZE@O_hFb5B`FtgLJGGtjNO&wZVg&Q6>ch(|LLG>Sk;raGFJ=pO4MGk3r zuzOCf^im=oO3i~Oy1~s(?$X++D)qYK92V&zkhe+hED+`=WWsl|;otIJlVZE`yfpRT zK# zzvQWZko>)iW8m^mq(ZMR*`1?o_YJ%+cK3?1%h#=RT4Y0ODGg$&re!ksq25uN?S^Xl zJliK!L)+M4T^P^ZhAz`ngly{u_`418Y~<>eVyZ+UymzM34)16z7&1VCVclDKAzRMC z=1k?hD3=e{y#*HO>&DoQ()(khoaWu4O*21`X5K~99%Hohx|%Ue>g1W5rgxL+^>D5} z-`jK80}PLs?wHCZ?IDOGl=gqaq$-ahz(zK#@hqoOD_00&|b)94^Tqj*t}l}G}HdI zFjvC8E4C8Efl9zJ>;s)V`3I!Ky&9;?U!pEc_@0|L_M1AOR$R z1dsp{Kmter2_OL^fCP{L5_lgGmf>;IoknsooaP2m@lznnB-7r;Ld zt_jb>UI3p_s|2_G~$uB2=JGquGFb{Jsy~zD;^6>m&%X& z)8P-M^Z7>t({s|%fOPuDyGH`kvjxGQ&i>{y8JI4~i2>vP1 z(Ko))mnUWl`J!Bqily9qwy==R&QEl+XMNdoaz-pjc}dQe@;v)oJd)zl%f8%FF;gze zO7r|F7KJ|F1OO z-4(Fd+yAc+?WW$}{(piJnoQ1&36BWhO8(R2UnjpN{41d>{7TYDemptzuE4;HA^{|T z1dsp{Kmter2_OL^fCP}hyF%bhWYxU^|i)$6~i3^oqAntUsxH_>~xl~zNBUX_s?az<~ftuqsZEH0lX%U9M&<;4rDYpY}yk_fToh=fV0tB|*@wrZMck~+)q@hcaW z9qH>0y>?TrlPk-lp5b}Ts@|$Sy7YL zk^aznI>PL`fGW)L`1Fx)rAy>|<@v?y zm)1yjBHejBHobT{ynB>YUaQ}!)-}U&45ivs4A4bYvD~RCyIvQT&sSa~gUc`#x}?o^ zrKOA0(dmn)2O)G{tSY*{J-%CgEH-`mba+3^@VU>qkEcE8+-GcXoF>2tM%S$=IS;L4 z?ZV}XpWf2-m6gi!T9uw%U0b|-l~aHI%1Y(J3(HguWY)P%nygfwhc>gkR9UqReO9Zd zDWUVw$e;l(Ev_ytp07-#&mE6WFP$8Orm0#=U9ps^=o~$^bu2c0@??1T8AilOc77lC zkZ@9M3OCiA0RV0*jgHFC(dJCeW+Hv|(dhI?QiG~i(_0(bW>w_BkF7rvn@**|yHB%f z@kIXlsHaLi?X6X)dNev+Oodsbw)xNJsJgE9k&bE}M zItL}wdmBjx`|cSglUIy?3LXhO&1nNY*WmIUWbyhMxML`G8LXSzv{QAjLGW`-WR-bY z)|MP|;b)bar5i!T^hUkfGcumJrEdGuYI?H?I%OhwuAX|{`NnxQ!R$fn8eBtJR~u%P z{o=<-pFOzqPGPjf}QT~GJwL~MEq`koVAn(m=Nwb%E%lP7(6-Dh2eC~5bp*`7S; z6w{4(Y`P2=j|E?xS-iXW#PI8O8@szqms`3xKD57cjfkMXbA5~Nmt(BIyC*Y^yU)3g z$35D2pE-mL@a)bYE1(hk8+UgI;S8(Ze$#ZC^}5|9dF^z7IeM!8K+H10w7nNOd{17D z_VrH8NZh>>Pl)+U%a@TFGf#B)PB+G))45c*(=}ncoMM{VW=rLROEWr1ypzo(+(}B` z7!4NPHZ`QnH&AgN!FcSKZ?9>c;kF1lL7N9;Qff7|Z7DELfvHZF zXxYv1`lzO*HQ5jl%&bFi)Sd+K(~hrEtot8AblNcfW@F3yFI#KqrnjhF-8-PMmeQ!g zbdKG{M4Hb3k4C>368=osO+KFZqr|Pmk*VKG22m6JId_+W{;~_oY{M(7infuONch}PI2kV_2sqM zz8G+tD5ev*t*-CzcxYdAnnja6b08K?rjcO?1oxuF6aKKX!DfvE!p;T_E%$}K4;wV5 z@#dgdbT1BVIMMXOiF{$@%Jr+{qci52uhdi02T?fFz={>ZGaE44qgy ztVy`l(F~O=HT0HBAXs*#W3@XLp)qOK>l?)U0(+0b{;4BywRGyP_*}@o8V_PsYgzB;4TXBK97b!O&fScDD|cx~L3VO41Aw^eeXMXw9NCcUQ72oS3Rlyyb&K*EYT zI5XKW^rl<5sX@TbzSnK7(I8OAI>ZtcI8^TdjT(y&!t-mjW;J%;V*F;{IyCqO-<{yu z%z`vG(~Fuxh7EWG+A&nf(~EXNa*x>>yg`G$G^CIz%c3a5#v}HwBT%FlmG6~|Jgwffi!hheORI3h1P? z{ZtM%Jb#fx8eX=%kSo2Eh=)@1;E8Ty6V@!j z@~nCHZZcrlAYYPSntE{Z-8;&39oj( z2G1u5JV~)k=04OrO0(TiO`m7`1U!4%*kN55&)tSD(^G_O>jwC{4e)H_>Xu@vL?XO* zrqT}YXy_v#C@`#hD?`J~NI4lcy^_m^>)ry3^mSuk&Drw)*eIuYw`kMM52TrQ(X_`H zExoR043j!}=BDZ0WO_ZEtIzlL9QFXic56}9~$HvDtBYy&k`17t4czyQ7BpplLJ-2tm>2~bK z>>o_Mx4Iqo{Qd1m7+62v-<1q)B=GW2*a{Uot8yt@`WoH2D))Ge^(0D{~ulzVu(Eh8A zf{KF_RCJ9mpNGGSg60pa>L5@?V~3{N*r00%Xz+%P!f9lT&QD=t8#Ie`55ouX6Z;~r z>zz#~W{PUag#mnNz+bi&D^?Hh9M68! zaXiI?9M8jI=yMV82)sbvZg_6h_Km_A`MnQ6#;ou4F~|A}gRHO6XMJ=&U)_LpyYRch z1J}@p@L)AWdEw8V%uW*)re3nc*HGL$<$PEsD3u}ZD&|W?DR-!d2w1J0&B$`TR2B;* z$7nm*d>s}Nz&O7NFX`A)Tc!qUi0F({Yi+^=bi;tRa~Pc( zU8A!=Haf4o0_k%EUOLjIndV5d120#hp(?wKZ+<|f+vWJhm;-%TeJ)9`oz) zd=-`|+1F*OKzi3;E3Nvj15)M2e;LvOPa^l|q_L*Ir!BsB+eO#Ae^#9m@S z6w(#UvuqMJ1Dm3ifG8ETD1(h4o~O%G=8pI9|)(wA1tZ{eR0KbjlZRF7iflWQ+BCstt~_EY{CLB3y9ZM+75jxh6FzL4tge)>uV)cvrxdUXHa|z zZlCYh1xxwz`NI+mDB?#Z6+d+f7N^i4K0sJdnu*n}2crU^>_j>6m?|PlmeXZ9idDiR5ee&sEWcyc7ax1-e*|yTs zKr1b|tds^nmDUb4&ZeTZXn#Zfzgsou9JM$a@tNi#Tb={iSb5g9Fe!6%{Q|5KY$`k8 zEnJsii%(77d(g|l%9bry(+IC2fbK`9dEre}YCS{Nw&39kEPQQrAl}7f+F7A%1Ua-U2D-njfo@`$;IxtG>Yy+1oi{-2=9mekj7wKz7 zEYs^BcoH5Prcrq%Y(Ayt8`E zRi>N7J(!vyOSHqZtr!+*xg48n7&@#K;_WK~A>4I?_5`(#0evAX)V4E&d7B-0F@tH+ RMZr5BI~@}714GH?|39O~&9ndj literal 0 HcmV?d00001 diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..d1d8a91 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +We track work in Beads instead of Markdown. Run \`bd quickstart\` to see how. diff --git a/Cargo.lock b/Cargo.lock index aa0bf6e..c7999a2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -32,6 +32,15 @@ version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" +[[package]] +name = "announcement" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e58b38a7f1ae19278fb8b7aa4c3ac3957f0e6389a132522789b1dd2e1912189" +dependencies = [ + "event-listener 5.4.1", +] + [[package]] name = "anstyle" version = "1.0.13" @@ -436,6 +445,7 @@ dependencies = [ name = "dagx" version = "0.3.0" dependencies = [ + "announcement", "async-executor", "async-std", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index a45c129..3ea20e6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ tracing = ["dep:tracing"] unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tarpaulin)', 'cfg(tarpaulin_include)'] } [dependencies] +announcement = "0.1.0" futures = "0.3" parking_lot = "0.12" dagx-macros = { version = "0.3.0", path = "./dagx-macros" } diff --git a/benches/basic/creation.rs b/benches/basic/creation.rs index 94a8c6b..496b5fd 100644 --- a/benches/basic/creation.rs +++ b/benches/basic/creation.rs @@ -1,7 +1,7 @@ //! DAG creation benchmarks use criterion::Criterion; -use dagx::{DagRunner, Task}; +use dagx::DagRunner; use std::hint::black_box; // Simple task for benchmarking diff --git a/benches/basic/execution.rs b/benches/basic/execution.rs index e6d6326..c9087d0 100644 --- a/benches/basic/execution.rs +++ b/benches/basic/execution.rs @@ -1,7 +1,7 @@ //! Basic DAG execution benchmarks use criterion::Criterion; -use dagx::{task_fn, DagRunner, Task}; +use dagx::{task_fn, DagRunner}; pub fn bench_dag_execution(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); diff --git a/dagx-macros/src/lib.rs b/dagx-macros/src/lib.rs index 76c1f1f..12a4dbd 100644 --- a/dagx-macros/src/lib.rs +++ b/dagx-macros/src/lib.rs @@ -1,17 +1,17 @@ //! Procedural macros for dagx //! -//! This crate provides the `#[task]` attribute macro that automatically implements -//! the `Task` trait by deriving Input and Output types from the `run()` method signature. +//! This crate provides the `#[task]` attribute macro that automatically generates +//! an `execute()` method by deriving Input and Output types from the `run()` method signature. use proc_macro::TokenStream; use quote::quote; use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType, Type}; -/// Attribute macro to automatically implement the `Task` trait. +/// Attribute macro to automatically generate the execution logic for a task. /// /// Apply this to an `impl` block containing a `run()` method (sync or async). The macro: -/// - Derives `Input` and `Output` types from the `run()` signature -/// - Automatically implements the `Task` trait +/// - Derives input types from the `run()` signature parameters +/// - Generates an `execute()` method that handles listener extraction and announcement /// - Supports both sync and async run methods /// - Supports stateless (no self) and stateful (&self, &mut self) tasks /// - Handles various input patterns (no inputs, single input, multiple inputs) @@ -25,7 +25,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType /// Unit structs for pure computations. Use **no `self` parameter**: /// /// ```ignore -/// use dagx::{task, Task}; +/// use dagx::task; /// /// struct Add; /// @@ -42,7 +42,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType /// Tasks that read configuration or constant data. Use **`&self`**: /// /// ```ignore -/// use dagx::{task, Task}; +/// use dagx::task; /// /// struct Multiplier { /// factor: i32, @@ -61,7 +61,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType /// Tasks that accumulate or modify state. Use **`&mut self`**: /// /// ```ignore -/// use dagx::{task, Task}; +/// use dagx::task; /// /// struct Counter { /// count: i32, @@ -81,7 +81,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType /// ## No Inputs (Source Tasks) /// /// ```ignore -/// use dagx::{task, Task}; +/// use dagx::task; /// /// struct LoadData { /// value: i32, @@ -93,110 +93,51 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType /// self.value /// } /// } -/// ```ignore +/// ``` /// /// ## Single Input /// /// ```ignore -/// use dagx::{task, Task}; +/// use dagx::task; /// /// struct Double; /// /// #[task] /// impl Double { -/// async fn run(&mut self, input: &i32) -> i32 { +/// async fn run(input: &i32) -> i32 { /// input * 2 /// } /// } -/// ```ignore +/// ``` /// /// ## Multiple Inputs (up to 8) /// /// ```ignore -/// use dagx::{task, Task}; +/// use dagx::task; /// /// struct Combine; /// /// #[task] /// impl Combine { -/// async fn run(&mut self, a: &i32, b: &String, c: &bool) -> String { +/// async fn run(a: &i32, b: &String, c: &bool) -> String { /// format!("{}: {} ({})", b, a, c) /// } /// } -/// ```ignore +/// ``` /// /// # Requirements /// /// - The impl block must contain exactly one `async fn run()` method /// - The `run()` method can be stateless (no self parameter) or stateful (`&mut self`) /// - All input parameters must be references (e.g., `&i32`, not `i32`) -/// - The macro requires `Task` to be in scope: `use dagx::Task;` -/// - For stateless tasks, the struct must implement `Default` (e.g., unit structs) /// /// # Generated Code /// -/// The macro transforms your implementation into a full `Task` trait implementation. -/// -/// For stateless tasks (no self parameter): -/// -/// ```ignore -/// // Your code: -/// #[task] -/// impl Add { -/// async fn run(a: &i32, b: &i32) -> i32 { -/// a + b -/// } -/// } -/// -/// // Generated: -/// impl Task for Add { -/// type Input = (i32, i32); -/// type Output = i32; -/// -/// async fn run(&mut self, input: Self::Input) -> Self::Output { -/// let (a, b) = input; -/// Self::run_impl(&a, &b).await -/// } -/// } -/// -/// impl Add { -/// #[inline] -/// async fn run_impl(a: &i32, b: &i32) -> i32 { -/// a + b -/// } -/// } -/// ```ignore -/// -/// For stateful tasks (with &mut self): -/// -/// ```ignore -/// // Your code: -/// #[task] -/// impl Counter { -/// async fn run(&mut self, inc: &i32) -> i32 { -/// self.count += inc; -/// self.count -/// } -/// } -/// -/// // Generated: -/// impl Task for Counter { -/// type Input = i32; -/// type Output = i32; -/// -/// async fn run(&mut self, input: Self::Input) -> Self::Output { -/// let inc = input; -/// self.run_impl(&inc).await -/// } -/// } -/// -/// impl Counter { -/// async fn run_impl(&mut self, inc: &i32) -> i32 { -/// self.count += inc; -/// self.count -/// } -/// } -/// ```ignore +/// The macro transforms your implementation into an `execute()` method that: +/// - Extracts inputs from announcement listeners +/// - Calls your run() method +/// - Announces the result to dependents +/// - Returns the Arc-wrapped output #[proc_macro_attribute] pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream { let impl_block = parse_macro_input!(item as ItemImpl); @@ -220,7 +161,7 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream { "impl block must contain a run() method\n\n\ Expected signature: fn run(&mut self, ...) -> OutputType\n\ or: async fn run(&mut self, ...) -> OutputType\n\ - The #[task] macro requires a run() method to implement the Task trait.", + The #[task] macro requires a run() method to implement task execution.", ) .to_compile_error() .into(); @@ -287,31 +228,6 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream { ReturnType::Type(_, ty) => ty.clone(), }; - // Build Input type based on parameter count - let input_type = match params.len() { - 0 => quote! { () }, - 1 => { - let (_name, ty) = ¶ms[0]; - quote! { #ty } - } - _ => { - let types: Vec<_> = params.iter().map(|(_, ty)| ty).collect(); - quote! { ( #(#types),* ) } - } - }; - - // Generate parameter destructuring for the wrapper run() method - let (param_destructure, param_refs) = if params.is_empty() { - (quote! { _ }, quote! {}) - } else if params.len() == 1 { - let (name, _) = ¶ms[0]; - (quote! { #name }, quote! { &#name }) - } else { - let names: Vec<_> = params.iter().map(|(name, _)| name).collect(); - let refs: Vec<_> = params.iter().map(|(name, _)| quote! { &#name }).collect(); - (quote! { ( #(#names),* ) }, quote! { #(#refs),* }) - }; - // Clone the run method and rename it to run_impl let mut run_impl_method = run_method.clone(); run_impl_method.sig.ident = syn::Ident::new("run_impl", run_method.sig.ident.span()); @@ -323,81 +239,128 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream { .iter() .any(|arg| matches!(arg, FnArg::Receiver(_))); - // Generate the Task trait implementation based on whether we have self and async/sync - let expanded = if has_self_receiver { - // Stateful task - consumes self but delegates to a method that borrows + // Generate listener extraction code + let listener_extractions: Vec<_> = params + .iter() + .enumerate() + .map(|(i, (name, ty))| { + quote! { + let #name = { + if listeners.len() <= #i { + return Err(format!( + "Expected at least {} listener(s), got {}", + #i + 1, + listeners.len() + )); + } + let listener = listeners[#i] + .downcast_ref::>() + .ok_or_else(|| { + format!("Type mismatch: expected Listener<{}> at index {}", + stringify!(#ty), #i) + })?; + listener.listen().await + .map_err(|_| "Listener closed before receiving value")? + }; + } + }) + .collect(); + + // Generate parameter refs for the run call + let param_refs: Vec<_> = params.iter().map(|(name, _)| quote! { &#name }).collect(); + + // Generate the execute method body based on whether we have self and async/sync + let run_call = if has_self_receiver { if is_async { - // Async with self - quote! { - impl Task for #struct_name { - type Input = #input_type; - type Output = #output_type; - - async fn run(mut self, input: Self::Input) -> Self::Output { - let #param_destructure = input; - self.run_impl(#param_refs).await - } - } - - impl #struct_name { - #run_impl_method - } - } + quote! { self.run_impl(#(#param_refs),*).await } } else { - // Sync with self - wrap in async block - quote! { - impl Task for #struct_name { - type Input = #input_type; - type Output = #output_type; - - async fn run(mut self, input: Self::Input) -> Self::Output { - let #param_destructure = input; - self.run_impl(#param_refs) - } - } - - impl #struct_name { - #run_impl_method - } - } + quote! { self.run_impl(#(#param_refs),*) } } + } else if is_async { + quote! { Self::run_impl(#(#param_refs),*).await } } else { - // Stateless task - if is_async { - // Async stateless - quote! { - impl Task for #struct_name { - type Input = #input_type; - type Output = #output_type; + quote! { Self::run_impl(#(#param_refs),*) } + }; - async fn run(self, input: Self::Input) -> Self::Output { - let #param_destructure = input; - Self::run_impl(#param_refs).await - } - } + // Generate the self parameter for execute method + let self_param = if has_self_receiver { + quote! { mut self } + } else { + quote! { self } + }; - impl #struct_name { - #[inline] - #run_impl_method - } + // Generate the Input type based on parameter count + let input_type = match params.len() { + 0 => quote! { () }, + 1 => { + let (_, ty) = ¶ms[0]; + quote! { #ty } + } + _ => { + let types: Vec<_> = params.iter().map(|(_, ty)| ty).collect(); + quote! { (#(#types,)*) } + } + }; + + // Generate the complete execute method and TaskExecute trait implementation + let expanded = quote! { + impl #struct_name { + #[doc(hidden)] + pub async fn execute( + #self_param, + listeners: Vec>, + announcer: Box, + ) -> Result, String> { + // Extract inputs from listeners + #(#listener_extractions)* + + // Call the user's run method + let output: #output_type = #run_call; + + // Wrap in Arc for sharing + let output_arc = std::sync::Arc::new(output); + + // Announce the result + let announcer_typed = announcer + .downcast::>() + .map_err(|_| "Type mismatch: announcer type does not match task output type")?; + let _ = announcer_typed.announce((*output_arc).clone()); + + // Return Arc-wrapped output + Ok(output_arc as std::sync::Arc) } - } else { - // Sync stateless - quote! { - impl Task for #struct_name { - type Input = #input_type; - type Output = #output_type; - async fn run(self, input: Self::Input) -> Self::Output { - let #param_destructure = input; - Self::run_impl(#param_refs) - } - } + #run_impl_method + } - impl #struct_name { - #[inline] - #run_impl_method - } + // Implement TaskExecute trait for type erasure + impl dagx::TaskExecute for #struct_name { + type Input = #input_type; + type Output = #output_type; + + fn execute( + self, + listeners: Vec>, + announcer: Box, + ) -> std::pin::Pin, String>> + Send>> { + Box::pin(async move { + self.execute(listeners, announcer).await + }) + } + + fn create_announcement_erased() -> (Box, Box) { + let (announcer, announcement) = announcement::Announcement::<#output_type>::new(); + ( + Box::new(announcer) as Box, + Box::new(announcement) as Box, + ) + } + + fn create_listener_erased(announcement: &dyn std::any::Any) -> Box { + let announcement = announcement + .downcast_ref::>() + .expect("Type mismatch: announcement type does not match task output type"); + Box::new(announcement.listener()) as Box } } }; diff --git a/examples/01_basic.rs b/examples/01_basic.rs index f58d42f..66e6b67 100644 --- a/examples/01_basic.rs +++ b/examples/01_basic.rs @@ -30,7 +30,7 @@ //! Result: 5 //! ``` -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; // Step 1: Define a source task (no dependencies) // diff --git a/examples/02_fan_out.rs b/examples/02_fan_out.rs index 838d410..e2d1629 100644 --- a/examples/02_fan_out.rs +++ b/examples/02_fan_out.rs @@ -55,7 +55,7 @@ //! Squared: 100 //! ``` -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; // Source task with a value field struct Source { diff --git a/examples/03_fan_in.rs b/examples/03_fan_in.rs index 8d9d9d0..8ea70a6 100644 --- a/examples/03_fan_in.rs +++ b/examples/03_fan_in.rs @@ -61,7 +61,7 @@ //! Active: true //! ``` -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; struct FetchName; diff --git a/examples/04_parallel_computation.rs b/examples/04_parallel_computation.rs index db03e6b..2f7ace3 100644 --- a/examples/04_parallel_computation.rs +++ b/examples/04_parallel_computation.rs @@ -52,7 +52,7 @@ //! This proves dagx runs tasks on multiple CPU cores simultaneously. //! ``` -use dagx::{task, DagResult, DagRunner, Task}; +use dagx::{task, DagResult, DagRunner}; use std::time::{Duration, Instant}; use tokio::time::sleep; @@ -144,22 +144,32 @@ async fn main() -> DagResult<()> { }; // ============================================================================ - // SEQUENTIAL EXECUTION: Run the same 4 tasks one after another + // SEQUENTIAL EXECUTION: Run tasks one at a time in separate DAG executions // ============================================================================ println!("Running SEQUENTIAL computation (same work, one at a time)..."); let (sequential_time, sequential_result) = { let start = Instant::now(); - // Run tasks sequentially (not using DAG, just raw execution) - let task1 = ComputeSum::new(1, 251, WORK_DELAY); - let task2 = ComputeSum::new(251, 501, WORK_DELAY); - let task3 = ComputeSum::new(501, 751, WORK_DELAY); - let task4 = ComputeSum::new(751, 1001, WORK_DELAY); + // Run each task separately to force sequential execution + let dag1 = DagRunner::new(); + let sum1_handle = dag1.add_task(ComputeSum::new(1, 251, WORK_DELAY)); + dag1.run(|fut| { tokio::spawn(fut); }).await?; + let r1 = dag1.get(sum1_handle)?; - let r1 = task1.run(()).await; - let r2 = task2.run(()).await; - let r3 = task3.run(()).await; - let r4 = task4.run(()).await; + let dag2 = DagRunner::new(); + let sum2_handle = dag2.add_task(ComputeSum::new(251, 501, WORK_DELAY)); + dag2.run(|fut| { tokio::spawn(fut); }).await?; + let r2 = dag2.get(sum2_handle)?; + + let dag3 = DagRunner::new(); + let sum3_handle = dag3.add_task(ComputeSum::new(501, 751, WORK_DELAY)); + dag3.run(|fut| { tokio::spawn(fut); }).await?; + let r3 = dag3.get(sum3_handle)?; + + let dag4 = DagRunner::new(); + let sum4_handle = dag4.add_task(ComputeSum::new(751, 1001, WORK_DELAY)); + dag4.run(|fut| { tokio::spawn(fut); }).await?; + let r4 = dag4.get(sum4_handle)?; let result = r1 + r2 + r3 + r4; let elapsed = start.elapsed(); diff --git a/examples/circuit_breaker.rs b/examples/circuit_breaker.rs index 8fb8938..24c7ba7 100644 --- a/examples/circuit_breaker.rs +++ b/examples/circuit_breaker.rs @@ -150,7 +150,7 @@ //! Result 7: Ok("Success") //! ``` -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; use std::sync::Arc; use std::time::Instant; use tokio::sync::Mutex; diff --git a/examples/complex_dag.rs b/examples/complex_dag.rs index 339a71a..952e127 100644 --- a/examples/complex_dag.rs +++ b/examples/complex_dag.rs @@ -66,7 +66,7 @@ //! Combined: 6 items, sum=57, avg=9.50 //! ``` -use dagx::{task, DagResult, DagRunner, Task}; +use dagx::{task, DagResult, DagRunner}; // Extract: Load data from different sources struct LoadData { diff --git a/examples/conditional_workflow.rs b/examples/conditional_workflow.rs index 9d1d1e3..51e86cb 100644 --- a/examples/conditional_workflow.rs +++ b/examples/conditional_workflow.rs @@ -55,7 +55,7 @@ //! Expected: Categories: [low, invalid, high] //! ``` -use dagx::{task, DagResult, DagRunner, Task}; +use dagx::{task, DagResult, DagRunner}; // Task that validates and processes input struct ValidateAndDouble { diff --git a/examples/data_pipeline.rs b/examples/data_pipeline.rs index 0c27fe8..2adec0c 100644 --- a/examples/data_pipeline.rs +++ b/examples/data_pipeline.rs @@ -75,7 +75,7 @@ //! ============================ //! ``` -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; // Data source tasks struct LoadValue1; diff --git a/examples/debug_tracing.rs b/examples/debug_tracing.rs index 119d8c9..5be3a08 100644 --- a/examples/debug_tracing.rs +++ b/examples/debug_tracing.rs @@ -114,7 +114,7 @@ //! FulfillOrder: 30ms //! ``` -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; use std::time::Instant; use tokio::time::{sleep, Duration}; diff --git a/examples/error_handling.rs b/examples/error_handling.rs index 158cbab..f6006ca 100644 --- a/examples/error_handling.rs +++ b/examples/error_handling.rs @@ -114,7 +114,7 @@ //! === Error Handling Example Complete === //! ``` -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; // Source task that provides a string input struct StringSource(&'static str); diff --git a/examples/large_dag.rs b/examples/large_dag.rs index b6ba2f7..815dc9f 100644 --- a/examples/large_dag.rs +++ b/examples/large_dag.rs @@ -109,7 +109,7 @@ //! - Memory overhead is minimal (~ 200 bytes per task) //! ``` -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; use std::time::Instant; // Simple computation task diff --git a/examples/parallelism_proof.rs b/examples/parallelism_proof.rs index 1be2ba8..894aa0b 100644 --- a/examples/parallelism_proof.rs +++ b/examples/parallelism_proof.rs @@ -47,7 +47,7 @@ //! This proves TRUE PARALLEL EXECUTION! //! ``` -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; use std::time::{Duration, Instant}; use tokio::time::sleep; diff --git a/examples/retry_strategies.rs b/examples/retry_strategies.rs index 278ad42..73c9dcf 100644 --- a/examples/retry_strategies.rs +++ b/examples/retry_strategies.rs @@ -152,7 +152,7 @@ //! Result: Err("Retry exhausted after 3 attempts: Connection timeout") //! ``` -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; use tokio::time::{sleep, Duration}; // Retry strategy configuration diff --git a/examples/timeout.rs b/examples/timeout.rs index 5f5c246..ccdca3b 100644 --- a/examples/timeout.rs +++ b/examples/timeout.rs @@ -98,7 +98,7 @@ //! === Timeout Example Complete === //! ``` -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; use tokio::time::{sleep, timeout, Duration}; // Fast task that completes quickly diff --git a/examples/tracing_example.rs b/examples/tracing_example.rs index 87924eb..85f6af0 100644 --- a/examples/tracing_example.rs +++ b/examples/tracing_example.rs @@ -33,7 +33,7 @@ //! - **ERROR**: Errors, panics, cycle detection //! -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; use tracing_subscriber::{fmt, EnvFilter}; // Simple source task diff --git a/src/builder.rs b/src/builder.rs index d5bba4e..b0ca918 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -4,8 +4,8 @@ //! The type parameter tracks whether dependencies have been specified yet. use crate::deps::DepsTuple; +use crate::node::TaskExecute; use crate::runner::DagRunner; -use crate::task::Task; use crate::types::{NodeId, TaskHandle}; #[cfg(feature = "tracing")] @@ -69,13 +69,13 @@ impl IsUnitType for () {} /// assert_eq!(dag.get(&b).unwrap(), 20); /// # }; /// ``` -pub struct TaskBuilder<'a, Tk: Task, Deps> { +pub struct TaskBuilder<'a, Tk: TaskExecute, Deps> { pub(crate) id: NodeId, pub(crate) dag: &'a DagRunner, pub(crate) _phantom: std::marker::PhantomData<(Tk, Deps)>, } -impl<'a, Tk: Task, Deps> TaskBuilder<'a, Tk, Deps> { +impl<'a, Tk: TaskExecute, Deps> TaskBuilder<'a, Tk, Deps> { /// Provide all dependencies exactly once as a tuple. /// /// The dependencies must match the task's `Input` type exactly: diff --git a/src/builder/tests.rs b/src/builder/tests.rs index b6f095e..5c4773c 100644 --- a/src/builder/tests.rs +++ b/src/builder/tests.rs @@ -1,8 +1,10 @@ //! Unit tests for builder module +// Allow tests to reference dagx:: paths (for proc macro generated code) +extern crate self as dagx; + use crate::builder::IsUnitType; use crate::runner::DagRunner; -use crate::task::Task; use crate::types::TaskHandle; // Initialize tracing subscriber for tests (idempotent) diff --git a/src/deps.rs b/src/deps.rs index d2497ce..9bbb951 100644 --- a/src/deps.rs +++ b/src/deps.rs @@ -8,7 +8,7 @@ //! use the provided implementations. use crate::builder::TaskBuilder; -use crate::task::Task; +use crate::node::TaskExecute; use crate::types::{NodeId, TaskHandle}; pub(crate) trait DepsTuple { @@ -48,13 +48,13 @@ impl DepsTuple for (TaskHandle,) { } // Support for &TaskBuilder -impl<'a, Tk: Task, Deps> DepsTuple for &TaskBuilder<'a, Tk, Deps> { +impl<'a, Tk: TaskExecute, Deps> DepsTuple for &TaskBuilder<'a, Tk, Deps> { fn to_node_ids(&self) -> Vec { vec![self.id] } } -impl<'a, Tk: Task, Deps> DepsTuple for (&TaskBuilder<'a, Tk, Deps>,) { +impl<'a, Tk: TaskExecute, Deps> DepsTuple for (&TaskBuilder<'a, Tk, Deps>,) { fn to_node_ids(&self) -> Vec { vec![self.0.id] } @@ -89,7 +89,7 @@ macro_rules! impl_deps_tuple { } // For &TaskBuilder tuples - impl<'a, $($Tk: Task, $D),+> DepsTuple<($($Tk::Output,)+)> for ($(&TaskBuilder<'a, $Tk, $D>,)+) { + impl<'a, $($Tk: TaskExecute, $D),+> DepsTuple<($($Tk::Output,)+)> for ($(&TaskBuilder<'a, $Tk, $D>,)+) { #[allow(non_snake_case)] fn to_node_ids(&self) -> Vec { let ($($T,)+) = self; diff --git a/src/deps/tests.rs b/src/deps/tests.rs index c3e40a2..cf06c19 100644 --- a/src/deps/tests.rs +++ b/src/deps/tests.rs @@ -1,8 +1,10 @@ //! Unit tests for deps module +// Allow tests to reference dagx:: paths (for proc macro generated code) +extern crate self as dagx; + use crate::deps::DepsTuple; use crate::runner::DagRunner; -use crate::task::Task; use crate::types::{NodeId, TaskHandle}; // Test task for dependency testing diff --git a/src/extract.rs b/src/extract.rs deleted file mode 100644 index 83fe7e9..0000000 --- a/src/extract.rs +++ /dev/null @@ -1,291 +0,0 @@ -//! Input extraction for task dependencies via oneshot channels. -//! -//! This module handles extracting typed inputs from task dependencies via oneshot channels. -//! -//! # How it works -//! -//! When a task executes, it needs to extract its inputs from type-erased oneshot receivers. -//! Outputs are wrapped in Arc for efficient fanout (cheap Arc clones instead of data clones). -//! This module provides the ExtractInput trait with implementations for: -//! - `()` for tasks with no dependencies -//! - Common types (`i32`, `String`, etc.) for tasks with a single dependency -//! - `(A, B, ...)` tuples for tasks with multiple dependencies (awaited concurrently) -//! -//! # Why macros? -//! -//! Rust lacks variadic generics, so we use macros to generate implementations for: -//! - Common primitive/standard types (one impl per type) -//! - Tuples of different sizes (one impl per size, up to 8 elements) -//! -//! This is standard practice in Rust for working with tuples of different arities. - -use std::any::Any; -use std::sync::Arc; - -use futures::channel::oneshot; - -/// Helper trait for async input extraction from oneshot channels. -/// -/// Type erasure occurs only at the ExecutableNode trait boundary - by the time -/// we're in ExtractInput, we know the concrete type and can safely downcast. -pub(crate) trait ExtractInput: Sized + Clone { - fn extract_from_channels( - receivers: Vec>, - ) -> impl std::future::Future> + Send; -} - -// Unit type - no dependencies -impl ExtractInput for () { - async fn extract_from_channels(_receivers: Vec>) -> Result { - Ok(()) - } -} - -// Implementations for common primitive and standard library types. -// These allow tasks to receive these types as single inputs without wrapping in tuples. -// Receives Arc from channel, clones the inner value for task consumption. -// For custom types, implement ExtractInput following the same pattern. -macro_rules! impl_extract_single { - ($($t:ty),+) => { - $( - impl ExtractInput for $t { - async fn extract_from_channels(mut receivers: Vec>) -> Result { - if receivers.len() != 1 { - return Err(format!("Expected 1 dependency, got {}", receivers.len())); - } - let rx = *receivers.pop() - .unwrap() - .downcast::>>() - .map_err(|_| format!("Type mismatch: expected Arc<{}>", std::any::type_name::<$t>()))?; - let arc_value = rx.await.map_err(|_| "Channel closed before receiving value".to_string())?; - Ok((*arc_value).clone()) - } - } - )+ - }; -} - -impl_extract_single!( - i8, i16, i32, i64, i128, isize, u8, u16, u32, u64, u128, usize, f32, f64, bool, char, String -); - -// Implement ExtractInput for Result types to support error handling patterns -impl ExtractInput for Result -where - T: 'static + Clone + Send + Sync, - E: 'static + Clone + Send + Sync, -{ - async fn extract_from_channels( - mut receivers: Vec>, - ) -> Result { - if receivers.len() != 1 { - return Err(format!("Expected 1 dependency, got {}", receivers.len())); - } - let rx = *receivers - .pop() - .unwrap() - .downcast::>>>() - .map_err(|_| { - format!( - "Type mismatch: expected Arc>", - std::any::type_name::(), - std::any::type_name::() - ) - })?; - let arc_value = rx - .await - .map_err(|_| "Channel closed before receiving value".to_string())?; - Ok((*arc_value).clone()) - } -} - -// Implement ExtractInput for Option types -impl ExtractInput for Option -where - T: 'static + Clone + Send + Sync, -{ - async fn extract_from_channels( - mut receivers: Vec>, - ) -> Result { - if receivers.len() != 1 { - return Err(format!("Expected 1 dependency, got {}", receivers.len())); - } - let rx = *receivers - .pop() - .unwrap() - .downcast::>>>() - .map_err(|_| { - format!( - "Type mismatch: expected Arc>", - std::any::type_name::() - ) - })?; - let arc_value = rx - .await - .map_err(|_| "Channel closed before receiving value".to_string())?; - Ok((*arc_value).clone()) - } -} - -// Implement ExtractInput for Vec types -impl ExtractInput for Vec -where - T: 'static + Clone + Send + Sync, -{ - async fn extract_from_channels( - mut receivers: Vec>, - ) -> Result { - if receivers.len() != 1 { - return Err(format!("Expected 1 dependency, got {}", receivers.len())); - } - let rx = *receivers - .pop() - .unwrap() - .downcast::>>>() - .map_err(|_| { - format!( - "Type mismatch: expected Arc>", - std::any::type_name::() - ) - })?; - let arc_value = rx - .await - .map_err(|_| "Channel closed before receiving value".to_string())?; - Ok((*arc_value).clone()) - } -} - -// Implement ExtractInput for HashMap types -impl ExtractInput for std::collections::HashMap -where - K: 'static + Clone + Send + Sync + Eq + std::hash::Hash, - V: 'static + Clone + Send + Sync, -{ - async fn extract_from_channels( - mut receivers: Vec>, - ) -> Result { - if receivers.len() != 1 { - return Err(format!("Expected 1 dependency, got {}", receivers.len())); - } - let rx = *receivers - .pop() - .unwrap() - .downcast::>>>() - .map_err(|_| { - format!( - "Type mismatch: expected Arc>", - std::any::type_name::(), - std::any::type_name::() - ) - })?; - let arc_value = rx - .await - .map_err(|_| "Channel closed before receiving value".to_string())?; - Ok((*arc_value).clone()) - } -} - -// Implement ExtractInput for Arc types - SPECIAL CASE: just clone the Arc, don't unwrap! -// This allows zero-copy data sharing for users who wrap their data in Arc. -impl ExtractInput for Arc -where - T: 'static + Clone + Send + Sync, -{ - async fn extract_from_channels( - mut receivers: Vec>, - ) -> Result { - if receivers.len() != 1 { - return Err(format!("Expected 1 dependency, got {}", receivers.len())); - } - let rx = *receivers - .pop() - .unwrap() - .downcast::>>>() - .map_err(|_| { - format!( - "Type mismatch: expected Arc>", - std::any::type_name::() - ) - })?; - let arc_arc_value = rx - .await - .map_err(|_| "Channel closed before receiving value".to_string())?; - // Return the inner Arc (clone is cheap - just refcount increment) - Ok(Arc::clone(&*arc_arc_value)) - } -} - -/// Macro to implement ExtractInput for tuple types. -/// -/// Generates an implementation that extracts values from multiple dependency channels -/// and combines them into a tuple. All channels are awaited concurrently using join_all. -/// -/// We need separate implementations for each tuple size (2-8) because Rust doesn't support -/// variadic generics. This is standard practice when working with tuples in Rust. -/// -/// # Compile-time Safety -/// -/// The number of dependencies is guaranteed to match the tuple size at compile time through: -/// 1. The `DepsTuple` trait bound on `depends_on()` ensures type/arity matching -/// 2. The `IsUnitType` constraint on `From` prevents using tasks without calling `depends_on()` -/// -/// Therefore, no runtime validation is needed - if it compiles, the dependency count is correct. -macro_rules! impl_extract_tuple { - ($($T:ident : $idx:tt),+) => { - impl<$($T: 'static + Clone + Send + Sync),+> ExtractInput for ($($T,)+) { - async fn extract_from_channels(receivers: Vec>) -> Result { - // Verify we have the correct number of receivers - let expected_count = [$($idx),+].len(); - if receivers.len() != expected_count { - return Err(format!("Expected {} dependencies, got {}", expected_count, receivers.len())); - } - - // Convert Vec to iterator so we can consume it - let mut iter = receivers.into_iter(); - - // Extract each receiver by consuming from the iterator - // Receivers now contain Arc-wrapped values for efficient fanout - // Dereference Box after downcasting to get the receiver itself - // Use type names as variable names to avoid numeric suffix issues - $( - #[allow(non_snake_case)] - let $T = *iter.next() - .ok_or_else(|| format!("Missing receiver at index {}", $idx))? - .downcast::>>() - .map_err(|_| format!("Type mismatch at index {}: expected Arc<{}>", - $idx, std::any::type_name::<$T>()))?; - )+ - - // Await all channels concurrently - this is a key optimization! - // We can start receiving as soon as ANY dependency completes, not waiting - // for them sequentially. - let arc_results = futures::join!( - $( - async move { - $T.await.map_err(|_| format!("Channel {} closed", $idx)) - } - ),+ - ); - - // Clone inner values from Arc (cheap for small types, necessary for ownership) - Ok(($( - (*arc_results.$idx?).clone() - ,)+)) - } - } - }; -} - -// Generate ExtractInput implementations for tuples of size 2-8. -// Each invocation creates an impl for a specific tuple size. -// Supporting up to 8 elements covers the vast majority of use cases. -impl_extract_tuple!(A:0, B:1); -impl_extract_tuple!(A:0, B:1, C:2); -impl_extract_tuple!(A:0, B:1, C:2, D:3); -impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4); -impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5); -impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6); -impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6, H:7); - -#[cfg(test)] -mod tests; diff --git a/src/extract/tests.rs b/src/extract/tests.rs deleted file mode 100644 index dacbbe9..0000000 --- a/src/extract/tests.rs +++ /dev/null @@ -1,410 +0,0 @@ -//! Unit tests for extract module - -#![allow(dead_code)] // Test helpers generate unused code via macro - -use crate::extract::ExtractInput; -use crate::task::Task; -use futures::channel::oneshot; -use std::any::Any; -use std::sync::Arc; - -// Test task for error scenarios -#[allow(dead_code)] -struct TestTask { - value: i32, -} - -#[crate::task] -#[allow(dead_code)] -impl TestTask { - #[cfg(not(tarpaulin_include))] - async fn run(&self) -> i32 { - self.value - } -} - -#[allow(dead_code)] -struct TestTaskWithInput; - -#[crate::task] -#[allow(dead_code)] -impl TestTaskWithInput { - #[cfg(not(tarpaulin_include))] - async fn run(input: &i32) -> i32 { - input * 2 - } -} - -#[tokio::test] -async fn test_extract_result_type_single_dependency_error() { - // Test Result with wrong dependency count - let deps = vec![ - Box::new(oneshot::channel::>().1) as Box, - Box::new(oneshot::channel::>().1) as Box, - ]; // Two dependencies when expecting 1 - - let result = Result::::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Expected 1 dependency, got 2")); -} - -#[tokio::test] -async fn test_extract_result_type_success() { - // Test Result successful extraction - let (tx, rx) = oneshot::channel::>>(); - - // Send value in background - tokio::spawn(async move { - let _ = tx.send(Arc::new(Ok(42))); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Result::::extract_from_channels(deps).await; - assert!(result.is_ok()); - assert_eq!(result.unwrap(), Ok(42)); -} - -#[tokio::test] -async fn test_extract_option_type_single_dependency_error() { - // Test Option with wrong dependency count - let deps = vec![ - Box::new(oneshot::channel::>().1) as Box, - Box::new(oneshot::channel::>().1) as Box, - Box::new(oneshot::channel::>().1) as Box, - ]; // Three dependencies when expecting 1 - - let result = Option::::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Expected 1 dependency, got 3")); -} - -#[tokio::test] -async fn test_extract_option_type_success() { - // Test Option successful extraction - let (tx, rx) = oneshot::channel::>>(); - - // Send value in background - tokio::spawn(async move { - let _ = tx.send(Arc::new(Some(99))); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Option::::extract_from_channels(deps).await; - assert!(result.is_ok()); - assert_eq!(result.unwrap(), Some(99)); -} - -#[tokio::test] -async fn test_extract_vec_type_single_dependency_error() { - // Test Vec with wrong dependency count - let deps = vec![]; // Zero dependencies when expecting 1 - - let result = Vec::::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Expected 1 dependency, got 0")); -} - -#[tokio::test] -async fn test_extract_vec_type_success() { - // Test Vec successful extraction - let (tx, rx) = oneshot::channel::>>(); - - // Send value in background - tokio::spawn(async move { - let _ = tx.send(Arc::new(vec![1, 2, 3])); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Vec::::extract_from_channels(deps).await; - assert!(result.is_ok()); - assert_eq!(result.unwrap(), vec![1, 2, 3]); -} - -#[tokio::test] -async fn test_extract_primitive_single_dependency_error() { - // Test primitive types with wrong dependency count - let deps = vec![ - Box::new(oneshot::channel::().1) as Box, - Box::new(oneshot::channel::().1) as Box, - ]; // Two dependencies when expecting 1 - - let result = i32::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Expected 1 dependency, got 2")); - - // Test with String - let deps = vec![ - Box::new(oneshot::channel::().1) as Box, - Box::new(oneshot::channel::().1) as Box, - ]; - let result = String::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Expected 1 dependency, got 2")); - - // Test with f64 - let deps = vec![ - Box::new(oneshot::channel::().1) as Box, - Box::new(oneshot::channel::().1) as Box, - ]; - let result = f64::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Expected 1 dependency, got 2")); -} - -#[tokio::test] -async fn test_extract_unit_type() { - // Test that unit type always succeeds regardless of dependencies - let deps = vec![]; - let result = <()>::extract_from_channels(deps).await; - assert!(result.is_ok()); - - // Even with dependencies (ignored), unit type succeeds - let deps = vec![ - Box::new(oneshot::channel::().1) as Box, - Box::new(oneshot::channel::().1) as Box, - ]; - let result = <()>::extract_from_channels(deps).await; - assert!(result.is_ok()); -} - -#[tokio::test] -async fn test_extract_tuple_basic() { - // Test tuple extraction with channels - let (tx1, rx1) = oneshot::channel::>(); - let (tx2, rx2) = oneshot::channel::>(); - - // Send values in background - tokio::spawn(async move { - let _ = tx1.send(Arc::new(42)); - let _ = tx2.send(Arc::new(100)); - }); - - let deps = vec![ - Box::new(rx1) as Box, - Box::new(rx2) as Box, - ]; - - // Extract a tuple (i32, i32) - let result = <(i32, i32)>::extract_from_channels(deps).await; - assert!(result.is_ok()); - assert_eq!(result.unwrap(), (42, 100)); -} - -#[tokio::test] -async fn test_extract_channel_closed() { - // Test error when channel is closed - let (tx, rx) = oneshot::channel::>(); - // Explicitly drop tx to close the channel - drop(tx); - - let deps = vec![Box::new(rx) as Box]; - let result = i32::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Channel closed")); -} - -#[tokio::test] -async fn test_extract_type_mismatch() { - // Test type mismatch during downcast - let (tx, rx) = oneshot::channel::(); - - // Send a String - tokio::spawn(async move { - let _ = tx.send("hello".to_string()); - }); - - // Try to extract as wrong type - put wrong receiver type in deps - let deps = vec![Box::new(rx) as Box]; - let result = i32::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Type mismatch")); -} - -#[tokio::test] -async fn test_extract_all_primitive_types() { - // Test that all primitive types in impl_extract_single! work correctly - macro_rules! test_type { - ($t:ty, $val:expr) => { - let (tx, rx) = oneshot::channel::>(); - - tokio::spawn(async move { - let _ = tx.send(Arc::new($val)); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = <$t>::extract_from_channels(deps).await; - assert!(result.is_ok()); - assert_eq!(result.unwrap(), $val); - }; - } - - test_type!(i8, 42i8); - test_type!(i16, 42i16); - test_type!(i32, 42i32); - test_type!(i64, 42i64); - test_type!(String, "test".to_string()); -} - -#[tokio::test] -async fn test_extract_larger_tuples() { - // Test extraction for larger tuple sizes - let (tx1, rx1) = oneshot::channel::>(); - let (tx2, rx2) = oneshot::channel::>(); - let (tx3, rx3) = oneshot::channel::>(); - let (tx4, rx4) = oneshot::channel::>(); - let (tx5, rx5) = oneshot::channel::>(); - - // Send values in background - tokio::spawn(async move { - let _ = tx1.send(Arc::new(0)); - let _ = tx2.send(Arc::new(1)); - let _ = tx3.send(Arc::new(2)); - let _ = tx4.send(Arc::new(3)); - let _ = tx5.send(Arc::new(4)); - }); - - let deps = vec![ - Box::new(rx1) as Box, - Box::new(rx2) as Box, - Box::new(rx3) as Box, - Box::new(rx4) as Box, - Box::new(rx5) as Box, - ]; - - // Extract a 5-element tuple - let result = <(i32, i32, i32, i32, i32)>::extract_from_channels(deps).await; - assert!(result.is_ok()); - assert_eq!(result.unwrap(), (0, 1, 2, 3, 4)); -} - -#[tokio::test] -async fn test_extract_arc_type() { - // Test Arc extraction - let (tx, rx) = oneshot::channel::>>(); - - let test_arc = Arc::new("test".to_string()); - let test_arc_clone = test_arc.clone(); - - tokio::spawn(async move { - let _ = tx.send(Arc::new(test_arc)); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Arc::::extract_from_channels(deps).await; - assert!(result.is_ok()); - assert_eq!(*result.unwrap(), *test_arc_clone); -} - -// Error path tests for ExtractInput implementations -use std::collections::HashMap; - -#[tokio::test] -async fn test_hashmap_extract_wrong_dependency_count() { - let deps = vec![ - Box::new(oneshot::channel::>>().1) as Box, - Box::new(oneshot::channel::>>().1) as Box, - ]; - - let result = HashMap::::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Expected 1 dependency, got 2")); -} - -#[tokio::test] -async fn test_hashmap_extract_type_mismatch() { - let (tx, rx) = oneshot::channel::>(); - tokio::spawn(async move { - let _ = tx.send(Arc::new("wrong type".to_string())); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = HashMap::::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Type mismatch")); -} - -#[tokio::test] -async fn test_hashmap_extract_channel_closed() { - let (tx, rx) = oneshot::channel::>>(); - drop(tx); - - let deps = vec![Box::new(rx) as Box]; - let result = HashMap::::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Channel closed")); -} - -#[tokio::test] -async fn test_result_extract_downcast_failure() { - let (tx, rx) = oneshot::channel::>(); - tokio::spawn(async move { - let _ = tx.send(Arc::new("wrong".to_string())); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Result::::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Type mismatch")); -} - -#[tokio::test] -async fn test_option_extract_downcast_failure() { - let (tx, rx) = oneshot::channel::>(); - tokio::spawn(async move { - let _ = tx.send(Arc::new("wrong".to_string())); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Option::::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Type mismatch")); -} - -#[tokio::test] -async fn test_vec_extract_downcast_failure() { - let (tx, rx) = oneshot::channel::>(); - tokio::spawn(async move { - let _ = tx.send(Arc::new("wrong".to_string())); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Vec::::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Type mismatch")); -} - -#[tokio::test] -async fn test_arc_extract_wrong_dependency_count() { - let deps = vec![]; - let result = Arc::::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Expected 1 dependency, got 0")); -} - -#[tokio::test] -async fn test_arc_extract_downcast_failure() { - let (tx, rx) = oneshot::channel::>(); - tokio::spawn(async move { - let _ = tx.send(Arc::new("wrong".to_string())); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Arc::::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Type mismatch")); -} - -#[tokio::test] -async fn test_tuple_wrong_dependency_count() { - let deps = vec![ - Box::new(oneshot::channel::>().1) as Box, - Box::new(oneshot::channel::>().1) as Box, - Box::new(oneshot::channel::>().1) as Box, - ]; - - let result = <(i32, i32)>::extract_from_channels(deps).await; - assert!(result.is_err()); - assert!(result - .unwrap_err() - .contains("Expected 2 dependencies, got 3")); -} diff --git a/src/lib.rs b/src/lib.rs index d3b4de1..0834fe2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -113,6 +113,7 @@ //! use dagx::{task, Task}; //! //! // Define a struct to hold your multiple values +//! #[derive(Clone)] //! struct UserData { //! name: String, //! age: i32, @@ -722,7 +723,6 @@ mod builder; pub mod cycle_prevention; mod deps; mod error; -mod extract; mod node; mod runner; mod task; @@ -731,6 +731,7 @@ mod types; // Public re-exports pub use builder::TaskBuilder; pub use error::{DagError, DagResult}; +pub use node::TaskExecute; // For macro-generated trait implementations pub use runner::DagRunner; pub use task::Task; pub use types::{Pending, TaskHandle}; diff --git a/src/node.rs b/src/node.rs index 0d2c44e..69178a4 100644 --- a/src/node.rs +++ b/src/node.rs @@ -11,138 +11,146 @@ use std::future::Future; use std::pin::Pin; use std::sync::Arc; -use futures::channel::oneshot; - use crate::error::DagError; -use crate::extract::ExtractInput; -use crate::task::Task; use crate::types::NodeId; -/// Type-erased channel endpoints (senders or receivers) -type ErasedChannels = Vec>; - -/// Type-erased channel pair (senders, receivers) -type ChannelPair = (ErasedChannels, ErasedChannels); +/// Type-erased listeners (for receiving inputs from dependencies) +type ErasedListeners = Vec>; /// Type-erased async execution result (Arc-wrapped for efficient fanout) type ExecuteFuture = Pin, DagError>> + Send>>; +/// Type-erased async execution result for TaskExecute trait (returns String errors) +type TaskExecuteFuture = + Pin, String>> + Send>>; + /// Internal trait for executing heterogeneous tasks. /// /// This trait provides type erasure necessary for storing different task types in a single /// collection. While the public API (Task, Handle, Node) is fully type-safe, internally we /// need dynamic dispatch to execute tasks with different input/output types. /// -/// This is the ONLY place where type erasure occurs - channels are created with full type +/// This is the ONLY place where type erasure occurs - announcements are created with full type /// information and only erased at this trait boundary. /// /// This is an implementation detail and not part of the public API. pub(crate) trait ExecutableNode: Send { - /// Create type-erased output channels for this node's dependents. + /// Create a type-erased announcement for this node's output. /// - /// Returns (senders, receivers) where: - /// - senders: Type-erased senders that this node will use to send outputs - /// - receivers: Type-erased receivers that dependents will use to receive inputs - fn create_output_channels(&self, num_dependents: usize) -> ChannelPair; + /// Returns a tuple of (Announcer, Announcement) both type-erased. + /// The Announcer will be used by this node to broadcast its output. + /// The Announcement can be used to create listeners for dependents. + fn create_announcement(&self) -> (Box, Box); - /// Execute the task with type-erased input receivers and output senders. + /// Create a type-erased listener from an announcement. + /// + /// This allows dependents to listen to this node's output. + fn create_listener(&self, announcement: &dyn Any) -> Box; + + /// Execute the task with type-erased input listeners and an output announcer. /// /// The TypedNode implementation knows the concrete types and can safely downcast. /// Consumes the node and returns the output value. - fn execute_with_channels( + fn execute_with_listeners( self: Box, - receivers: ErasedChannels, - senders: ErasedChannels, + listeners: ErasedListeners, + announcer: Box, ) -> ExecuteFuture; } /// Fully-typed node storage for a single task. /// -/// Stores a task of type T. Implements ExecutableNode for type erasure, -/// allowing heterogeneous tasks to be stored in a single Vec. +/// Stores a task that has an `execute()` method (generated by the #[task] macro). +/// Implements ExecutableNode for type erasure, allowing heterogeneous tasks to be +/// stored in a single Vec. /// /// # Ownership Model /// /// The task is owned directly (no Mutex needed) and consumed during execution. /// Each task executes exactly once, taking ownership and producing an output. /// -/// # Oneshot Channels +/// # Announcement Broadcasting /// -/// Outputs are communicated via oneshot channels for inter-task communication. -/// The output is also returned from execute_with_channels for central storage. -pub(crate) struct TypedNode { +/// Outputs are communicated via announcement channels for inter-task communication. +/// Each task has ONE announcer that broadcasts to multiple listeners. +/// The output is also returned from execute_with_listeners for central storage. +pub(crate) struct TypedNode { pub(crate) id: NodeId, pub(crate) task: T, + pub(crate) _output_type: std::marker::PhantomData ()>, } -impl TypedNode { +impl TypedNode { pub(crate) fn new(id: NodeId, task: T) -> Self { - Self { id, task } + Self { + id, + task, + _output_type: std::marker::PhantomData, + } } } -impl ExecutableNode for TypedNode +// Implementation for tasks created with the #[task] macro +// These tasks have an execute() method generated by the macro +impl ExecutableNode for TypedNode where - T::Input: 'static + Clone + ExtractInput, - T::Output: 'static + Clone, + T: TaskExecute, { - fn create_output_channels(&self, num_dependents: usize) -> ChannelPair { - let mut senders = Vec::new(); - let mut receivers = Vec::new(); - - for _ in 0..num_dependents { - // Create channels that pass Arc-wrapped outputs (efficient for fanout) - let (tx, rx) = oneshot::channel::>(); - senders.push(Box::new(tx) as Box); - receivers.push(Box::new(rx) as Box); - } - - (senders, receivers) + fn create_announcement(&self) -> (Box, Box) { + T::create_announcement_erased() } - fn execute_with_channels( + fn create_listener(&self, announcement: &dyn Any) -> Box { + T::create_listener_erased(announcement) + } + + fn execute_with_listeners( self: Box, - receivers: ErasedChannels, - senders: ErasedChannels, + listeners: ErasedListeners, + announcer: Box, ) -> ExecuteFuture { Box::pin(async move { - // Extract typed input from type-erased receivers - // TypedNode knows T::Input, so it can safely downcast - let input = match T::Input::extract_from_channels(receivers).await { - Ok(input) => input, - Err(_msg) => { - // Failed to extract input - likely channel closed or type mismatch - return Err(DagError::InvalidDependency { task_id: self.id.0 }); - } - }; + // Call the task's execute() method (generated by the #[task] macro) + // This handles listener extraction, calling run(), and announcing + let result = self.task.execute(listeners, announcer).await; - // Execute task (consumes self.task) - let output = self.task.run(input).await; - - // Arc Wrapping Strategy: - // Wrap output in Arc ONCE for efficient fan-out sharing. - // For 1:N dependencies, this enables O(1) sharing (Arc clone = atomic increment) - // instead of O(N) cloning (N deep copies of output data). - // Downstream tasks receive &T after ExtractInput extracts from Arc. - let arc_output = Arc::new(output); - - // Send Arc to all dependents via their channels - // TypedNode knows T::Output, so downcast is safe - for sender_any in senders { - if let Ok(sender) = sender_any.downcast::>>() { - // Clone Arc (just atomic refcount increment, NOT data clone) - // This is the key performance optimization for fan-out patterns - // Ignore send errors - dependent might have been cancelled - let _ = sender.send(Arc::clone(&arc_output)); + match result { + Ok(output_arc) => Ok(output_arc), + Err(err_msg) => { + // Execution failed - convert error message to DagError + eprintln!("Task execution failed: {}", err_msg); + Err(DagError::InvalidDependency { task_id: self.id.0 }) } } - - // Return Arc-wrapped output (for storage by runner) - Ok(arc_output as Arc) }) } } +/// Trait for tasks that have an execute() method (generated by #[task] macro). +/// +/// This trait is automatically implemented by the macro and should not be +/// implemented manually. +pub trait TaskExecute: Send { + /// Input type as a tuple of parameter types (for dependency checking) + /// - `()` for no parameters + /// - `T` for single parameter + /// - `(A, B, C, ...)` for multiple parameters + type Input: Send; + + /// Output type of the task + type Output: Send + Sync + Clone; + + fn execute( + self, + listeners: Vec>, + announcer: Box, + ) -> TaskExecuteFuture; + + fn create_announcement_erased() -> (Box, Box); + + fn create_listener_erased(announcement: &dyn Any) -> Box; +} + #[cfg(test)] mod tests; diff --git a/src/node/tests.rs b/src/node/tests.rs index c990071..e296f4b 100644 --- a/src/node/tests.rs +++ b/src/node/tests.rs @@ -1,7 +1,9 @@ //! Unit tests for node module +// Allow tests to reference dagx:: paths (for proc macro generated code) +extern crate self as dagx; + use crate::node::{ExecutableNode, TypedNode}; -use crate::task::Task; use crate::types::NodeId; // Simple test task @@ -17,25 +19,33 @@ impl TestTask { } #[test] -fn test_create_output_channels() { +fn test_create_announcement() { let node = TypedNode::new(NodeId(0), TestTask { value: 42 }); - // Create channels for 3 dependents - let (senders, receivers) = node.create_output_channels(3); + // Create ONE announcement that can broadcast to multiple listeners + let (announcer, announcement) = node.create_announcement(); - assert_eq!(senders.len(), 3); - assert_eq!(receivers.len(), 3); + // Verify both are created (type-erased) + assert!(std::any::Any::type_id(&*announcer) != std::any::TypeId::of::<()>()); + assert!(std::any::Any::type_id(&*announcement) != std::any::TypeId::of::<()>()); } #[test] -fn test_create_output_channels_zero_dependents() { +fn test_create_multiple_listeners() { let node = TypedNode::new(NodeId(0), TestTask { value: 42 }); - // Create channels for 0 dependents (sink node) - let (senders, receivers) = node.create_output_channels(0); + // Create announcement + let (_announcer, announcement) = node.create_announcement(); - assert_eq!(senders.len(), 0); - assert_eq!(receivers.len(), 0); + // Create multiple listeners from the same announcement (broadcast pattern) + let listener1 = node.create_listener(announcement.as_ref()); + let listener2 = node.create_listener(announcement.as_ref()); + let listener3 = node.create_listener(announcement.as_ref()); + + // All listeners are created from the same announcement + assert!(std::any::Any::type_id(&*listener1) != std::any::TypeId::of::<()>()); + assert!(std::any::Any::type_id(&*listener2) != std::any::TypeId::of::<()>()); + assert!(std::any::Any::type_id(&*listener3) != std::any::TypeId::of::<()>()); } struct TaskWithDependency; @@ -48,15 +58,19 @@ impl TaskWithDependency { } #[tokio::test] -async fn test_execute_with_channels_missing_input() { +async fn test_execute_with_listeners_missing_input() { // Test when task expects input but doesn't receive it - let node = Box::new(TypedNode::new(NodeId(0), TaskWithDependency)); + let node = TypedNode::new(NodeId(0), TaskWithDependency); - // TaskWithDependency expects an i32 input, but we provide no receivers - let receivers = vec![]; - let senders = vec![]; + // Create a real announcer even though we won't use it + let (announcer, _announcement) = node.create_announcement(); - let result = node.execute_with_channels(receivers, senders).await; + let node = Box::new(node); + + // TaskWithDependency expects an i32 input, but we provide no listeners + let listeners = vec![]; + + let result = node.execute_with_listeners(listeners, announcer).await; assert!(result.is_err()); if let Err(crate::error::DagError::InvalidDependency { task_id }) = result { @@ -67,49 +81,62 @@ async fn test_execute_with_channels_missing_input() { } #[tokio::test] -async fn test_execute_with_channels_success() { - // Test successful execution with channels - let dependency_node = Box::new(TypedNode::new(NodeId(0), TestTask { value: 21 })); +async fn test_execute_with_listeners_success() { + // Test successful execution with announcements and listeners + let dependency_node = TypedNode::new(NodeId(0), TestTask { value: 21 }); let dependent_node = Box::new(TypedNode::new(NodeId(1), TaskWithDependency)); - // Create a channel from dependency to dependent (Arc-wrapped) - let (tx, rx) = futures::channel::oneshot::channel::>(); + // Create announcement from dependency + let (dep_announcer, dep_announcement) = dependency_node.create_announcement(); - // Execute dependency node with sender - let dep_senders = vec![Box::new(tx) as Box]; - let dep_receivers = vec![]; + // Create listener for dependent from dependency's announcement + let listener = dependency_node.create_listener(dep_announcement.as_ref()); + + // Execute dependency node with announcer + let dependency_node = Box::new(dependency_node); + let dep_listeners = vec![]; // Execute in background let dep_handle = tokio::spawn(async move { dependency_node - .execute_with_channels(dep_receivers, dep_senders) + .execute_with_listeners(dep_listeners, dep_announcer) .await }); - // Execute dependent node with receiver - let receivers = vec![Box::new(rx) as Box]; - let senders = vec![]; + // Execute dependent node with listener + // Create a real announcer for the dependent (even though no one listens to it) + let dep_node_for_announcer = TypedNode::new(NodeId(1), TaskWithDependency); + let (dependent_announcer, _dep_announcement) = dep_node_for_announcer.create_announcement(); + + let listeners = vec![listener]; let result = dependent_node - .execute_with_channels(receivers, senders) + .execute_with_listeners(listeners, dependent_announcer) .await; // Wait for dependency to complete let _ = dep_handle.await; assert!(result.is_ok()); + let output = result.unwrap(); + let arc_value = output.downcast::().unwrap(); + assert_eq!(*arc_value, 42); // 21 * 2 = 42 } #[tokio::test] -async fn test_execute_with_channels_sink_node() { +async fn test_execute_with_listeners_sink_node() { // Test that sink nodes (no dependents) return their output - let node = Box::new(TypedNode::new(NodeId(0), TestTask { value: 42 })); + let node = TypedNode::new(NodeId(0), TestTask { value: 42 }); - // No senders means this is a sink node - let receivers = vec![]; - let senders = vec![]; + // Create a real announcer (even though no one will listen) + let (announcer, _announcement) = node.create_announcement(); - let result = node.execute_with_channels(receivers, senders).await; + let node = Box::new(node); + + // Sink node has no listeners + let listeners = vec![]; + + let result = node.execute_with_listeners(listeners, announcer).await; assert!(result.is_ok()); // Check output was returned (Arc-wrapped) @@ -119,29 +146,28 @@ async fn test_execute_with_channels_sink_node() { } #[tokio::test] -async fn test_execute_with_channels_non_sink_node() { - // Test that non-sink nodes send output via channels +async fn test_execute_with_listeners_non_sink_node() { + // Test that non-sink nodes broadcast output via announcement let node = TypedNode::new(NodeId(0), TestTask { value: 42 }); - // Create output channels for one dependent - let (senders, mut receivers) = node.create_output_channels(1); + // Create announcement and listener + let (announcer, announcement) = node.create_announcement(); + let listener = node.create_listener(announcement.as_ref()); - // Execute with senders (non-sink node) + // Execute with announcer (non-sink node) let node = Box::new(node); - let exec_receivers = vec![]; - let result = node.execute_with_channels(exec_receivers, senders).await; + let listeners = vec![]; + let result = node.execute_with_listeners(listeners, announcer).await; assert!(result.is_ok()); - // Verify output was sent through channel (Arc-wrapped) - let rx = receivers - .pop() - .unwrap() - .downcast::>>() + // Verify output was broadcast through announcement + let listener = *listener + .downcast::>() .unwrap(); - let received_arc = rx.await; - assert!(received_arc.is_ok()); - assert_eq!(*received_arc.unwrap(), 42); + let received = listener.listen().await; + assert!(received.is_ok()); + assert_eq!(received.unwrap(), 42); // Also verify output was returned (Arc-wrapped) let output = result.unwrap(); @@ -150,27 +176,30 @@ async fn test_execute_with_channels_non_sink_node() { } #[tokio::test] -async fn test_execute_with_multiple_dependents() { - // Test that output is sent to multiple dependents +async fn test_execute_with_multiple_listeners() { + // Test that output is broadcast to multiple listeners let node = TypedNode::new(NodeId(0), TestTask { value: 100 }); - // Create output channels for three dependents - let (senders, receivers) = node.create_output_channels(3); + // Create announcement and multiple listeners + let (announcer, announcement) = node.create_announcement(); + let listener1 = node.create_listener(announcement.as_ref()); + let listener2 = node.create_listener(announcement.as_ref()); + let listener3 = node.create_listener(announcement.as_ref()); // Execute node let node = Box::new(node); - let exec_receivers = vec![]; - let result = node.execute_with_channels(exec_receivers, senders).await; + let listeners = vec![]; + let result = node.execute_with_listeners(listeners, announcer).await; assert!(result.is_ok()); - // Verify all three dependents receive the output (Arc-wrapped) - for receiver in receivers { - let rx = receiver - .downcast::>>() + // Verify all three listeners receive the output + for listener in [listener1, listener2, listener3] { + let listener = *listener + .downcast::>() .unwrap(); - let received_arc = rx.await; - assert!(received_arc.is_ok()); - assert_eq!(*received_arc.unwrap(), 100); + let received = listener.listen().await; + assert!(received.is_ok()); + assert_eq!(received.unwrap(), 100); } } diff --git a/src/runner.rs b/src/runner.rs index ec4120d..282206c 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -22,9 +22,7 @@ use tracing::{debug, error, info, trace}; use crate::builder::TaskBuilder; use crate::error::{DagError, DagResult}; -use crate::extract::ExtractInput; use crate::node::{ExecutableNode, TypedNode}; -use crate::task::Task; use crate::types::{NodeId, Pending, TaskHandle}; // Guard to ensure run_lock is released even on early return or panic @@ -195,9 +193,7 @@ impl DagRunner { /// ``` pub fn add_task(&self, task: Tk) -> TaskBuilder<'_, Tk, Pending> where - Tk: Task + 'static, - Tk::Input: 'static + Clone + ExtractInput, - Tk::Output: 'static + Clone, + Tk: crate::node::TaskExecute + 'static, { let id = self.alloc_id(); @@ -321,39 +317,41 @@ impl DagRunner { let edges = self.edges.lock().clone(); - // Create all channels upfront - // Map: (producer_id, consumer_id) -> receiver index - // We store receivers in a Vec per consumer to maintain order - let mut consumer_receivers: HashMap>> = - HashMap::new(); - let mut producer_senders: HashMap>> = - HashMap::new(); + // Create ONE announcement per task + // Each task broadcasts its output to all dependents via announcement + let mut announcements: HashMap> = HashMap::new(); + let mut announcers: HashMap> = HashMap::new(); - // Create channels for each producer-consumer relationship - // We need to lock nodes temporarily to call create_output_channels + // Create announcements for all tasks { let nodes_lock = self.nodes.lock(); + for (node_id, node_opt) in nodes_lock.iter().enumerate() { + if let Some(node) = node_opt { + let (announcer, announcement) = node.create_announcement(); + announcers.insert(NodeId(node_id), announcer); + announcements.insert(NodeId(node_id), announcement); + } + } + } + // Create listeners for each consumer from their dependencies + let mut consumer_listeners: HashMap>> = + HashMap::new(); + + { + let nodes_lock = self.nodes.lock(); for (consumer_id, producer_ids) in &edges { + let mut listeners = Vec::new(); for &producer_id in producer_ids { - // Get the producer node (it's still in the vector at this point) - if let Some(Some(producer_node)) = nodes_lock.get(producer_id.0) { - // Ask the producer to create ONE channel for this consumer - let (mut senders, mut receivers) = producer_node.create_output_channels(1); - - // Store the sender for the producer - producer_senders - .entry(producer_id) - .or_default() - .push(senders.pop().unwrap()); - - // Store the receiver for the consumer (order matters!) - consumer_receivers - .entry(*consumer_id) - .or_default() - .push(receivers.pop().unwrap()); + if let Some(announcement) = announcements.get(&producer_id) { + if let Some(Some(producer_node)) = nodes_lock.get(producer_id.0) { + // Producer node creates listener from its own announcement + let listener = producer_node.create_listener(announcement.as_ref()); + listeners.push(listener); + } } } + consumer_listeners.insert(*consumer_id, listeners); } } @@ -412,11 +410,14 @@ impl DagRunner { }; if let Some(node) = node { - // Take the receivers for this consumer - let receivers = consumer_receivers.remove(&node_id).unwrap_or_default(); + // Take the listeners for this consumer + let listeners = consumer_listeners.remove(&node_id).unwrap_or_default(); - // Take the senders for this producer - let senders = producer_senders.remove(&node_id).unwrap_or_default(); + // Take the announcer for this producer + let announcer = announcers.remove(&node_id).unwrap_or_else(|| { + // Create dummy announcer for tasks with no dependents + Box::new(()) as Box + }); // Execute inline with panic handling. // @@ -424,7 +425,7 @@ impl DagRunner { // DagError::TaskPanicked, matching the behavior of async runtimes when // they spawn tasks. This guarantees consistent error handling whether // a task executes inline (single-task layer) or spawned (multi-task layer). - let result = AssertUnwindSafe(node.execute_with_channels(receivers, senders)) + let result = AssertUnwindSafe(node.execute_with_listeners(listeners, announcer)) .catch_unwind() .await .unwrap_or_else(|panic_payload| { @@ -473,15 +474,18 @@ impl DagRunner { }; if let Some(node) = node { - // Take the receivers for this consumer - let receivers = consumer_receivers.remove(&node_id).unwrap_or_default(); + // Take the listeners for this consumer + let listeners = consumer_listeners.remove(&node_id).unwrap_or_default(); - // Take the senders for this producer - let senders = producer_senders.remove(&node_id).unwrap_or_default(); + // Take the announcer for this producer + let announcer = announcers.remove(&node_id).unwrap_or_else(|| { + // Create dummy announcer for tasks with no dependents + Box::new(()) as Box + }); // Create a 'static future by taking ownership let inner_future = async move { - let task_future = node.execute_with_channels(receivers, senders); + let task_future = node.execute_with_listeners(listeners, announcer); let result = task_future.await; diff --git a/src/runner/tests.rs b/src/runner/tests.rs index c4d0671..287a9f0 100644 --- a/src/runner/tests.rs +++ b/src/runner/tests.rs @@ -1,8 +1,10 @@ //! Unit tests for runner module +// Allow tests to reference dagx:: paths (for proc macro generated code) +extern crate self as dagx; + use crate::error::DagError; use crate::runner::DagRunner; -use crate::task::Task; use crate::types::TaskHandle; // Initialize tracing subscriber for tests (idempotent) diff --git a/src/task.rs b/src/task.rs index a25f8cd..48fe859 100644 --- a/src/task.rs +++ b/src/task.rs @@ -3,94 +3,197 @@ //! This module defines the core Task trait that all DAG nodes must implement, //! along with convenience functions for creating tasks from closures. +use std::any::Any; use std::future::Future; +/// Helper trait for extracting inputs from listeners (internal use only) +trait ExtractFromListeners: Sized + Send + Sync + Clone + 'static { + fn extract( + listeners: Vec>, + ) -> impl Future> + Send; +} + +// Implementation for unit type (no dependencies) +impl ExtractFromListeners for () { + async fn extract(_listeners: Vec>) -> Result { + Ok(()) + } +} + +// Implementation for single input types (not tuples) +macro_rules! impl_extract_single { + ($($t:ty),+) => { + $( + impl ExtractFromListeners for $t { + async fn extract(mut listeners: Vec>) -> Result { + if listeners.len() != 1 { + return Err(format!("Expected 1 dependency, got {}", listeners.len())); + } + let listener = *listeners.pop() + .unwrap() + .downcast::>() + .map_err(|_| format!("Type mismatch: expected Listener<{}>", std::any::type_name::<$t>()))?; + listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string()) + } + } + )+ + }; +} + +impl_extract_single!(i8, i16, i32, i64, i128, isize, u8, u16, u32, u64, u128, usize, f32, f64, bool, char, String); + +// Implementation for Result types +impl ExtractFromListeners for Result +where + T: 'static + Clone + Send + Sync, + E: 'static + Clone + Send + Sync, +{ + async fn extract(mut listeners: Vec>) -> Result { + if listeners.len() != 1 { + return Err(format!("Expected 1 dependency, got {}", listeners.len())); + } + let listener = *listeners.pop() + .unwrap() + .downcast::>>() + .map_err(|_| format!("Type mismatch: expected Listener>", std::any::type_name::(), std::any::type_name::()))?; + listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string()) + } +} + +// Implementation for Option types +impl ExtractFromListeners for Option +where + T: 'static + Clone + Send + Sync, +{ + async fn extract(mut listeners: Vec>) -> Result { + if listeners.len() != 1 { + return Err(format!("Expected 1 dependency, got {}", listeners.len())); + } + let listener = *listeners.pop() + .unwrap() + .downcast::>>() + .map_err(|_| format!("Type mismatch: expected Listener>", std::any::type_name::()))?; + listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string()) + } +} + +// Implementation for Vec types +impl ExtractFromListeners for Vec +where + T: 'static + Clone + Send + Sync, +{ + async fn extract(mut listeners: Vec>) -> Result { + if listeners.len() != 1 { + return Err(format!("Expected 1 dependency, got {}", listeners.len())); + } + let listener = *listeners.pop() + .unwrap() + .downcast::>>() + .map_err(|_| format!("Type mismatch: expected Listener>", std::any::type_name::()))?; + listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string()) + } +} + +// Implementation for Arc types +impl ExtractFromListeners for std::sync::Arc +where + T: 'static + Clone + Send + Sync, +{ + async fn extract(mut listeners: Vec>) -> Result { + if listeners.len() != 1 { + return Err(format!("Expected 1 dependency, got {}", listeners.len())); + } + let listener = *listeners.pop() + .unwrap() + .downcast::>>() + .map_err(|_| format!("Type mismatch: expected Listener>", std::any::type_name::()))?; + listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string()) + } +} + +// Implementation for HashMap types +impl ExtractFromListeners for std::collections::HashMap +where + K: 'static + Clone + Send + Sync + Eq + std::hash::Hash, + V: 'static + Clone + Send + Sync, +{ + async fn extract(mut listeners: Vec>) -> Result { + if listeners.len() != 1 { + return Err(format!("Expected 1 dependency, got {}", listeners.len())); + } + let listener = *listeners.pop() + .unwrap() + .downcast::>>() + .map_err(|_| format!("Type mismatch: expected Listener>", std::any::type_name::(), std::any::type_name::()))?; + listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string()) + } +} + +// Macro to implement ExtractFromListeners for tuple types +macro_rules! impl_extract_tuple { + ($($T:ident : $idx:tt),+) => { + #[allow(non_snake_case)] + impl<$($T),+> ExtractFromListeners for ($($T,)+) + where + $($T: 'static + Clone + Send + Sync),+ + { + async fn extract(listeners: Vec>) -> Result { + let expected_count = [$($idx),+].len(); + if listeners.len() != expected_count { + return Err(format!("Expected {} dependencies, got {}", expected_count, listeners.len())); + } + + let mut iter = listeners.into_iter(); + $( + let $T = *iter.next().unwrap() + .downcast::>() + .map_err(|_| format!("Type mismatch at index {}: expected Listener<{}>", $idx, std::any::type_name::<$T>()))?; + )+ + + let results = futures::join!( + $( + async move { + $T.listen().await.map_err(|_| format!("Listener {} closed", $idx)) + } + ),+ + ); + + Ok(($( + results.$idx? + ,)+)) + } + } + }; +} + +impl_extract_tuple!(A:0, B:1); +impl_extract_tuple!(A:0, B:1, C:2); +impl_extract_tuple!(A:0, B:1, C:2, D:3); +impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4); +impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5); +impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6); +impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6, H:7); + /// A unit of async work with typed inputs and outputs. /// -/// Use the [`task`](crate::task) macro to implement this trait. The macro automatically -/// derives `Input` and `Output` types from your `run()` method signature. +/// **NOTE**: For new code, use the [`task`](crate::task) macro which automatically generates +/// an `execute()` method and handles all the wiring. This trait is primarily for internal use +/// and backward compatibility with `task_fn`. /// -/// # Task Patterns -/// -/// dagx supports three main task patterns: -/// -/// ## 1. Stateless Tasks (Pure Functions) -/// -/// Tasks with no internal state - just transform inputs to outputs: -/// -/// ``` -/// use dagx::{task, Task}; -/// -/// struct Add; -/// -/// #[task] -/// impl Add { -/// async fn run(a: &i32, b: &i32) -> i32 { -/// a + b -/// } -/// } -/// ``` -/// -/// ## 2. Stateful Tasks with Shared State (`&self`) -/// -/// Tasks that read from internal state but don't modify it: -/// -/// ``` -/// use dagx::{task, Task}; -/// -/// struct Multiplier { -/// factor: i32, -/// } -/// -/// #[task] -/// impl Multiplier { -/// async fn run(&self, input: &i32) -> i32 { -/// input * self.factor -/// } -/// } -/// ``` -/// -/// ## 3. Stateful Tasks with Mutable State (`&mut self`) -/// -/// Tasks that modify internal state during execution: -/// -/// ``` -/// use dagx::{task, Task}; -/// -/// struct Counter { -/// count: i32, -/// } -/// -/// #[task] -/// impl Counter { -/// async fn run(&mut self, increment: &i32) -> i32 { -/// self.count += increment; -/// self.count -/// } -/// } -/// ``` -/// -/// # Input Patterns -/// -/// - `()`: No dependencies (source task) -/// - `&T`: Single dependency -/// - `&A, &B, &C, ...`: Multiple dependencies (up to 8, order matters) -/// -/// # The Underlying Trait -/// -/// The `#[task]` macro generates an implementation of this trait: +/// The `#[task]` macro generates an `execute()` method on your struct that: +/// - Extracts inputs from announcement listeners +/// - Calls your `run()` method +/// - Announces the result to dependents +/// - Returns Arc-wrapped output /// /// # Output Wrapping /// /// **Task outputs are automatically wrapped in `Arc` internally** for efficient fan-out. -/// You define `Output = T`, but the framework wraps it in `Arc` before sending to channels. +/// You define output as `T`, but the framework wraps it in `Arc` when announcing. /// This enables O(1) sharing for 1:N dependencies - Arc is cloned (cheap), not your data. -/// -/// Downstream tasks receive `&T` after ExtractInput extracts from the Arc. This is transparent -/// to your task code - you work with `T`, the framework handles Arc wrapping/unwrapping. pub trait Task: Send { type Input: Send; - type Output: Send + Sync; // Sync required for Arc-wrapping and cross-thread sharing + type Output: Send + Sync + Clone; // Sync and Clone required for Arc-wrapping and announcement fn run(self, input: Self::Input) -> impl Future + Send; } @@ -100,12 +203,12 @@ pub trait Task: Send { /// **INTERNAL USE ONLY**: This function is for testing purposes only and should /// not be used in production code. Use the `#[task]` macro instead. #[doc(hidden)] -pub fn task_fn(f: F) -> impl Task +pub fn task_fn(f: F) -> impl Task + crate::node::TaskExecute where F: FnMut(I) -> Fut + Send + 'static, Fut: Future + Send + 'static, - I: Send + 'static, - O: Send + Sync + 'static, + I: ExtractFromListeners, + O: Send + Sync + Clone + 'static, { struct TaskFn where @@ -123,7 +226,7 @@ where F: FnMut(I) -> Fut + Send, Fut: Future + Send, I: Send, - O: Send + Sync, + O: Send + Sync + Clone, { type Input = I; type Output = O; @@ -133,6 +236,67 @@ where } } + impl crate::node::TaskExecute for TaskFn + where + F: FnMut(I) -> Fut + Send + 'static, + Fut: Future + Send + 'static, + I: ExtractFromListeners, + O: Send + Sync + Clone + 'static, + { + type Input = I; + type Output = O; + + fn execute( + mut self, + listeners: Vec>, + announcer: Box, + ) -> std::pin::Pin< + Box< + dyn Future, String>> + + Send, + >, + > { + Box::pin(async move { + use std::sync::Arc; + + // Extract the input using the ExtractFromListeners trait + let input = I::extract(listeners).await?; + + // Call the closure + let output: O = (self.f)(input).await; + + // Wrap in Arc + let output_arc = Arc::new(output); + + // Announce the result + let announcer_typed = announcer + .downcast::>() + .map_err(|_| { + "Type mismatch: announcer type does not match task output type".to_string() + })?; + let _ = announcer_typed.announce((*output_arc).clone()); + + // Return Arc-wrapped output + Ok(output_arc as Arc) + }) + } + + fn create_announcement_erased() -> (Box, Box) { + let (announcer, announcement) = announcement::Announcement::::new(); + ( + Box::new(announcer) as Box, + Box::new(announcement) as Box, + ) + } + + fn create_listener_erased(announcement: &dyn std::any::Any) -> Box { + let announcement = announcement + .downcast_ref::>() + .expect("Type mismatch: announcement type does not match task output type"); + Box::new(announcement.listener()) as Box + } + } + TaskFn { f, _phantom: std::marker::PhantomData, diff --git a/src/types.rs b/src/types.rs index e0f4e81..f1981e6 100644 --- a/src/types.rs +++ b/src/types.rs @@ -4,7 +4,7 @@ //! including node identifiers, task handles, and type-state markers. use crate::builder::{IsUnitType, TaskBuilder}; -use crate::task::Task; +use crate::node::TaskExecute; /// Opaque node identifier (internal use) #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] @@ -62,7 +62,7 @@ pub struct Pending; // TaskBuilder can be converted to a TaskHandle ONLY if Input is () (unit type) // This enforces at compile time that tasks with non-unit inputs MUST call .depends_on() -impl<'a, Tk: Task, Deps> From> for TaskHandle +impl<'a, Tk: TaskExecute, Deps> From> for TaskHandle where Tk::Input: IsUnitType, { @@ -74,7 +74,7 @@ where } } -impl<'a, Tk: Task, Deps> From<&TaskBuilder<'a, Tk, Deps>> for TaskHandle +impl<'a, Tk: TaskExecute, Deps> From<&TaskBuilder<'a, Tk, Deps>> for TaskHandle where Tk::Input: IsUnitType, { diff --git a/src/types/tests.rs b/src/types/tests.rs index bf4ce00..4471371 100644 --- a/src/types/tests.rs +++ b/src/types/tests.rs @@ -1,6 +1,8 @@ //! Unit tests for types module -use crate::task::Task; +// Allow tests to reference dagx:: paths (for proc macro generated code) +extern crate self as dagx; + use crate::types::{NodeId, TaskHandle}; use std::marker::PhantomData; diff --git a/tests/arc_tests.rs b/tests/arc_tests.rs index 98d3490..b0a0235 100644 --- a/tests/arc_tests.rs +++ b/tests/arc_tests.rs @@ -9,7 +9,7 @@ //! //! These tests serve as both validation and documentation for proper Arc usage. -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; use std::collections::{HashMap, HashSet}; use std::sync::Arc; use std::time::{Duration, Instant}; diff --git a/tests/boundaries/zero_tasks.rs b/tests/boundaries/zero_tasks.rs index 974cc37..0886808 100644 --- a/tests/boundaries/zero_tasks.rs +++ b/tests/boundaries/zero_tasks.rs @@ -1,7 +1,7 @@ //! Tests for empty DAGs and minimal task configurations use crate::common::task_fn; -use dagx::{task, DagResult, DagRunner, Task}; +use dagx::{task, DagResult, DagRunner}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 8723804..81e2944 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -7,7 +7,7 @@ pub use dagx::task_fn; // Re-export commonly used test structs #[allow(dead_code)] pub mod tasks { - use dagx::{task, Task}; + use dagx::task; // Simple seed task with mutable state pub struct Seed { diff --git a/tests/cycle_prevention.rs b/tests/cycle_prevention.rs index d6cf1a7..923bb3b 100644 --- a/tests/cycle_prevention.rs +++ b/tests/cycle_prevention.rs @@ -3,7 +3,7 @@ //! These tests show that even if you try to create cycles, the Rust compiler //! prevents it through the type-state pattern. -use dagx::{task, DagRunner, Task, TaskHandle}; +use dagx::{task, DagRunner, TaskHandle}; /// Demonstrates that TaskBuilder is consumed by depends_on, /// preventing you from wiring the same task multiple times diff --git a/tests/execution/integration.rs b/tests/execution/integration.rs index 7f539dc..82fab70 100644 --- a/tests/execution/integration.rs +++ b/tests/execution/integration.rs @@ -1,6 +1,6 @@ //! Integration tests for end-to-end functionality -use dagx::{DagResult, DagRunner, Task}; +use dagx::{DagResult, DagRunner}; use crate::common::task_fn; use std::time::Instant; use tokio::time::{sleep, Duration}; diff --git a/tests/parallelism/timing.rs b/tests/parallelism/timing.rs index 52b48c3..386cc5b 100644 --- a/tests/parallelism/timing.rs +++ b/tests/parallelism/timing.rs @@ -1,7 +1,7 @@ //! Tests for timing and speedup verification use crate::common::task_fn; -use dagx::{DagResult, DagRunner, Task}; +use dagx::{DagResult, DagRunner}; use std::sync::Arc; use std::sync::Mutex; use std::time::{Duration, Instant}; diff --git a/tests/runtimes/async_std.rs b/tests/runtimes/async_std.rs index edb55b7..1cb8a70 100644 --- a/tests/runtimes/async_std.rs +++ b/tests/runtimes/async_std.rs @@ -1,6 +1,6 @@ //! Runtime compatibility tests for async-std -use dagx::{DagRunner, Task}; +use dagx::DagRunner; use crate::common::task_fn; struct Value(i32); diff --git a/tests/runtimes/compatibility.rs b/tests/runtimes/compatibility.rs index b3ec541..942ae76 100644 --- a/tests/runtimes/compatibility.rs +++ b/tests/runtimes/compatibility.rs @@ -1,6 +1,6 @@ //! Runtime integration tests - verify dagx works correctly across different runtimes -use dagx::{DagResult, DagRunner, Task}; +use dagx::{DagResult, DagRunner}; struct Value(i32); #[dagx::task] diff --git a/tests/runtimes/quirky.rs b/tests/runtimes/quirky.rs index eccc8f7..43a1c05 100644 --- a/tests/runtimes/quirky.rs +++ b/tests/runtimes/quirky.rs @@ -486,7 +486,7 @@ fn test_futures_executor_multi_pool() { #[test] fn test_sync_tasks_on_async_executor() { - use dagx::Task; + struct SyncAdd; @@ -519,7 +519,7 @@ fn test_sync_tasks_on_async_executor() { #[test] fn test_sync_tasks_on_pollster() { - use dagx::Task; + struct SyncDouble; diff --git a/tests/runtimes/smol.rs b/tests/runtimes/smol.rs index e4e4c21..46e5858 100644 --- a/tests/runtimes/smol.rs +++ b/tests/runtimes/smol.rs @@ -1,6 +1,6 @@ //! Runtime compatibility tests for smol -use dagx::{DagRunner, Task}; +use dagx::DagRunner; use crate::common::task_fn; struct Value(i32); diff --git a/tests/tracing/with_tracing.rs b/tests/tracing/with_tracing.rs index 3cb962e..498d8e2 100644 --- a/tests/tracing/with_tracing.rs +++ b/tests/tracing/with_tracing.rs @@ -1,6 +1,6 @@ //! Tests with tracing feature enabled -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; use tracing_subscriber::{fmt, EnvFilter}; struct Value(i32); diff --git a/tests/tracing/without_tracing.rs b/tests/tracing/without_tracing.rs index a6712f2..94f713c 100644 --- a/tests/tracing/without_tracing.rs +++ b/tests/tracing/without_tracing.rs @@ -1,6 +1,6 @@ //! Tests that the library works correctly without tracing feature -use dagx::{task, DagRunner, Task}; +use dagx::{task, DagRunner}; struct Value(i32);