diff --git a/Implab.Test/AsyncTests.cs b/Implab.Test/AsyncTests.cs
--- a/Implab.Test/AsyncTests.cs
+++ b/Implab.Test/AsyncTests.cs
@@ -242,6 +242,38 @@ namespace Implab.Test {
}
[TestMethod]
+ public void ChainedMapTest() {
+
+ using (var pool = new WorkerPool(1,10)) {
+ int count = 10000;
+
+ double[] args = new double[count];
+ var rand = new Random();
+
+ for (int i = 0; i < count; i++)
+ args[i] = rand.NextDouble();
+
+ var t = Environment.TickCount;
+ var res = args
+ .ChainedMap(
+ x => pool.Invoke(
+ () => Math.Sin(x * x)
+ ),
+ 4
+ )
+ .Join();
+
+ Console.WriteLine("Map complete in {0} ms", Environment.TickCount - t);
+
+ t = Environment.TickCount;
+ for (int i = 0; i < count; i++)
+ Assert.AreEqual(Math.Sin(args[i] * args[i]), res[i]);
+ Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
+ Console.WriteLine("Max workers: {0}", pool.MaxRunningThreads);
+ }
+ }
+
+ [TestMethod]
public void ParallelForEachTest() {
int count = 100000;
diff --git a/Implab.suo b/Implab.suo
index 34b35ad721068c074ea216725753a1e954db220f..453df4ad2d602dd29c42a1de303bdb9f56e12f58
GIT binary patch
literal 75776
zc%1D$4R{;Xk?SN*oRC0B1BBliYEnooMY1i)PDqKYpAc{o6FWZzhh#0UZ4p@#lAPEf
zKq>SGD5WKZ0xcv2C|qgF(Q7GfY00HPq2&U!&9yYW{vAE|dUq6#pK@?CH#56WTCG>w
zeJjaWW}m;YcBS1n^Zw?|o0&K7*=LUa!<`SF@JB(DJ}OKV{(g9-FkSWC6o49bGgrQ!
z3h?*ChYz!_F9A4gr>aTs0!)K8I}xB3U<LrT`AmR1fH?rO0FD5dEjPPxB*0Ms3jmG=
zm<R9yfcXH&02~W&9Ki7aCjhtrngLEy`pPNr`9XksfKvfZ1Naa?1HkD33jsb1a0bAc
z0E+-V0?-KH0B`~<2516s1GE6N0R#X%0A2teKr6rn080Rt0-Ob~jQX}9oDHxXU=2V!
zKoDRxKnNfVuma#5fRzB}0-OiX4X_Ge9Y6;_C&2jtYXQ0dHUO*#xDa3?fC%t0fK32B
z0Gk0W0*C-?0k{UB7eG?#rBC_nhtC0kL4YX0#Q<9YVgN$`aexFs5+DVDZJ37748SnJ
zHh}E_BLJfSI{+>P*a>hMz~ul}0DPQiPrzRW+9Cw`W*|R&!+bgxdJXGK4F=EdyK`jW
z!9xdXkY_CS$=`JX#srjbC1CG|s}NDmgMUgMzcKsXT;;PHS}6^++W|G5hVrk2{}Rx|
zzE2zE(;!T#!7;#tA9-k97Jn4lv6FD76&fU{iwOJ=0&IgG?xNnWpB4dr9Je1<XfT5x
z`8InH`x^qb3G#tj)prck;~)6}8~<2`Z2aSM^HuNS_(%T0cmVPj9RDW+@QnYZN}hV-
zAIB(;|8rG%J<2s3|LSt+jeox&kALJH$OG8;$LHDj$7>w_9jg4+DA(sJpIw5?AMhG^
zzy{^|W6I}+$|v#y<N=$Nd*l(lLjL$Kme}tJ<=O8j?T+I<qoQxSDz6>N^-i7qzx35b
zZ|rwORzI}p>A+|2?Re2l|52w`0yBAE$xk&(0hnMh8++$P>(fE0cX*&PC8ec!M$AML
z@lG+aRUD8S>(@)EG~6$9yPWjbSRWXUWrkDIvbZ#yNr|z>`p)6rSTu5;G}@imD#e%e
zx|;hv{l0#;yU*hio5dM53^S%_!>gi^R3e?|&p6jd)5Bt{D>K{|O*n~!0V$Qe=uRYJ
z8=~>P#P+l^C}za&Xfo~WjK<?)Z%m?_&`3s#r=z{mSTr;0>_ke_@J&SeNC#a=DSWRh
zlSoNe&LeX2R0-8rFJS*&0&~m;;GiksT0PKjhk&odfQ$9g+=J;T<#{zu<T1T4YYf02
z@}LdyMh3XgAlw=8K7nxa^Iw!N<`=-<4j=&yTY>rn@I#!FiqUZf@INLan8lAWPGbT5
zIR7B;O#<F1P!Kn$g@u6sbR|!-`0Iom|JULVQ<?ifo=CvADWb&)KjN+9z>gz#^%VJf
znwb|&B+TdI0_%tJtUdt$Md*k1RH_*DsCsCz5#Y>TknDOunu<aR@!2#$Kgl8MX%-xY
zH)C+u1>X%rePCPE(_H8T=z*sK@E_mJz-O9j3j6uz#+?QnFEPN-PZTgp*!{$zNd9&r
zP!uKkCPvRoS?**#O)2wtJgHRvjq=kfsEHWy8>~f<`t&eKg|cjnQp;3&dMT}SE`@gT
zz@G~~-2km16}7^Orv+Ne0ssBLRa|h@2KR2b$6xXJFkA-!+TqzSJmG=WKw<tf3hmVk
z_+v%sUuc2<tMvt1{Hx&q5Y1wvFc-<RBw#gDhzmCY|HJhj%A{uT<65=50Dk@XM@>n7
zY61MX<~@M}KR(a+Kh8H9!W^fOlpBGp^(7c98ECfz(IRPP#3g_q<@4h?@aNZm0`eb*
zI+t@+m$Z|{d$j0G=u+k%p7uwX`^b7@{A?{2B>o}l=H&$ZIG@e7SpF{R;ltT7%Jtk2
zcLOy4irHWEQ=^6ckNT0lGvw<v(EBciUXMIj+s5l@1yeko2Ia^8jB7Bn_)%VBe9sAW
zy<Nk1){ykWxX$6jOG$Fn=S_0&#AoYh-BwQCw3K>EIdgOhYk}ezjd1K+JK^m(aGH8}
zvJ?876I#AKD*?B|lYN4k`<`8PdUE`Smlk;@2ml9Zg<KHaL`#d%3>f@$)dE-oaMcd9
zc!ehVE&!kSmJcZO!+$SO<0V>*dO{AX7jT5(eay*47_hAE@Eu}krEeXOJ5u7K8VJG@
zKKQ;Fz)MeKEt$pYFFV$9^n-HJQ))kOL2lTlE^5zKqNNr7+)9rK!J7`E)d$xOcqR<_
zV=rz7eAom0@I-*KMNqSgZ@H<DVNN&(Jb*U@B??oYLz=K}22>*&sR_b$NI1V9^{(aY
z$FfXT?mQW{jF8;CJ-g11=B=}%npHPa66AU(NqjX)^B>C<!nLxt%j7k>;-$_4Al2
zACy{HPC((Tw_fdhBmC)S&%Js6CpUMTcSS9GYQgo_AJ`G-2|m=@(fHzR+umf~&u;wR
z$fM%EbMD%H{lJf(5T6y4l4|pLYx;}RuN?f(z>Sw5ethZ!7eAu+TmzH}HPBmuc9BYz
zrnQ9&{@k>Na1%~eE+@3e2vE-E%8k#=qTjKENkxl_t%R})>KWD>Xhzysg}YCrmR$|)
z-VM3SYauCH%Z-o+)6OE!Wx_V7KctN5QbLf2dHTDlO*7OUe&L>Z@(SYc-|;_cMmvZn
zWoXQyB+yL~agwM(3@9It!ms`C7b9s&%1TIVHILdAR^oNE+Da(e{e;pV`jmG9L|2-m
zjVP76kff=fQ2&j3U6g68$X`s3KsgxeC_(afO5r~l1MLUlC!KIT2Y%EQEl~s@tR2KE
z(pt1nBi9&fz)mZ1{c{e~dLQAALz!dnKTK`Zr}Vi><HwmL0=;B_$~{c7cZX6R#pG_5
ztgQcV);LL7IOo@QH{1<V?`QU)6pgFGGB~V1$R=U6{A)<|m&XypfU4&O;hU5}UJVw?
ze=Wc%iUCT?{mk}zmQJ#OAMK`RDA!v1R~N__37UH%R5K>VKR14~1+(>Un4|y~aMX|@
zM>G@9L>bCW@=>#}IonId*3J?O_|XP+^|d_sab8#jzl*@5@@vFv0Y6G?=P1|3`XBN{
zw?SUWf3Z9`*O|qSC184t+?;_Hz7UOnoLjnRwvcEp#ME@^sk;x<sI^_~fv&)j**mNA
zSFPLWaw>(y+>sizj}@P_b9d#FddsmQD5n%&>WZKV@@odnA-KoYrUR~;X#Ih-xPb}}
ztxoW_AW7{$da{jHCr$7~6W#Hx2+E1I94%5#daSny%BX&t;i4ctO7K=cd3d1K1ZS_p
zIWIN9Z@dyf+D5Xg5ww1rUq#68I*A*l$<o4dY30gg?E8Z9Cax1|UvMuT_|r}PJ)1Tx
z-?e`p8I(~x)2<&1ty5Ui{;pl8X!Fy<t4?O#XV2zVSx#u16!E%Il4nHHpyKxd=o_LU
zvG>7zg0sm5W*J>CnQO)_>DZenE_RYI5mOdGZQ*_Q^)xNK^4z`GEPeKptuLs>r0jIH
zoDV-v!$Tx0ND7wU{_U^Ze--c!dEZ!d^gD9~_b$@IqF(plswWL~gR7EmVGBQg?c&%^
z%NL2v6*#pp(NU&v))B5<-&%2W;{L}sUNFyhM`DUTeR^Y$CG2SZWTT1EJR2lFzK;A)
zdP;4-LY)O^pV#)<HBWWiu;cfEeZP5PT4LoqJ=)W%7i2qhhw0g~SJ+rO{j4$bd*4|d
zk6iuoqInnZ-*eVCU;W%S_D$ELJtlOQihmW<+xgHq5`m0+-?F-^_PsIaJ^QBEo9ABl
z+Q1Aw+%mc+Yh(hLtpT$fR<5ydN%&*ypdMNoMTmf&tX$Irb4C-*^9_0A;;e~;He~e$
zPLeH~Y1U~FRzqGvsydN|R76u;r%rL5MTYwm>8GZwf8y-hpK9Cv+AoHtzcEdx&@AP~
z-$kX?e~`FAGCTR@6V-U)npeMbXxE`LPrUO!@gIKKa-4n-PSc>pt{<ZL5j77uFa}gT
z8?9#HF+ID}JpF}Okd!xsPCsS%@pX5<e&y)WPWQ~(cmDhVJ&N@hp}el><`{c;m$Sf2
zR1fFCFLCNQ*upGl)UdR3<*ZL3_K8leeAxPsX@zncq#Si<`@Dc7S6PRrZjLq7>M8nt
zH2Xcv|42@rhIav~#|U*(`P-PP9#C6#7UjsKkvzEp_{EC$u0Vr{b%<Gty5O^4MF9#M
zq?_a^R(>XP7B8ND-ovxczxkDeFYo`u1NZ6;c(!i->e9iN9zH;xQuzpjJot5Zz4bO<
zM<bqN{$Z+LKfS1^h1Oab|M~NlmIM6b?md^a-|)NrH|_IY@%-z;w`b|qYc}DqI7dhQ
z{PZoe-@fPjx9)B3dONVY)BB3<9JH0c{AP#EgLk;zTKx2O*LUy!L3rkq$KIveJ7UlW
z)9|9)nCR^>0-X%A+(t8!u@(yTk4Fi7`nNVA=XaK`{|izP(DoiDDvVRTFr@~4e&-6-
z^v_26mnMU}8Z7ufpf5htB6TY|mT*@4n$vdX+a2;bFiYtFafHBE|L8AspK_hab{lt=
zH=!@Zph<pvZi(ZU3MC_cW#;n@FMr~`ciSKQ#HxpP`5t91jhgDsN4VPB{J%~SH?)QS
zuX_*n&VO@Ged4Z1o@@HyMRRmVjXLQM%r0uvB+WV6VH#Gn+N`$G1MIg#=~5`Wa+AJP
zmJg#q7qTWlu_s)8GVBiW(Q^Z(5n|`qtG1CX0^dZ<%Ybkm{5RSS!utOLp)%vI*#v&n
zI<|7)$Nf5=G>N~(7(ciE7xxV@&T6dxEv^AxXo3GAUu8CJ^}AOKa@%}L`TUIXiK_DJ
zl<Vt-{5=F;fcqN(b^&}5;3j~Z0loxq3&5?)^IwL~uK?T*a0kF{fIR?r0_+923*c@|
z-uG$lzYh1`5cKy*{FCbYLVGA4QJ(pc^0`m>++QG1b@~}R^EAM}sGdc`_j3S0H+cUy
z@cSVFN4YT`JK1SGQ+uGRQ>!=aInMn0r|@JOA+>-k-5rpgPVGFTxBjr2qcb}{esk$D
znIGMCpa$(xC<PX`8nC3+>9K@z)KcKLKAIWX6_Wr_fKMu`YT5RYpHN&uKHjU?xpFfn
z)-x(T66EXIqoGh+%Dz0B^l~k<XJbWs$I8W=HLTS?80ldU2Y)LXYY{MC2RH5J)sdM0
zN4V+M)(-0IP-yKZ<+GZ0x21aLQ)MF_R~hQ4xXKQ(tj%I6MIyE0Q{@~}2Y$ejA-i!e
z$x2dQUp&=Pz31!2FrPCx^I1juC6RUw^cBpNx5t@PaAV#JODcIz%FN|>3TKY_>L@ET
z71z^Yj?OaFU}2p(Vpvo%SJ2bhTIo~fxa4UT=`Fp_ucvw)rj<$I*@1rs;xmc`Lmev-
z&z3A6v%A=E-tDK_WIOEZHQa{r;j#POdUUv0F%H+0<!HDUFijqqDP<;SbG&*lE|w@#
zH6_B`VREE`A+l@BUYmu=by1a;Yiwy)->dEtp(LyRK5subl#l`Q@G9(`!cs8{d2zY=
z@ZnPAYS}8t<mw!qwoGly)V54*%ha|^ZOhcQOl`~5)lec^rmp4^nU$%7p{@Yx+_k>N
zFC8oE-1&I3+UBhPeZJPb;lkLK|9F1vJ`VGG^E#5&`^Z-$PO=2b``z%ipLWh-;?$#S
zLa-FqqoXIBIX!x=21=9u`6P35PKKCoqvd?_=N`55Q8h}N-V*D9Pu^$ThId`mOV)XF
zwi_H>7{BmRp$-1pX^%;ecERJ`p8&-Sz}+Wd8W$KBz)!p79TX!1pY+2Y?oMx~x7rkI
zJ<@$0H<z>?yD31uY>3*|M`guzJz^eKxNF2Tzjfx}RAY3KBrek18In$9X%hR0)=g(@
zKP)sqZmF8t#Fk{Lvs{HGnM(2&oITaw3!2)r`GY=xz*%Peuxe$9$uf4)GgSYH8cm6-
zdsJ6eLq1zjYsi;!NY6Ujat`TP-l<`%Ew?4-_mAZ46m|$vqS&0zuuP}h)bq^48TSY9
z?+?;*6d$|MVtr*g!b5jVJ^0)&zxdrheD~*zuKxCm3-tWJPwc&X{#_jY??&q4dDDL9
z?VWwhzy9jr-GBa6NE)50gL4c&XfIuD`PLBwCmekCf#Z(-&bK}|Mf&xddVcqLG^qIj
z=l)kd!`Pv%t7{J}dAN7o@iR{Qdh^NG>g}N^al~@$A26e={l9m&{N;@f$DWI>dh+7?
zov~%LdTppim{vIcfimzMEBjFSSI)MCIhj_V#`_Ok1!j0T;r@dY1^LW|Jmyd&jSBMl
zC73n>RWY(xU=%9Z_9w>{F%x}V-v5F2%K04lQKDh-hm6Z#d<P1aAJ496;lO`Cz^9e#
z^?<JvFs+3)>Le-fJovu?zUd<9f<C{7PNvaMi!AIv!82@F`Ny2UFbywWt?3#~XQUx#
zAQ6j65j=h&?OcJ!7eph@V3fWPQ=?m&F1%2Auq%^_#s?bfhtiQmDi-a{9n;Wa`jiHr
zPxM4QUUyq_i{xr;+thfW4i)E!>A?)1w@_YOn{c+Bq2yB+NDL*#l$5GNY+cbEk}y@M
zyRlfMD4(HAhDAL3fn~78UC|oaM9nyptpTRt#Y7|%i914<2kD>xF;d_}j!--!rIM*=
zTEe3^(#}9kOs9iVIywNwD#ytnv#}ejl&hi)C9lO&t+G-WXCN3FNhVU6U{s7H2HF!N
zT~a0kErXKivB{LQEt(ik$3_E#Qe>;tw-V@0&t+eCBqDTV%1P{tl|$ly6i%du#0);U
zGTtYR%uj5SQmJU46dH-9G3RhJCgJE`tkZwGl2QZ+fsxRs^pM=ISY>rJ*laVAxaky^
zhnln&?|HMMC}4eJMx1(*&YS*JN*u+rR1oV-v^`D3e}<1v-yFp%s@<Fhq;;FO$t??{
z*453Nq2i5m+07B=q)MxCJc|1Z$@8h1m<L?)tFbdga@P$*&c~kHbo7nVU;g;zPdjxZ
z&d5pDeUv@^#u@)J^>-IHO#AUW{m*$HKXmOavkOyzi6ef}*loI44|3Sgb>vW&zZdv>
zNB!_NDSGtHwS#^4J3G7&sVO-Ked?vDs%_U+_b-`U9Lgu^^;FPrM~<QT({WyhaP%K@
zPLJ>U-g)4<!TRAx-`u%;+LTYdrn`eXhp^OmJcRSX6$9rF9oW5Kw`Z~E{NuKLO?TC&
zuL~^J9%w@hmccJcJ{>|o5;Bhkir4dw*)@#(aw};soh+L7$!|5QRPVAaX12x5wwO&6
ziy8BO)5g;Lb))S6#^*WAECqUEo?6;k;%uY%SJmtHrncM__Y;syUz+8v#ZoI+-G5O%
z7B@c64jwy}6<>d&fJ*$cYJN-Vna8vq<GoeNwPd|>A?d1Gy#r%a+8X&M{W-H1<aCu=
zS9n%e165NZ{gKo`ZFot+@=X?rPFt?QswtPAG(StIt4H0^tu?_VF8I4N?D4q5?r^K4
z)z{*6wD^NAN1NB(;%N1F!YzT8c3;@#xoqdkp=3<#?U6s6k#sVZxOit|Nze8~YHLqa
z{!#fziYYM`lVa(fzGylrW+H>hL?Q<7?A&>&-y86?wD?;cu275L(Gm`L9j&cizr*Ei
zX=-hDd0M=_mdn004y56aQi_wXuQ0;A&EdADfGgney4%B!mezpF;cxc&9IZjGFVGzD
zdfh=6Bh1+?r87OM>!Kt&Gx)Ta9*sxvD$PONwUw*3RAMNamN;m+WgMDNM0I~AB}Tbw
z>E`jE44aR$Z+&qb=o-pgGHh)BE*=5V%K1;*gk1bVrqE?)>;?q+Jk}w&W-A2){1pEX
z<y^D))$u2__*V$>IatX&N(!elN<4o{i@%HbcE%+3($8O(hIz(u<$7flej7+h>2y4i
ziS|dO6bDr|R*tHSnBE#m#4{<hEAr6vg^JNMB*w)7sgIutSB@mhwRA(}S}r2SBT_7e
zo+2CsU02Z#K5=GEWw1r#(TucBif6>s0MD#hxvsQLj15axNZM6-)-))_`zA{NR(ex9
zBW8HI)b8;TSrz&QrC5@y#<DWC%Nz2!yijA_)>gNpr8(5@XlwF?9N}=l<7)PYz5ez<
zmcZ~xPxh)fas4AbVmvxT?lCfPJe};e%2StDaXhqrgVEL$Zgz)STpmZL$?J2pxZ6Aq
zzsnPJczi9cw&p;SyUEq2qAjm=m0Ke!<u*=>a4+?sA5lyk?N6kJILQ6ldqwV$7>#q%
zdRrxjF*I`~D&Z<2Ee-X?Mv*!W+G>OvW=M!GpTMj5<AmTuT9ImH;ssQV*JKSui}&48
zBcFs2oVbfu{X%>ud*8SZ!?UAnbZ`Rq4E<8%4*&SjJ`70Tg{2s;RWaYHNWTE;;{+R&
z&)~#X8R-|~ui5*B`%v?VM$_O#TyyD@uTE&Q#oYqI39D1oFN0V0%~C=31q_x&b!B}1
zjp6<WZv6wEy}`6P#`rx(_(}ic6U^&>aPP%{Y5d0LZ{T@G%Q)~~18}8j{D%8)xa<G3
zLhk&b4MI}rA9L;!o|5YkS_M1-*H52?(jqATS@8Km!F>5Y53p4knQIAih<u399+43A
zVt^J-3jjY(+Kn9e)%vdhd0>qQNk#95e%a_*kf4^3$QwXz7yZ;|q5f|}P|9!I{@$W_
zS!w@!g<Si$D_&Y?vyN*@Y_#_Z6P~|?+=#i^n!&#WV3u-iRQ@)DysVtRg&Nh(Ch@l#
z<EL`7V`a_YPXa_u;%_q6zt^CLk(&emRf2p58KwZ>8zi5g<fOCQGLNA`<?656a^bVv
zmHtwwbi-9!I@W6JO6E#98Ztx~^W!S)9a7!CloKtcP<4Aa=y5nabI<5mt$%K{azX1J
z$?qo)a>|%m;mKvP^cC%lIkvB`C*D8a>yuAXI7(I83b?Ll#Z6g-kqtN?d}!Z7Hn~kq
ze**misMC?G029#6Z%loi0>K+<f=25PpEf&9!u1zMPsg~$X4fCc9fy@`WB;S#D?nWT
zV9fPq4*cKZz~5W}{mVun=l^89{xdHBM+JFJZ5ICzgxvanV!2;vDIME)ky9TF5S)<q
zeq;Jk1C9HHD#$sw^yhl>gm(6uyt@kU^dPhEXlrG1?0hlqy&qP6gQr%{!QMOJ{YZ>Q
z;#}o3IN_XhjA_R<z$gO23FWV5O!op-=M&ep&A1Ra1z}v23bJs^UU0_Lqc;GuvY_nL
zQtCdyjg>lFZ0)Kyuex;gUmVfvuKfH=*R8X3V`^;Wi+^K$z|>Dx-Z*8`zh8g<eM7rb
zGdlO)q#J`%9lxWy5{I8XyyvmQXB~C&j@!Tg&eRhgmY&h2LXLaD6@OIcXeeX(v-k=C
z<xf1F=xo8f{P{yc$nI`7#@|#dtE~Ok#ex4(fye$_+yZ1p{{Lry?I!Uzn<zhe1o#E>
z<;Q4(SFoGh(O}<ycGlj{KUVwI;TK+C7utXGPjq{oHmxoHF<<@bXCM6LpUyb*wi&Ir
zKjj(tH{FrX5=Z<e-hfD#JpZ@7_cZ;!;b*^Sb&THk`!fqfqQa@5v&?6H$!z+`GVF>+
z|A{vEUnmyS#Pns=_A+@IJ34#+%A*)_6xY$gDsy|)FWJ=W{oH*Rs|j~A1;@ohRxc*7
z)mT`CrgA|*4N+Kkj66|O`93}#WxQv{uG!a}`R<cb?>XV{<qvJ#qYgWU?b@rlSIZB@
z<*xwUBcpU@>k{A2#ev&jV*MG<KgRwCAx&A0(oX?iSFVlm8~cAP7X<Q9GK>EifLBc8
zFaP}iJiyzg@f+*^f`ZIJ%;N6>cv-m~@A0E|^6JBH|KQT4Gv5;a`+MJe{FdM`F}2`<
zH1-OG`<K;ezgrk2Aw;BwKt8uZU$%=XdfuiWoDT5VILz-xZHKx@J^heY+cnvuQ_L+;
z?%b1wB3M=1nJ){3VrRaTI(DBf!~AXK+=9x)-CaB9x%THgzR$jtS$Fz|+N}>2=R`r$
zhhp^iwVY8pjl-A<!UqJtb%i>xtv23XPI|zuvr(06nRZBN*>VxFc4F`F(=SNQ2mX~u
zaeQN8f)jaHn0_(T2^rF<*}Ld+`mh`NebUg`WJe9OqXybh1IrUNaPr#8?dW?Gr$pAG
z@7b~2?TCSu^QX6W3{CWM**k{bZ##zGtMe2lA9u5N3)#Dc?A=24ZXtWOknJu0{`MBP
zt%bbSLU!Iw?F;V31An^7zh~2i<-7LJn>lT<eBO-}ovGT=Q|#CLGjQrY+EV3b;GkWe
z=M0=&J(Q-`URGioby5FI61_db8uIx|(bJ<@j^P{=gJ*m+-B&{1pAh7jAXf`=Y$go1
zhB1;bkQ~O!LBLbeFmBNG0G|W6L6GD4V1T(BmFMhKeW|>b4tH5c3cBkfcg>9vhG$8I
z@f045R%|(H%=x1~pHo}4q4Dz?hOMncY*uV{;}Ov$r<D#19{6*?ryHP^G|)ci-Coj4
z`$+?hnrbhdbb!C&^QfB+0JOuiVR*u$(n?PbNv?V|w)rx5yV4H|mCHWeGruNn+vS$3
zDXCf8W%Z0*K08^nh1tpMJfZe?$;}onsGkt6SiH(?fqsdtFvfAVXeR$n2l?Cj;2Le1
zxU0cKUWtCV_QL;m_@?@8nD1{(Sas_s^{hefW&F7D8ppFVbjZJIsj(#jeTl$6?|Xdh
zWz||@+v;OmeQc|bZS@&z^{F0?R$PaG98Ld;LROw?+Y@%OP;5__ZAE!+_Y2rsbX$vV
zR*OD4W|->LXxZ9mTN_=9HrfiWH{+{2Ui!@~Rnt;twds}LDN}x%WcBb%J_|s#mdn<j
zS7YfW-d>9Fm>q1bxve#Sulc^MHMh0q)m1uMYi?`JZLN7}T65eR5ejs5+U@_|mDV`^
zq;dT3uz=^tnvXw;){rZeYh(Q7KY#iH!F>Gj<_es@yG4-qTa+vD;#Vg;J88XzHAa5q
zI=xsw@>w@LdF#i~)QdVON3`OJx~JB+=VJJ>?{YMwywXv|XiD-)$6HFSh2n7w@f<F7
zRA4Js2yD?l0S`u4M-d9;?{kp@=Ww^C+2alb0i${IMqJI$d8M*;DjZQAB@+D@)P-;q
zj%SUwPrjJ%#z#-cR@v?nYgjL!z95_=2&A{=tzRvL)40oRCeq~z0`lZa*#qU%`gq2)
z?m-zsOjQqf0@+zyIif|eUzNj5OMKT{j8Bxr2FJ<%DM#i1Z5b@<?8+@{hCU`A#}Ea&
zum$`yyI`yhhTTE>A}sR`cq&Of*33JyJ>G5Vc~-)sEcH*b5&UFNj{{|>!HE+`7v&<}
zF_+JAa#9$}@i@*Ij`z3^0i<Y*O<!H@P^>&?Z3<eNPqcjhc#Be^aYV>!KR@NYK0m`q
zN{<P}&yTP=j_mmP<vIJH+T(}w<XV^;u-q4L&a`|-$_W&=iATbAs#b`_*Nv#v>8-fe
Wo_uO04<2jYWh|n%-Tsrk{r?{w1?ye_
diff --git a/Implab/Parallels/ArrayTraits.cs b/Implab/Parallels/ArrayTraits.cs
--- a/Implab/Parallels/ArrayTraits.cs
+++ b/Implab/Parallels/ArrayTraits.cs
@@ -39,26 +39,14 @@ namespace Implab.Parallels {
}
protected override bool TryDequeue(out int unit) {
- int index;
- unit = -1;
- do {
- index = m_next;
- if (index >= m_source.Length)
- return false;
- } while (index != Interlocked.CompareExchange(ref m_next, index + 1, index));
-
- unit = index;
- return true;
+ unit = Interlocked.Increment(ref m_next) - 1;
+ return unit >= m_source.Length ? false : true;
}
protected override void InvokeUnit(int unit) {
try {
m_action(m_source[unit]);
- int pending;
- do {
- pending = m_pending;
- } while (pending != Interlocked.CompareExchange(ref m_pending, pending - 1, pending));
- pending--;
+ var pending = Interlocked.Decrement(ref m_pending);
if (pending == 0)
m_promise.Resolve(m_source.Length);
} catch (Exception e) {
@@ -101,26 +89,14 @@ namespace Implab.Parallels {
}
protected override bool TryDequeue(out int unit) {
- int index;
- unit = -1;
- do {
- index = m_next;
- if (index >= m_source.Length)
- return false;
- } while (index != Interlocked.CompareExchange(ref m_next, index + 1, index));
-
- unit = index;
- return true;
+ unit = Interlocked.Increment(ref m_next) - 1;
+ return unit >= m_source.Length ? false : true;
}
protected override void InvokeUnit(int unit) {
try {
m_dest[unit] = m_transform(m_source[unit]);
- int pending;
- do {
- pending = m_pending;
- } while ( pending != Interlocked.CompareExchange(ref m_pending,pending -1, pending));
- pending --;
+ var pending = Interlocked.Decrement(ref m_pending);
if (pending == 0)
m_promise.Resolve(m_dest);
} catch (Exception e) {
@@ -148,5 +124,48 @@ namespace Implab.Parallels {
var iter = new ArrayIterator<TSrc>(source, action, threads);
return iter.Promise;
}
+
+ public static Promise<TDst[]> ChainedMap<TSrc, TDst>(this TSrc[] source, ChainedOperation<TSrc, TDst> transform, int threads) {
+ if (source == null)
+ throw new ArgumentNullException("source");
+ if (transform == null)
+ throw new ArgumentNullException("transform");
+ if (threads <= 0)
+ throw new ArgumentOutOfRangeException("Threads number must be greater then zero");
+
+ var promise = new Promise<TDst[]>();
+ var res = new TDst[source.Length];
+ var pending = source.Length;
+ var semaphore = new Semaphore(threads, threads);
+
+ AsyncPool.InvokeNewThread(() => {
+ for (int i = 0; i < source.Length; i++) {
+ if(promise.State != PromiseState.Unresolved)
+ break; // stop processing in case of error or cancellation
+ var idx = i;
+ semaphore.WaitOne();
+ try {
+ var p1 = transform(source[i]);
+ p1.Anyway(() => semaphore.Release());
+ p1.Cancelled(() => semaphore.Release());
+ p1.Then(
+ x => {
+ res[idx] = x;
+ var left = Interlocked.Decrement(ref pending);
+ if (left == 0)
+ promise.Resolve(res);
+ },
+ e => promise.Reject(e)
+ );
+
+ } catch (Exception e) {
+ promise.Reject(e);
+ }
+ }
+ return 0;
+ });
+
+ return promise.Anyway(() => semaphore.Dispose());
+ }
}
}
diff --git a/Implab/Parallels/AsyncPool.cs b/Implab/Parallels/AsyncPool.cs
--- a/Implab/Parallels/AsyncPool.cs
+++ b/Implab/Parallels/AsyncPool.cs
@@ -36,7 +36,6 @@ namespace Implab.Parallels {
}
});
worker.IsBackground = true;
-
worker.Start();
return p;
diff --git a/Implab/Parallels/DispatchPool.cs b/Implab/Parallels/DispatchPool.cs
--- a/Implab/Parallels/DispatchPool.cs
+++ b/Implab/Parallels/DispatchPool.cs
@@ -63,9 +63,10 @@ namespace Implab.Parallels {
}
bool StartWorker() {
- var current = m_runningThreads;
+ int current;
// use spins to allocate slot for the new thread
do {
+ current = m_runningThreads;
if (current >= m_maxThreads || m_exitRequired != 0)
// no more slots left or the pool has been disposed
return false;
@@ -84,24 +85,33 @@ namespace Implab.Parallels {
protected abstract bool TryDequeue(out TUnit unit);
- protected virtual void WakeNewWorker() {
+ protected virtual void WakeNewWorker(bool extend) {
if (m_suspended > 0)
m_hasTasks.Set();
else
StartWorker();
}
+ /// <summary>
+ /// Запускает либо новый поток, если раньше не было ни одного потока, либо устанавливает событие пробуждение одного спящего потока
+ /// </summary>
+ protected void StartIfIdle() {
+ int threads;
+ do {
+
+ }
+ }
+
+ protected virtual void Suspend() {
+ m_hasTasks.WaitOne();
+ }
+
bool FetchTask(out TUnit unit) {
do {
// exit if requested
if (m_exitRequired != 0) {
// release the thread slot
- int running;
- do {
- running = m_runningThreads;
- } while (running != Interlocked.CompareExchange(ref m_runningThreads, running - 1, running));
- running--;
-
+ var running = Interlocked.Decrement(ref m_runningThreads);
if (running == 0) // it was the last worker
m_hasTasks.Dispose();
else
@@ -112,7 +122,7 @@ namespace Implab.Parallels {
// fetch task
if (TryDequeue(out unit)) {
- WakeNewWorker();
+ WakeNewWorker(true);
return true;
}
@@ -122,19 +132,21 @@ namespace Implab.Parallels {
do {
runningThreads = m_runningThreads;
if (runningThreads <= m_minThreads) {
+ // check wheather this is the last thread and we have tasks
+
exit = false;
break;
}
} while (runningThreads != Interlocked.CompareExchange(ref m_runningThreads, runningThreads - 1, runningThreads));
if (exit) {
- Interlocked.Decrement(ref m_runningThreads);
return false;
}
- // keep this thread and wait
+ // entering suspend state
Interlocked.Increment(ref m_suspended);
- m_hasTasks.WaitOne();
+ // keep this thread and wait
+ Suspend();
Interlocked.Decrement(ref m_suspended);
} while (true);
}
diff --git a/Implab/Parallels/WorkerPool.cs b/Implab/Parallels/WorkerPool.cs
--- a/Implab/Parallels/WorkerPool.cs
+++ b/Implab/Parallels/WorkerPool.cs
@@ -10,20 +10,27 @@ namespace Implab.Parallels {
MTQueue<Action> m_queue = new MTQueue<Action>();
int m_queueLength = 0;
+ readonly int m_threshold = 1;
- public WorkerPool(int minThreads, int maxThreads)
+ public WorkerPool(int minThreads, int maxThreads, int threshold)
: base(minThreads, maxThreads) {
- InitPool();
+ m_threshold = threshold;
+ InitPool();
+ }
+
+ public WorkerPool(int minThreads, int maxThreads) :
+ base(minThreads, maxThreads) {
+ InitPool();
}
public WorkerPool(int threads)
: base(threads) {
- InitPool();
+ InitPool();
}
public WorkerPool()
: base() {
- InitPool();
+ InitPool();
}
public Promise<T> Invoke<T>(Func<T> task) {
@@ -47,11 +54,20 @@ namespace Implab.Parallels {
protected void EnqueueTask(Action unit) {
Debug.Assert(unit != null);
- Interlocked.Increment(ref m_queueLength);
+ var len = Interlocked.Increment(ref m_queueLength);
m_queue.Enqueue(unit);
- // if there are sleeping threads in the pool wake one
- // probably this will lead a dry run
- WakeNewWorker();
+
+ if (ThreadCount == 0)
+ // force to start
+ WakeNewWorker(false);
+ }
+
+ protected override void WakeNewWorker(bool extend) {
+ if (extend && m_queueLength <= m_threshold)
+ // in this case we are in active thread and it request for additional workers
+ // satisfy it only when queue is longer than threshold
+ return;
+ base.WakeNewWorker(extend);
}
protected override bool TryDequeue(out Action unit) {
@@ -65,5 +81,10 @@ namespace Implab.Parallels {
protected override void InvokeUnit(Action unit) {
unit();
}
+
+ protected override void Suspend() {
+ if (m_queueLength == 0)
+ base.Suspend();
+ }
}
}
diff --git a/Implab/Promise.cs b/Implab/Promise.cs
--- a/Implab/Promise.cs
+++ b/Implab/Promise.cs
@@ -103,11 +103,16 @@ namespace Implab {
/// <summary>
/// Выполняет обещание, сообщая об ошибке
/// </summary>
+ /// <remarks>
+ /// Поскольку обещание должно работать в многопточной среде, при его выполнении сразу несколько потоков
+ /// могу вернуть ошибку, при этом только первая будет использована в качестве результата, остальные
+ /// будут проигнорированы.
+ /// </remarks>
/// <param name="error">Исключение возникшее при выполнении операции</param>
/// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
public void Reject(Exception error) {
lock (m_lock) {
- if (m_state == PromiseState.Cancelled)
+ if (m_state == PromiseState.Cancelled || m_state == PromiseState.Rejected)
return;
if (m_state != PromiseState.Unresolved)
throw new InvalidOperationException("The promise is already resolved");