diff --git a/Implab.Test/AsyncTests.cs b/Implab.Test/AsyncTests.cs --- a/Implab.Test/AsyncTests.cs +++ b/Implab.Test/AsyncTests.cs @@ -14,7 +14,7 @@ namespace Implab.Test { p.Then(x => res = x); p.Resolve(100); - Assert.AreEqual(res, 100); + Assert.AreEqual(100, res); } [TestMethod] @@ -101,18 +101,18 @@ namespace Implab.Test { public void WorkerPoolSizeTest() { var pool = new WorkerPool(5, 10, 0); - Assert.AreEqual(5, pool.ThreadCount); + Assert.AreEqual(5, pool.PoolSize); - pool.Invoke(() => { Thread.Sleep(1000000); return 10; }); - pool.Invoke(() => { Thread.Sleep(1000000); return 10; }); - pool.Invoke(() => { Thread.Sleep(1000000); return 10; }); + pool.Invoke(() => { Thread.Sleep(100000000); return 10; }); + pool.Invoke(() => { Thread.Sleep(100000000); return 10; }); + pool.Invoke(() => { Thread.Sleep(100000000); return 10; }); - Assert.AreEqual(5, pool.ThreadCount); + Assert.AreEqual(5, pool.PoolSize); for (int i = 0; i < 100; i++) - pool.Invoke(() => { Thread.Sleep(1000000); return 10; }); - Thread.Sleep(100); - Assert.AreEqual(10, pool.ThreadCount); + pool.Invoke(() => { Thread.Sleep(100000000); return 10; }); + Thread.Sleep(200); + Assert.AreEqual(10, pool.PoolSize); pool.Dispose(); } @@ -149,10 +149,10 @@ namespace Implab.Test { [TestMethod] public void WorkerPoolDisposeTest() { var pool = new WorkerPool(5, 20); - Assert.AreEqual(5, pool.ThreadCount); + Assert.AreEqual(5, pool.PoolSize); pool.Dispose(); - Thread.Sleep(100); - Assert.AreEqual(0, pool.ThreadCount); + Thread.Sleep(500); + Assert.AreEqual(0, pool.PoolSize); pool.Dispose(); } @@ -244,7 +244,7 @@ namespace Implab.Test { [TestMethod] public void ChainedMapTest() { - using (var pool = new WorkerPool(8,100,0)) { + using (var pool = new WorkerPool(4,4,0)) { int count = 10000; double[] args = new double[count]; diff --git a/Implab.suo b/Implab.suo index 453df4ad2d602dd29c42a1de303bdb9f56e12f58..2e658bab65aab728cc1f2d27570c91f2140bb05f GIT binary patch literal 75264 zc%1D$4R}=5l{Y{jAPQ1M#J`AQ1S1pR$6QQx3W~Nc3ISJ*{!YHTDM*Ox?PukRzBBOv*+CR?#!E)H}l?| zpD+XS@}0b&dGDTc?>+b2bI(2Z+&5l7^_}niXx1Ny>i7Z~OFkQ(NX9FkjRB}&H&dni zu>hY94-d1)R{{)|kCHh)1sDf)b|ye2zytuS^N9de04D=X0yqg^vQ+Kl6o4rJGXYKo zm0Q%mDKM>Hw?&Hh{AM+yHg}2Y?fx0bo7Ae1HW2mjEoJyiLd=fW-hS0h$0j z04o5z06u^v080Uy0WJkt2G9<$9AFJV3qUKtDuC4hZ2)TlIsh&MSO*{gd=X#+KqtUP zfXe~=0Gj~53eW{0%H?vSeC~mB0H7Bj2yg|!W`GbtA3#4q7$5==1;9Ft!8r~v05Ax! z1z;<{5WqHos{pnGTn%syz_kEhqO^zMFAjC#g>>VP9)4l`m;qeFc&SF`-FF_|vb^J+ zLlr197CWV9t$;BMIcx^(J#givRP*2;lOk`7em7M-z?XH2F+gtG9%e-FSQaJYqXzxFW~@MF9Eyi9`;{3y5C zi^y*XSSKh4DizNdDBC~E1J?er3|aff_ogep#rBW#fyn@rFWCOEonrexml$jR3*
~T*8Y|G&}#oKBDH^%9Vi1>`^R>|+CP4a?Y~8l-b(p;m3+qb-!5OH z3|K2)e^EYPCZACjpbXe3-=mD^lCQf-vi;*(+5YjqPd;NG8r$eovW z#5`ixK%g}$#>D=(5D$j?TLu4SAt2UPcZks#+%L3R%=E9e+C3184@AX<{o+78DuimQ zTL-#ALH{yws6D({>|fYrsqeA(ID4$tZo5UO7ba9N%y^6&SRV98!?AEr+}sh24G5vO z_&|3sY^Edx#AxE8Jsb|L4fc13x5UgIAuhBBBQbMpu)kmE3W@Z_yEQKM$AVqKP%u7Z zZbeFC@Wd}(+(IwJD1O%#4@X7J=SfL@Du?2$Cdj|@p^sS$d=Uky)d_st2XZ9@Qml*i zJ$Ou!-&f*98Pf&a6M#RIL2KcQI7pvfxYLt;gyQCxzo=hKPlLY+Km-~#1NC9xe(aMn z(QzTjKc*uX#g95pZ5sU8|Df!R0Nx-_(63Vpa{&JZa+*f*SCORrSL655Ja>aW5r(Hx zN{b$T#9PIIA6x8-G1B!oBOe$zFr6@L@I0Rfuv_HPqNvkjxIy>^ec4 z3PKL?-55X*)k8XHFE{{ShTyIZo(({GU|m$xzR(QN32(XKKYkmBbBvZ0^7+?Cod#?# zA;8c>DPWYa`{JQUdRhz=1*v`$qVJ0d>6HC3CeP1!j$Gw8>QBp|Btlf)U?~cerw6E3 zDCx$iwTz{2FQ8+ctDv6j@MnRu6`%pMq6Qf8*r2vd@ZSYe#R6B2aBqcs{1xB#!L=Kp z3EuU=8+Oog(#xMAsIM--AIeDooGj$O(q5p(za0Me(Ozr_`XY&zFpOr>N#R<+kK;Ym zOO4{kv1)r7{M!AGl9JTJ2J#ojyt6p)d>fEPP>*Hr*F%Oj=f@WG~ zz<&xSer5jKD2ERu@+g&a58MT4|0^VZ(LQRk;D5A_%#;~mC2-#r!1XAD)pguKM=+U> z*^qzaXB>kW#gFn$p|vyy5*OzI2|FQA&EwrrBP6W^_(LDE#io2|faW~lk5ga+IMZ*~)<^j(y9dXn;omlmlfaDxPCfK(8y zlolJQ2MjKHWdkg3xM~7g9HfpubHf?Gasq`e`0oH}9F!J4PDo;P01h8~k11Iw2Fz;{ zJVOi(^r;C_M@pQu1U&GD6Q0)tIOyA0N=C7|@{Tnhe2`Cia^(XHq=t2Bq4n87X=#8z ztIQEz_|imab;7j?-tj^D$i?-54>`aEZ@4*Y1TDMxm6h@sri5+44tTwgBOm2CqzQS` zt!U9mjR&s1WK}iVUGvGul1`TIG?}#YQ@wdhVw@dJ8D|Goqi&?c!}U#s%GC(%_ps+d zO@L|0O`ItSwEqfGJ`X9?+V^%19%aub*Z!aXMPdKay<2Vz{OlFs4I<}MozBPO-y45j?|-{*zh?O5vHPxg zUhBOY$Q4?kHv{bg%~gz!EiCY7rDF&y#mVwzh8o!ll(W8a-RqO+@0i1gY(>RdLR|&z z3>`X_k*4Lo_A_bCu7G-Pht#F9keDdtT1bOgXA$Q@G6>~|lrdY17t%0Je><(yIIRyC zd0?6}f*Ae`|D$EJh03HjZF8szv{Q{ZLa9Lvs2>i&uRZV=qS}&}(2&?@9<3`Z$7|?l zD=b_0!!kc~%ip*uT`{U{1Zl3*X`0#x+P~4Ri#km<`WMq9P!Gm33RC?%D$Ad^j`f4$ zr#9hg4*X~o8$7s}V_`L8z6F~YR(@zatq z(Ehpc<6JNs|N5vFU;&BhmGy{vDl<`svQqu1o@`8T$;i@~p9OxLd#byZ2S4@;%i(tc zWK?R2II_Tx8rxF&Iurk)Otk8xh4L5Y)UdBJiXU^p>=?;D184ZWwEbh>(nfm=k@kgn z)SY+U?n4zyYgd!I&3#JZF5CXA(rUJt;h@cMA zs|UTI>FyORBLzAHyi2bqz>Mwqj!8Gf_!4lM~&o@o?L6h zMXB*p3|@K63N9(-6yLY(uY*@Q^=YeT-LK)RF_4q+OzW77@v-{jn zy>aSev&f#;rVRfqoxD?jryjqk@o3SbJe2RTXUS2P8R{%bW!(_fX9Q|N#piC|4MEn} zyP-e9-ekQ|LDxm6Yeo*~*q5j-wo+vxBoBZrOU4au`{3=pYkv9XM;|`;Y5)cNu<9U- z`S4>m+(%UfQO2_K-5Wzs^zR+qckVY9+_YfIvsNuE%5e{ldSXyEI4Wr;oA~jo2gfct zd=cq%1$He=b(H9vbdqJ~<4aBrKl1Xr_0yd9hR0~rr`4+2fdk?mI!6>sa1dx*5>ihntj+bXCWk%EMK|H7&p_J7ptSae6|=%u%Q z7?_}iTSE70g+u_GYrrH2s{ofhpzCYc)`?GyNzT<*N#yz)4 znQfH9T`BiW9{(c8{4;GNMLz&}v81WZU1MMzg5@Xah)(#=UxM>__@@odJqamy^1OND zmpwgs)tyHUe{kTPeGh5!i9Bw8a6#|;PamS5QmF_%Joq(aeXbn9(f=^huZxNb*(}7- z{?w+|bQ0i)yZ2n(wBvUN?%40R_N~8>Cnjl?YvMq&)KU%e^TE3&fAYYyci&gv_KACU ztK*1fAGDc2|4J#gk_yWx{e39$T*CjOdo3T&JGjO2)cw!-CcZjjuO@edNC&*yE>|Xc zdu#)BuG zbkeGk{qK14kG}ZKiqtOKSbPcVYtq`8I`6>j^3fEZ9Q6NCD)vV3qrc3<^0iU|KU%qa zc5_=**IPR_?0xo=BY(*xftmchG{^Q!GbN!vGV!gN53Yas)21I?zx?T)&KIkQX1hy8 zxFWm$e~qkeuqRfThQB;?xNG{+ldHpfpMSINh09OYY&FUwlSKpthl}!ckTz5|2=4%q zd?nCdDZM>mflN+BZXFNJa0PDoy_Wnwlw<$b2KRos6%Nw52>cQ)F9EU){_Cv*J?ZM;`vi^*UcRh~K7t9*T%eBJ@)2LQeaa67 z0_+CZ1Moe7`vCR=+%LcTpnQHgm(K^%q^Ugq z8QwVv@GpvYapL<;fM4o-{~P%Iw*YS&z*9cT$7mi_I$L3d9!I)j`VOkA5qXqL`!$uL zGIAl2 z#@yNXo0vyX=K>sYI;Yk}`w=lQ{6I=gS(VF3^-ODcgXz!7Q~JpiFV3MRnORzkT93D- z8Po3~yel7&dwXbFY*l8psumdMj9eqpsZqO`YQ8g3)1B1+URn#<2T#_)|4^cxDA#cD z*K;QTY8q1Bn%r%z>iLmLla1Ry($|eK{WaSZ&50KG*n-pIKizw%g3U*#*Q(i($(}FG zgEMKUmW#u~Bjlp$$Rrmf8zz&BNjlZ}%uO23<1FK3OArlNVi5;_Ju*L&)KP4R)?l_0 zi*G`LU%ud-x^&QHhjOcjwtFS*)~w2%zyGo1p^P#c?MthdJ1?tpPo2cXbS~sf2j_3m z@);mWzrm5I@^YNfDK7s0SDo|ZI5U~zIh-k`8lFy1Te9HADge4Y{!0&Asa)N@gEv z!cDoXkJhh~<`u{FhctQ_1uu zWlg3u^OatcDWq({-c$LUs7kXeAGGBI_A;Y~RjorDFKrjs1!})hp~`V-w`!x>e0CP{ z$wX-(U%;U~Ygo%Uv}Y-%j=r_LtT~r8=k%I$xraGXHBes5JJW0Q=H&`GUOHZS-s4|u zE);i_HRiI$Jeut`*Yzv>>pwEjJ^yOf(4)Iv5Ei(4_Wk_jx9{nZBG5?(mbTId9aHTn zjB54dy<-o*`Kxa|_0ChjoO|OF@6FWm13$Ctn(2Ev{NME^V5g1youg~=Y5(@`hwuN> zSG?lTSPh&bH0-9_Cjvh{DKP8s8~aY5@sr2T9V7nwsFvS-3JppPn{)jupR37lH?OGt z?fj>^)|@e6_M`P@-=ejKCPy8ZWBq`U>Cb_&Z#3q;^w%F)NTvhVADl^~ zJsVPlLxuxK;$it(@@&L@0;5n#-Zaum8Y9uyrS%u+4>6qsKiUXb{2~4PXI_DV`NzF0 zY#jI>0k}!N?f`tPfN3@GK`X7DW$=FqJZYoQ2Hdie?o6Y7%*`VH5bj~a@;_q#g>iWC z*NV2GSX}HgyThT7=*R6BV&)~deL>J~_5|qzAv(0F?y}3|7u({|V1J;tx-aGrM?=A` z2aYOB88dNM5V!d(PZy~?9Hehc%K~AU29qx+=Q88MD z*xG{IL^76C-JYpXtZL)%q8Q0U;!e_((%SofY)!Kx zhqpg2MkCQ+OvJ4?VrF+ph{Zf&EEs@ng~=arbN<=JG$?DlxKM#9m! zCn$u%fu``*HZdNDnn6u;MkFc@2Ezlf(2%=V^luisn}OcgRQ9+f?5A6%oW&kA_Xz>f z7moG`aeTA6zgyfoJv=Bzqrq;`yEPcYlzqXFh^>E~2LH50L_ZJ$EumZHkmRpeWK|Vd zZ4;5W@icxYtApICV-Sm|a$^jT zR&5-VY8FVXs+u@KAseT%o0E)cmAvDhOrCA`d!7+)UT!zP{Ls|xz3-gVn`s5EWS!^I zU-jJ!BRhBWnO}Ny!>PB6fBwUFZZc~|oRJP`HsM(N&vV328o7QR%RvhJxrVx^r>+Hl zX37h@#NerSR`zy3Vs3FfsSGoD7V< z$B?P}l9*}ihA5qvIB{lI0wt3p?Ui&bv<|O4;ry3+OD11h)_gaOPmD4%a@^G%v+||O zS5L|23*X%$TUXNMZlGjy$7+pTVf>_cVqYh;U@=S9i8DKTEPV=(KM5!85tsR0wSB%+ zrC|0xd?%6wIcG$v@T79Q@QS*}9LX!}jT2|~>iE+XrfAfv@;!C#kKa+{X(LJ^{Wt1T zNf-HK5~->t}R!DR8;Tqc{(?JzYoI9w)+!&cW&Z?W4PPTSSrWu!UkeT{W)i`(R|Hu+4p z2DioJs&_g~4IYQnUGH`{tR4#^&D<`=;+=}?j6|B___h!m>i6SSjDxyc3RiH^a9=Pc za?tYaQK&@xQH*j{;a#IZn1JT~cvJ{-Ro|VXK^gKDXNi4l6zJ-UUpb&Z|D8DkVmA9f z4U%O1L1xfpd+fT2v>$69Tr-bu9sCu`;Ea}0qxhBaC)N0u5NRK*ND3wCj|+3$e@l(O zjY{yiLF~Ejzbp>$2$;fgWe|P~M8#OFKO7JC1jQ%^Rks(8s<;r_><{lj1!Zu~3M*i*DL1xYe=sf%iv4jR z8sOLTEtDhNPW^7froFe|NF)x6DnkxDe-IsokTcOUv~2ilGQsi6xYMi^J=* zIH1HF4GmV4t=`*YYOHg5O+KI7ZmDK-Ez$HQd1 z3Qt{1!ST@c14dh&uiomlS?nfnox^FeSsU#pm&NWe*_}2^W4*i1T4!lg(3VoW@~s|~ zQXR(xxEDLok0>Mz^@O8+9OQogSdrT&1p7H@-BrkD4AmSDia1J$iG5w6A*7CjwhB_g z-2Z}$i+B`&I`I_KP+6@L&!olJmN1*l+kZs`+C>QQ6z||wI}H{s=(OApSai`+EiY*0XJQjo`n5 zNPCdsQS7s#sXHIw+5)ZL)w7+)n|JEeH2F@xnQjGZsl|N4eZ=%*K9zTkDO{YU@~$x_ z#;!5Pub)hL^+oaID7)1c<=9=LRVM{-T~T+ok_jUbaDbe@pP4fXXwS>VpFleU%HtH4 zfg+sd*QdTp2G5QPkKXvhsZLIH)MPRK#FjE3U+eoHWgY?I_y>0}K`WI}{Eu_suQw3? z1o1B7z`u?p{h##5f4cc6Um(($+9>|#NOJsNT<#ZYN=Nc8jxJ^Zc#3kqU!Q)oK;!tS zfRcmDeqLTNSgJjDepU?XbMNOBPjJ_2x^sEa`;q9k#HsSxQ*=%``m|#mpuNUZRQ_uE zbkAgEUc9bt`kBBk2;-s@P=sgg1*cCvdIO-4@JLQArM`b9SNt3OEgOH{eEXOU|9RUZ z5BKekPH4UF4$T;xN=s>bA+{5Jde2M4mrOZ(+da?zeeA5K#n&~dkjA=P@kcc_Th!M- zGmijJ|3!xYFEOrvzCcLA>Z^~xE>lto{oBHU|3$(x|D1USXigUMPyYn4#UTEA{rq!} zzc8BMwd^Lj+2*rvOuFwIFI7G`@W;Qc@gBJI=bBunK33;{-~#g$wT~O^bu+trRe49qH{g4ZRg}$FZ%S=u@B4|zUIkwdoo4tQ|b@7h7UiG zWBKX#Kl=U$y#3EF0RARl>*Lq=|5{9W)Oi2@I=~Ub_;ve#?*8{JfKLqL*SCLq;JlO= z&wnSt2lDl3w;wH8@bo9oUA181$K-#WdFJK2Jg0>+#r#+2rM!Q6Zu^%Pr(LTYF>16T zM&-JHeLnUtuv0I`h9cN~WcA5!Hoa+ge(nAEnhVxeZhkVeB*GCmZ-8Aa)!A9L*W7jf zksDa#X@!(pEN4(_i+hEic1Dsu@JC)mk2%bYrB4=442}}E1$*bGrz!i($_^cYq^w|Q)@r9rDnT^j-S^Uw#FQ>Wn(uOjfnQi zRVhAVhd&FPtpE*l1&R~6+d)^QxabNLT&d!qaXIl_}QxB;5rT_3z*SFB1oK6>)f z_1I0-xte4?NSCkj?(V52Szf!isbq3$bnRm4v|YZPmFlzTos9Mq%JuHa-ogUqT+Lvz8k1O7L`lTu&9c%tu}l#6}pI{5Z_ zBE`gEpge9NQoIoCnJ{k9Z2(^f*g>Q?J{VxGn~ZeOEFYzD9DDnu)@;zR+&^lx5L z8|h|wN-ZYb*~EBExbiGYc^0KSi&Ct!D91++Q@R$-@@zzTHX@hVh-_rN5ntWWvTtuH znVK>>n^O2liTR(CE**KvHv?3v`6|zzlw$6RZ!N`W^bTdKdD&`ytmS^$YF@UQmsak| zR`aseylgek&1#NoBfRdm*21NHEJ{%?{*2AQ{%1?zd?_&=e+Jjv9F?#2@f(Z(#||U- zQ8W42Fn-7&|6 zwg*ohqGU+18alEVm+0>gBZYx&%`pN;9NKY|?zYBK#SpEp=3(~7NL(lFd|_{JC=V1A zPh=Nq56zaG`}0V;K3eD1SG-N4>?G zNUJh0?Wi5vJZeV*jb@ojEw079JZB78qxcN%Ab3XWN3wn|0C>qzI$AOCq~3KkSBWO) zWRvwYGLFxLjB>9w%XX9+;Yzz5;dUkL{rHpm@xOfp z_mMTO|Du1bBRBCV+~g5PFv{~;dyxQhpiKh6^? zig6U$;NK{@cCDu$UQBBye>)RsW2qs0eced^8F15`_a^to8?)9^m-E7;&?QR&J0=#dj0Wfa{60DNdAcJIhL?i#7#oZB7}M@}}&=vxLQjV6NUdvB`lJe6+OKIp`+Hx(w*E{<4?(4x_;V3Q1y*E26tyU}T zjwCxPX-41Eu2!@2=I70OZ{EC_Iq=K5uiyUQf;WgJoJA&+Lx*RQX{u+F046Ixjv=Ir z{XTT~@L_y36`&j7un1)nx=92+2~tNoNEog$GNvVkyy19gbtSEj8pQ27*#GpR*Iumo zB;2oPW6uC%glr|N0s9c#b&wsL_$SpM?vF6KGxN@ygpw}}y~I!AWQ25*AY2n<1AI%6 zC*^tEAgzUf{{)hzc{BL2v&~5150WU<0^*FbJV;O%0r>U;Y$FZCLC(yX))|2R47koD z=F5Lb74KAl58?MT_WsA&^-vCaQ9nPP#a+m*N| z080Uu0W2rT+Xy)uUV{705<~c0r;Bg9R^d&vKLh~WRF=Kr50h4FtkZT}Ep@&nc$xF+Beg!^H`vSX+N{52%a%gx}w7+@B= zKNo61&OsdRlB_oTQ2r749)|qbp&tB@CqM9DDT_LtfFEVcVh;STU|e>89xdx2UpW&l6H06+j> z5I_PL0vHC^0uTh)3NQi?0vH7d14IC#MD9y50Hn$UwkyCkfb9U6sBm5i_d5YT&7NHb z*UwnEjzyT5AoRcL*8eI+{x9IbUup7R855huua^IAR@cK+4^PrObi$np~Cl!Dn?JSGfeGk-ofVILl zDDNPp{ciXk*6nZFfa4z*OS9DVk7L0m6Zp~IRL6lIZPZtq#P2kLpSS)|w^iEL2!8{$ zaODoC7l0pWdKbIbj(>YV>%?A{0II;8nHc|a(;s$yW&G;{ZqNa}*a_brD60)x+yPe~ z++o?h@Ox8=OJw(sB?aLB2Y^Wt9{j!VdoBF$2kDf>|GDvFn8L2c?Vmp&{h^%|?Wtz* zV+ppK#NR|MskuWT{QuVg@0p2cG89t(r3~ea-|HPRF(e`@?_V;lfH z!%XY-F+H>gB>yP<6QNdziNn7ZzDH>tjAiJ=x#Univ_I-oicQ80{x1@F?nV#48vw^b zBR~NWia(KR8#yPL!C650uO(^uhkYzTG2_T1MC(69#{`+0v4`rvvTax`m6x#or`Xoa z;QuVZn-!%0qYY8$|K{KSoReEqQHk>PDC8wVdTNpS&njwj8Kkw{ z133v(P7q>tnDNI?tJn^Ojgxs(UUd!B9{20t9K7@2uj-I!4>_y}(9io*bL4TjU*e&U z&Z7-C0(Yx{nMEm44P^c5f$LBT-<-vZr>%LU_S`*hAN=!Euit;SF1}1`Np{~4Iz$vc zUb2n{e;-w`lC*1WV+3BqurhieQ}E`Unx3(ELK?MuBB78JNCYF{xP7G*mSVwx-5aD2 z{IRjkjTc-Do@ziG_j#^>uwxEDq6@JM9kq)Yo~Ep+qt!Ee}h{M9d$m zuj@(d`+WG7md~e$+KZ(B9yZQOWJuW|s)Hn{ z%|Ey9iJ_kHe`Oxq41V+-`_c^MegL@5TB?6*Id318YWV~7|C(b-dfwd(e)M6R#O{Ut zSJ;26&HjsHA7THMLnYXM(DzN)e+skzD7walXq1@F^oSXr$Y|6blVUSa+vo{iB9X~t z#tr$DB6jV#IOz`s6JtsOn;XVeL%NVw<4i@EO~D&0W)m|vTfZ^dp_$RuLe`1@{^907 zz1wNqd(mf~+j_4(gx>0UI&mk}nvh;L^60-^)X0RdA&H^OMSqw1_w6ZX1ysM|&_U5n zRe7eAVA20fes$}@DNp}x=q1;)zq{rp-4Y<3qQ^#rPk&cCXedLqQ7X8UF~ybAwNL!9 z=DLnU7j>_A*R#4q7rQ#s9OcJfinz=)#)O!|2GJ4hAQbRPCiJ$dElub({c*uzKPvwGd`-MV>*Q%)=F zf7NaOOHm%8>SBZ-Q_<~@)*RdL_JL2`@Xk$l+axzy__C?UT?qYV_SyIT@*fw!_*V4( zSN8q(@>6#2`fpu4x%9t&KK_^4dsfp%PpGT`ncVpM8-M=QTOaHF@N>U7YvldKr`PM| ztT_FzuQ2|bS&yqIH;<|~z`WQq11oPmYRQk+eHi;-f#;jLImaOEf7NaOD~0`s7eD&W zx=UyOq4U4!16WA;OJM)uu74}pe~ibUj=X81{Q=jBoW_A4ZAyw)$NA9mA&tlHdf){4 zXigoF|KOTcWj%@+{3tz@@t1M=jpv_m{p(T={Pz;szaB#$;OhcR-O%s5K>T>f8u(rb zPkJcyfCyMmTTDNkQ2_m|AUx$S5C0z<2mTWK|J0PT|BvYZ)!qN~*B^A#v50mxh+>;k z&kHOo0RPtkUSa}B*83ENTyAq4JZ%Qq=Yi2zJH6%(iweO1eSi^$-+26IJpay%|4D$$ zh*|u`{SWQg=%ry6|Bs2hz7T`f{&^n79UxnXcBPAcs4JlSw-C9%>fzV7zvd3ee{pp* z+R4r0znRE;N$BCnI!C*QSq_Bt{|UHG=fICPF{S_Iw!cTH{%K@`QRMFo4*XvNSjXsJ zz5Wcxq$ta#vGQP=IJb+wb!fdr9UJF4(bkSOAGEpPT36IPQ3u5~#kE*C7R2@VxE>ho z(rD?!b-*}I#JLKbufzF@IsoL|xCaT!59FhtAo7}HwCN*V7rXxl?)jnY|AA#x@Bg8c zTf6@UKFhWL$6D2Uxc?EZea5{$a8H9?_Fo^n;{G4F)&ST4;yxg_{|B0a3tj&^%Fm#r~$Oxi*9U zK_b_`8T?Hq^k1a0vpMkp6G_|ujPV=$-yz-=9Qgl@r04(eook7?b>KK}hU_SDYN#{)ob$a>2lQLS^}pypi0gXMk59e+7uPA{`d?h% zqOAYLy216pX!FK(pSboEdF5+F-UkHJo#-{c>LB#L>el}{*7|x;HY@W{>2=QAnMt}{ zGY_#G#_vvu@1TAh=l^n6ZbT-+!YmK?IOqj^kUDx{*y%ThNVv-%81WBF>G{|OoQ_3Z zp^xww(S1;>{hHYnwEyq0PujW@te`RUf|h_zhso5GhpZ2EbLHF2yEAHuZH-{8qYqJgATHP4B4li}{H%Yx*BvntVL@{fjj< zs+I@L9@He$8t_giSr|%I2UxqPmy{e(Cmn(qSgX_M8y!G38|WTwvKG1&sHBC`igMQo zca4;aWE-lyQ%jUhn?$Lulo?Rl#eO(ITTs0Z7WOLCYa`TF+po&8w#Q=`-DTz03ZHNVHOJeQ#Mq3Eam2-~s0AUcQy5YlXoTyF^Ugm z#5y5Ij&e^i<6HA}scRL*HJC-eSU=_{y;ha3em(2CgJ<`y6La>M7=1QISFQPJFIC?2 zc66@BcQ>bKGF?8TsydaQR~_guJAg~MV659uJ#U9-Nt9jS>NM1)^rf}bLnB7#=pyhA zN``Ez%}3GwDGKg~=W$34pUWIe;bq7*hv^LI04;xwieClFrgZ4dDDSK{3Qs(2mf1`h zFDlbA<}Vr2=UR`avvNwQey-Lsl+tEGj|cUh+WNl+p7ifuwQu`%!%zRjf8dL=WxdBv zc=R5%W~ToUC6i zO#*K0U3GM&hMY?e?fIj$&$ZCDMGmP~Yx;GTE^l@o>=}7b=fF-iZa+;YK{Y~UPdhvH zFvF(`x5}<%p@&RFZ&p9Ypq9UaO3hO#&aO=yRy{4CQg|lnZ~8H`8f;~7fI@Wi_ouxz z?eZ=E>AB(4ho7B%|JKJ}sF!v0Gq^Z~V)e~_mZ_WnaPZIDuYNgOH%DE%o9dAgwLM5V z^>fr{qfq7n+P6_FLyKi0nz?MtEmSik=)VmVt z%^qOvIl8ekZBV(#Dq368K$-S`*~8C_j$`BM(`o6SpSmYjpAVI3D`l>=wbNAyF*+hM z({h{3rqqRuF^9{hboxE7leX=}!luj0xTS2$sHbOp=pGFylcN>gqKdpXQ^ujPEhEO= z&OATNlr&kkB~@Bexs$z%l>HCFDecUh#3}#LkQsZ(cxyyPHj&EXlsG?N6jyKbU9McT zIGaN?ahs22)eKK6l|(aI$(v0xN{s5WQ&pa}akNhBs^jU^K;@2dkISl!-0$esOl4SU zW@puc{J&m_D{XXqOpdarV;b46%h#I&J@|~hIfT7f*o&=TFRolEBy6qLFJ1ArR^hv3 zjWMyX6_C7ItD`Cl+>MVJ8-LVqqs94?D5q<2>~b z4U320ys>@nh3Bo<{nWgfLpIsR*-3abI>j!VYc*o#znOl|FYkJ@V*H+QXE7eX=Tfar zJcYfOMZOrxN!h6LE9b3Cf+!n#`nO zjB~u{x-ZqL%8B6SY!XT?@3Hi`wtnsjUuedM>(WFaC@1(Te8*a6?=l z#AsI8?2RIDE^ewBOq z$m>d(e-isYi~XO){?FCB|ML~>JvY zyG5M;pagOLgE;@8`p$n4-fgAvZYv)xC%>lwYB~BbjJ(;(ch&RBrg(d0dW(tkU&Q$@ z*7s5o=f8;aU&Q$@R+OFM{1s2>D4}>uaxt@tiWM; z=3^#y&aCuxZnag$zVQ9g7l&i}G1XSeG95tT^y!vCOp=iQI=8ZE(h z#J)9FUPD%{r53YWVs=Z+Zk6}!me~K@$kJd{X)Q(xu@#}#WC`E6%v9w_$=efapt*sgv{jRIN3SWj@#1QthPiw*a_oe?8QgA2Y&)! zOHp-Y-v5)()xEzjLG3Ab4P+CPNG?|p@`WXCC%7{Yt`%0W+aR8Uw&5iZD@yx~nRn#+ zZ!i^KDdEut^$tBfKyQyBLUm2P{+xaGtorOpLO#x^0qAp`d^Cn$!rp@Txt2rDFQ{j0 z>p;I|=W)vh<8v$tIW>p&AkKfS?(<*8Y5}qSUwB$qucx(<#PZPJnB%nWiTs;)w99vX zG65YESDx|7YBOV`6FH@;Ic*%p&G`kUYIIsDq1H_vT`pIB=duQ%EaDmGrgI-Bcg>S~ z$l*iyjbpM-+6oEasZmPxilm7!$rz-3ay`)a5T$?Q;EIe=B<6gog~&PCK=Z6^(ZQ^D zP$TW2bdG8F(AW~~BN%c?c6Gg^OQqBdmTjPfqY{0oC+j&pc=`*HMubeJb)kM=2|Vwo z4r6=&ovs=*5W@9c-_-}s*)#vjO`U5lo2oEPTmpK+b=SRkiKpND&_HMXTeoa`Ppw93 z4`8lgRjy4B)6pITmoKh%L2V7|qS`d{-hCr+=nl`r2j(7!(R!fh|>N_gsdpm-S)B@!);$U<6!q&Z=AgJ>bLg3O!V@X+cQamMC7`jK~5*7 zwOKBO*riR;Zm^rI9qQSgk`Zo*IY`vA7iX6|Ke@ftjPfGSq@br-E2M(ZK=Xn#6D^Py z-Z|;XX*PJ<1)oOx%mY^so!4lCSS@sJ!)2-`rLnpIhmX#Dc%UR6Iy2G$SbP*iD}8E% z)G;S5vFE}y@LV^AaZ-cHl@LQ2h) z2RR_!Cfdgvs9mj%vqjD2+8owWw5U;jrZ(6ec62Z$h1K8rMVhF(3mdlP8}>bL+umL7 z>tB89#{I6#Uj7UD{wzI-oC>*2WG~!Q`@vm5y!no%o)0{?b-CW2u9wOPfBDTCuphkD z@&4i$wmTla^D*Dd=Z@buMGt?7i=Om)&vR zzuq$ut}Cr6=fCbgI57XcIdzeJkH6IT#6@#-YfT+^I8r<9O&zp`J19?d)B7fBclaM%s=|fUAM>SsT@#+)P7hVe-;Vc{5UghhlY8N^nrw z*-CX6vskgav{S7#NXgR!bu-K~)Hx*I{vbd|P( zKVa{b;)#CMy*&_*#v)sH2A1@1kHkj$gY++Uu_rLLKRy-?;9VRd@7#H5yUWw!bhfwJ z939Sfo73lU*;-p&?KX$Y+1T3TXm+|6-DYEAG5pJ8pRi@@MAy@ykrW9}Y+%xDD}S=&KVz)uV}v zlTwl=S2s;SuKcl>e=HI62e}y3IOZ%C8IDQucrYAHNZX`v!XF#v$=RNXs59j0oa|}* zC@iQ{SvZmi4h5wcN3L$0C^5)*#1<(O<;oz8SX1Y;2Ul2IYR@5F>-Jcp2`7-6Nd%Nq zB^ldL`_2e|x7~?4l zid2z|sR_NZ&`ZN>!9kUA?$Hw!4Uk^T4Ks}Mj%GldfgML7T#UYGTmW?Mpc~^_UbQqX z3mW&(g%107`Y&$P%Tl7mf<1cT)U~v=(M_Dx-anq!#xyn`okMaPV57v9v9pJR$m?j) z`?eq6QMASw^+6}(ON4K&EFoE&MHl@Hb23VXOyx)kIMMJK&b$}0Gm5tUVsttrUU2;t zp9iFtP^!L^oS3b@iOZ4v)CxHY(rL>OjW5w0DIt_d#-!z8DG3}oRA1MX90&yiYoxK> z$cPkPKHz8?Y94ACYG@d2cKDn8wUg6(lt+dV_P$^|=@0cJl7qnr6*cDt!-J9S_SNBp z6dUpfqJQ5R1ekLwfF&GFDF{oQOq0 zF^$=~li?mIkqCx|<6eKlkJ`Y(Wvk=9U`Sr(kF#=0Lbcc(@r&iEb$XI(cYpLtGj}q%;?2LN=ux>xVl1@kB>R8ig3k zb4Ok_8cTX2qaaZSd!-$T&R{&DMTNvXt)`~kxN16fB&2XWI1mg46Jzj83M3QIC`uWy zS!c+Med^zVw2Ds=n{4_ve<&$!+C*kdR?+%QY>c`1g-_>9p3PYmN0R<2CAM)s& z2-*${f(1U4WBy)u5&XvS+cM(Oqo24%3eMd#b<5zr_DkvF!J1XL^%nfl74u^`rJ@Ub=AZ4bq?9`1;j$LUQ;YC^P9C{ujDSu%4@_lwBYm z{Oi=Zzy0ds8@}ffnNH-HVcheAlKafLQ}q>vOyNLj3ITv%Bt z>fah`I75HE?%K0oJ7e!rTOX}E_Ajrl)z<%z>UX+Ia6%Q)CJdH z*V~?%wAh8#|Djh-yZ5;Ju3Z1Xi#uk&dd2kFyw%U)|7Z4izM;O_gZI6;=u@{$Z~f-; R&BLz{lE(+E*!&;q@c(`>ykr0X diff --git a/Implab/IPromise.cs b/Implab/IPromise.cs --- a/Implab/IPromise.cs +++ b/Implab/IPromise.cs @@ -15,19 +15,6 @@ namespace Implab get; } - /// - /// The current state of the promise. - /// - PromiseState State - { - get; - } - /// - /// Registers handler for the case when the promise is cencelled. If the promise already cancelled the - /// handler will be invoked immediatelly. - /// - /// The handler - void HandleCancelled(Action handler); } } diff --git a/Implab/Parallels/ArrayTraits.cs b/Implab/Parallels/ArrayTraits.cs --- a/Implab/Parallels/ArrayTraits.cs +++ b/Implab/Parallels/ArrayTraits.cs @@ -140,7 +140,7 @@ namespace Implab.Parallels { AsyncPool.InvokeNewThread(() => { for (int i = 0; i < source.Length; i++) { - if(promise.State != PromiseState.Unresolved) + if(promise.IsResolved) break; // stop processing in case of error or cancellation var idx = i; semaphore.WaitOne(); diff --git a/Implab/Parallels/DispatchPool.cs b/Implab/Parallels/DispatchPool.cs --- a/Implab/Parallels/DispatchPool.cs +++ b/Implab/Parallels/DispatchPool.cs @@ -9,10 +9,15 @@ namespace Implab.Parallels { public abstract class DispatchPool : IDisposable { readonly int m_minThreads; readonly int m_maxThreads; - int m_runningThreads = 0; - int m_maxRunningThreads = 0; - int m_suspended = 0; - int m_exitRequired = 0; + + int m_createdThreads = 0; // the current size of the pool + int m_activeThreads = 0; // the count of threads which are active + int m_sleepingThreads = 0; // the count of currently inactive threads + int m_maxRunningThreads = 0; // the meximum reached size of the pool + int m_exitRequired = 0; // the pool is going to shutdown, all unused workers are released + int m_releaseTimeout = 100; // the timeout while the working thread will wait for the new tasks before exit + int m_wakeEvents = 0; // the count of wake events + AutoResetEvent m_hasTasks = new AutoResetEvent(false); protected DispatchPool(int min, int max) { @@ -44,9 +49,15 @@ namespace Implab.Parallels { StartWorker(); } - public int ThreadCount { + public int PoolSize { get { - return m_runningThreads; + return m_createdThreads; + } + } + + public int ActiveThreads { + get { + return m_activeThreads; } } @@ -64,29 +75,120 @@ namespace Implab.Parallels { protected abstract bool TryDequeue(out TUnit unit); - protected virtual bool ExtendPool() { - if (m_suspended > 0) { + #region thread execution traits + int SignalThread() { + var signals = Interlocked.Increment(ref m_wakeEvents); + if(signals == 1) m_hasTasks.Set(); + return signals; + } + + bool FetchSignalOrWait(int timeout) { + var start = Environment.TickCount; + + // означает, что поток владеет блокировкой и при успешном получении сигнала должен + // ее вернуть, чтобы другой ожидающий поток смог + bool hasLock = false; + do { + int signals; + do { + signals = m_wakeEvents; + if (signals == 0) + break; + } while (Interlocked.CompareExchange(ref m_wakeEvents, signals - 1, signals) != signals); + + if (signals >= 1) { + if (signals > 1 && hasLock) + m_hasTasks.Set(); + return true; + } + + if (timeout != -1) + timeout = Math.Max(0, timeout - (Environment.TickCount - start)); + + // если сигналов больше не осталось, то первый поток, который дошел сюда сбросит событие + // и уйдет на пустой цикл, после чего заблокируется + + hasLock = true; + } while (m_hasTasks.WaitOne(timeout)); + + return false; + } + + bool Sleep(int timeout) { + Interlocked.Increment(ref m_sleepingThreads); + if (FetchSignalOrWait(timeout)) { + Interlocked.Decrement(ref m_sleepingThreads); return true; - } else - return StartWorker(); + } else { + Interlocked.Decrement(ref m_sleepingThreads); + return false; + } } + #endregion /// /// Запускает либо новый поток, если раньше не было ни одного потока, либо устанавливает событие пробуждение одного спящего потока /// - protected void WakePool() { - m_hasTasks.Set(); // wake sleeping thread; + protected void GrowPool() { + if (m_exitRequired != 0) + return; + if (m_sleepingThreads > m_wakeEvents) { + //Console.WriteLine("Waking threads (sleeps {0}, pending {1})", m_sleepingThreads, m_wakeEvents); + + // all sleeping threads may gone + SignalThread(); // wake a sleeping thread; - if (AllocateThreadSlot(1)) { - var worker = new Thread(this.Worker); - worker.IsBackground = true; - worker.Start(); + // we can't check whether signal has been processed + // anyway it may take some time for the thread to start + // we will ensure that at least one thread is running + + if (AllocateThreadSlot(1)) { + // if there were no threads in the pool + var worker = new Thread(this.Worker); + worker.IsBackground = true; + worker.Start(); + } + } else { + // if there is no sleeping threads in the pool + StartWorker(); } } - protected virtual void Suspend() { - m_hasTasks.WaitOne(); + private bool Suspend() { + //no tasks left, exit if the thread is no longer needed + bool last; + bool requestExit; + + // if threads have a timeout before releasing + if (m_releaseTimeout > 0) + requestExit = !Sleep(m_releaseTimeout); + else + requestExit = true; + + if (!requestExit) + return true; + + // release unsused thread + if (requestExit && ReleaseThreadSlot(out last)) { + // in case at the moment the last thread was being released + // a new task was added to the queue, we need to try + // to revoke the thread to avoid the situation when the task is left unprocessed + if (last && Sleep(0)) { // Sleep(0) will fetch pending task or will return false + if (AllocateThreadSlot(1)) + return true; // spin again... + else + SignalThread(); // since Sleep(0) has fetched the signal we neet to reschedule it + + } + + return false; + } + + // wait till infinity + Sleep(-1); + + return true; } #region thread slots traits @@ -95,11 +197,11 @@ namespace Implab.Parallels { int current; // use spins to allocate slot for the new thread do { - current = m_runningThreads; + current = m_createdThreads; if (current >= m_maxThreads || m_exitRequired != 0) // no more slots left or the pool has been disposed return false; - } while (current != Interlocked.CompareExchange(ref m_runningThreads, current + 1, current)); + } while (current != Interlocked.CompareExchange(ref m_createdThreads, current + 1, current)); UpdateMaxThreads(current + 1); @@ -107,7 +209,7 @@ namespace Implab.Parallels { } bool AllocateThreadSlot(int desired) { - if (desired - 1 != Interlocked.CompareExchange(ref m_runningThreads, desired, desired - 1)) + if (desired - 1 != Interlocked.CompareExchange(ref m_createdThreads, desired, desired - 1)) return false; UpdateMaxThreads(desired); @@ -120,11 +222,11 @@ namespace Implab.Parallels { int current; // use spins to release slot for the new thread do { - current = m_runningThreads; + current = m_createdThreads; if (current <= m_minThreads && m_exitRequired == 0) // the thread is reserved return false; - } while (current != Interlocked.CompareExchange(ref m_runningThreads, current - 1, current)); + } while (current != Interlocked.CompareExchange(ref m_createdThreads, current - 1, current)); last = (current == 1); @@ -136,7 +238,7 @@ namespace Implab.Parallels { /// /// true - no more threads left bool ReleaseThreadSlotAnyway() { - var left = Interlocked.Decrement(ref m_runningThreads); + var left = Interlocked.Decrement(ref m_createdThreads); return left == 0; } @@ -164,54 +266,41 @@ namespace Implab.Parallels { } } - bool FetchTask(out TUnit unit) { + protected abstract void InvokeUnit(TUnit unit); + + void Worker() { + TUnit unit; + //Console.WriteLine("{0}: Active", Thread.CurrentThread.ManagedThreadId); + Interlocked.Increment(ref m_activeThreads); do { // exit if requested if (m_exitRequired != 0) { // release the thread slot + Interlocked.Decrement(ref m_activeThreads); if (ReleaseThreadSlotAnyway()) // it was the last worker m_hasTasks.Dispose(); else - m_hasTasks.Set(); // wake next worker + SignalThread(); // wake next worker unit = default(TUnit); - return false; + break; } // fetch task if (TryDequeue(out unit)) { - ExtendPool(); - return true; + InvokeUnit(unit); + continue; } - //no tasks left, exit if the thread is no longer needed - bool last; - if (ReleaseThreadSlot(out last)) { - if (last && m_hasTasks.WaitOne(0)) { - if (AllocateThreadSlot(1)) - continue; // spin again... - else - // we failed to reallocate slot for this thread - // therefore we need to release the event - m_hasTasks.Set(); - } - - return false; - } + Interlocked.Decrement(ref m_activeThreads); // entering suspend state - Interlocked.Increment(ref m_suspended); // keep this thread and wait - Suspend(); - Interlocked.Decrement(ref m_suspended); + if (!Suspend()) + break; + //Console.WriteLine("{0}: Awake", Thread.CurrentThread.ManagedThreadId); + Interlocked.Increment(ref m_activeThreads); } while (true); - } - - protected abstract void InvokeUnit(TUnit unit); - - void Worker() { - TUnit unit; - while (FetchTask(out unit)) - InvokeUnit(unit); + //Console.WriteLine("{0}: Exited", Thread.CurrentThread.ManagedThreadId); } protected virtual void Dispose(bool disposing) { @@ -221,7 +310,10 @@ namespace Implab.Parallels { return; // wake sleeping threads - m_hasTasks.Set(); + if (m_createdThreads > 0) + SignalThread(); + else + m_hasTasks.Dispose(); GC.SuppressFinalize(this); } } diff --git a/Implab/Parallels/MTQueue.cs b/Implab/Parallels/MTQueue.cs --- a/Implab/Parallels/MTQueue.cs +++ b/Implab/Parallels/MTQueue.cs @@ -42,12 +42,13 @@ namespace Implab.Parallels { next = first.next; if (next == null) { // this is the last element, - // then try to update tail + // then try to update the tail if (first != Interlocked.CompareExchange(ref m_last, null, first)) { - // this is inconsistent situation which means that the queue is empty + // this is a ace condition if (m_last == null) + // the queue is empty return false; - // tail has been changed, that means that we need to restart + // tail has been changed, than we need to restart continue; } diff --git a/Implab/Parallels/WorkerPool.cs b/Implab/Parallels/WorkerPool.cs --- a/Implab/Parallels/WorkerPool.cs +++ b/Implab/Parallels/WorkerPool.cs @@ -57,16 +57,8 @@ namespace Implab.Parallels { var len = Interlocked.Increment(ref m_queueLength); m_queue.Enqueue(unit); - if(!ExtendPool()) - WakePool(); - } - - protected override bool ExtendPool() { - if (m_queueLength <= m_threshold*ThreadCount) - // in this case we are in active thread and it request for additional workers - // satisfy it only when queue is longer than threshold - return false; - return base.ExtendPool(); + if (len > m_threshold*ActiveThreads) + GrowPool(); } protected override bool TryDequeue(out Action unit) { @@ -81,9 +73,5 @@ namespace Implab.Parallels { 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 @@ -3,6 +3,7 @@ using System.Collections.Generic; using System.Reflection; using System.Diagnostics; using System.Threading; +using Implab.Parallels; namespace Implab { @@ -48,24 +49,53 @@ namespace Implab { /// public class Promise : IPromise { - struct ResultHandlerInfo { + struct HandlerDescriptor { public ResultHandler resultHandler; public ErrorHandler errorHandler; + public Action cancellHandler; + + public void Resolve(T result) { + if (resultHandler != null) + try { + resultHandler(result); + } catch (Exception e) { + Reject(e); + } + } + + public void Reject(Exception err) { + if (errorHandler != null) + try { + errorHandler(err); + } catch { + } + } + + public void Cancel() { + if (cancellHandler != null) + try { + cancellHandler(); + } catch { + } + } } + const int UnresolvedSate = 0; + const int TransitionalState = 1; + const int ResolvedState = 2; + const int RejectedState = 3; + const int CancelledState = 4; + readonly IPromise m_parent; - - LinkedList m_resultHandlers = new LinkedList(); - LinkedList m_cancelHandlers = new LinkedList(); + readonly bool m_cancellable; - readonly object m_lock = new Object(); - readonly bool m_cancellable; int m_childrenCount = 0; - - PromiseState m_state; + int m_state; T m_result; Exception m_error; + readonly MTQueue m_handlers = new MTQueue(); + public Promise() { m_cancellable = true; } @@ -73,8 +103,6 @@ namespace Implab { public Promise(IPromise parent, bool cancellable) { m_cancellable = cancellable; m_parent = parent; - if (parent != null) - parent.HandleCancelled(InternalCancel); } void InternalCancel() { @@ -82,22 +110,39 @@ namespace Implab { Cancel(false); } + bool BeginTransit() { + return UnresolvedSate == Interlocked.CompareExchange(ref m_state, TransitionalState, UnresolvedSate); + } + + void CompleteTransit(int state) { + if (TransitionalState != Interlocked.CompareExchange(ref m_state, state, TransitionalState)) + throw new InvalidOperationException("Can't complete transition when the object isn't in the transitional state"); + } + + public bool IsResolved { + get { + return m_state > 1; + } + } + + public bool IsCancelled { + get { + return m_state == CancelledState; + } + } + /// /// Выполняет обещание, сообщая об успешном выполнении. /// /// Результат выполнения. /// Данное обещание уже выполнено public void Resolve(T result) { - lock (m_lock) { - if (m_state == PromiseState.Cancelled) - return; - if (m_state != PromiseState.Unresolved) - throw new InvalidOperationException("The promise is already resolved"); + if (BeginTransit()) { m_result = result; - m_state = PromiseState.Resolved; - } - - OnStateChanged(); + CompleteTransit(ResolvedState); + OnStateChanged(); + } else if (m_state != CancelledState) + throw new InvalidOperationException("The promise is already resolved"); } /// @@ -111,16 +156,12 @@ namespace Implab { /// Исключение возникшее при выполнении операции /// Данное обещание уже выполнено public void Reject(Exception error) { - lock (m_lock) { - if (m_state == PromiseState.Cancelled || m_state == PromiseState.Rejected) - return; - if (m_state != PromiseState.Unresolved) - throw new InvalidOperationException("The promise is already resolved"); + if (BeginTransit()) { m_error = error; - m_state = PromiseState.Rejected; - } - - OnStateChanged(); + CompleteTransit(RejectedState); + OnStateChanged(); + } else if (m_state == ResolvedState) + throw new InvalidOperationException("The promise is already resolved"); } /// @@ -144,27 +185,27 @@ namespace Implab { var medium = new Promise(this, true); - var handlerInfo = new ResultHandlerInfo(); - + ResultHandler resultHandler; if (success != null) - handlerInfo.resultHandler = x => { + resultHandler = x => { success(x); medium.Resolve(x); }; else - handlerInfo.resultHandler = medium.Resolve; + resultHandler = medium.Resolve; + ErrorHandler errorHandler; if (error != null) - handlerInfo.errorHandler = x => { + errorHandler = x => { try { error(x); } catch { } medium.Reject(x); }; else - handlerInfo.errorHandler = medium.Reject; + errorHandler = medium.Reject; - AddHandler(handlerInfo); + AddHandler(resultHandler, errorHandler, medium.InternalCancel); return medium; } @@ -182,27 +223,28 @@ namespace Implab { var medium = new Promise(this, true); - var handlerInfo = new ResultHandlerInfo(); + ResultHandler resultHandler; + ErrorHandler errorHandler; if (success != null) - handlerInfo.resultHandler = x => { + resultHandler = x => { success(x); medium.Resolve(x); }; else - handlerInfo.resultHandler = medium.Resolve; + resultHandler = medium.Resolve; if (error != null) - handlerInfo.errorHandler = x => { + errorHandler = x => { try { medium.Resolve(error(x)); } catch { } medium.Reject(x); }; else - handlerInfo.errorHandler = medium.Reject; + errorHandler = medium.Reject; - AddHandler(handlerInfo); + AddHandler(resultHandler, errorHandler, medium.InternalCancel); return medium; } @@ -214,19 +256,17 @@ namespace Implab { var medium = new Promise(this, true); - var handlerInfo = new ResultHandlerInfo(); - + ResultHandler resultHandler; + if (success != null) - handlerInfo.resultHandler = x => { + resultHandler = x => { success(x); medium.Resolve(x); }; else - handlerInfo.resultHandler = medium.Resolve; + resultHandler = medium.Resolve; - handlerInfo.errorHandler = medium.Reject; - - AddHandler(handlerInfo); + AddHandler(resultHandler, medium.Reject, medium.InternalCancel); return medium; } @@ -249,15 +289,17 @@ namespace Implab { var medium = new Promise(this, true); - AddHandler(new ResultHandlerInfo { - errorHandler = e => { + AddHandler( + null, + e => { try { medium.Resolve(handler(e)); } catch (Exception e2) { medium.Reject(e2); } - } - }); + }, + medium.InternalCancel + ); return medium; } @@ -268,8 +310,8 @@ namespace Implab { var medium = new Promise(); - AddHandler(new ResultHandlerInfo { - resultHandler = x => { + AddHandler( + x => { // to avoid handler being called multiple times we handle exception by ourselfs try { handler(); @@ -278,13 +320,16 @@ namespace Implab { medium.Reject(e); } }, - errorHandler = x => { + + e => { try { handler(); } catch { } - medium.Reject(x); - } - }); + medium.Reject(e); + }, + + medium.InternalCancel + ); return medium; } @@ -304,17 +349,22 @@ namespace Implab { // создаем прицепленное обещание var chained = new Promise(); - AddHandler(new ResultHandlerInfo() { - resultHandler = result => chained.Resolve(mapper(result)), - errorHandler = delegate(Exception e) { - if (error != null) - try { - error(e); - } catch { } - // в случае ошибки нужно передать исключение дальше по цепочке - chained.Reject(e); - } - }); + ResultHandler resultHandler = result => chained.Resolve(mapper(result)); + ErrorHandler errorHandler = delegate(Exception e) { + if (error != null) + try { + error(e); + } catch { } + // в случае ошибки нужно передать исключение дальше по цепочке + chained.Reject(e); + }; + + + AddHandler( + resultHandler, + errorHandler, + chained.InternalCancel + ); return chained; } @@ -341,27 +391,32 @@ namespace Implab { // передать через него результаты работы. var medium = new Promise(this, true); - AddHandler(new ResultHandlerInfo { - resultHandler = delegate(T result) { - if (medium.State == PromiseState.Cancelled) - return; + ResultHandler resultHandler = delegate(T result) { + if (medium.IsCancelled) + return; - var promise = chained(result); + var promise = chained(result); - // notify chained operation that it's not needed - medium.Cancelled(() => promise.Cancel()); - promise.Then( - x => medium.Resolve(x), - e => medium.Reject(e) - ); - }, - errorHandler = delegate(Exception e) { - if (error != null) - error(e); - // в случае ошибки нужно передать исключение дальше по цепочке - medium.Reject(e); - } - }); + // notify chained operation that it's not needed + medium.Cancelled(() => promise.Cancel()); + promise.Then( + x => medium.Resolve(x), + e => medium.Reject(e) + ); + }; + + ErrorHandler errorHandler = delegate(Exception e) { + if (error != null) + error(e); + // в случае ошибки нужно передать исключение дальше по цепочке + medium.Reject(e); + }; + + AddHandler( + resultHandler, + errorHandler, + medium.InternalCancel + ); return medium; } @@ -371,19 +426,19 @@ namespace Implab { } public Promise Cancelled(Action handler) { - if (handler == null) - return this; - lock (m_lock) { - if (m_state == PromiseState.Unresolved) - m_cancelHandlers.AddLast(handler); - else if (m_state == PromiseState.Cancelled) - handler(); - } + AddHandler(null, null, handler); return this; } - public void HandleCancelled(Action handler) { - Cancelled(handler); + public Promise Finally(Action handler) { + if (handler == null) + throw new ArgumentNullException("handler"); + AddHandler( + x => handler(), + e => handler(), + handler + ); + return this; } /// @@ -415,15 +470,15 @@ namespace Implab { if (!evt.WaitOne(timeout, true)) throw new TimeoutException(); - switch (State) { - case PromiseState.Resolved: + switch (m_state) { + case ResolvedState: return m_result; - case PromiseState.Cancelled: + case CancelledState: throw new OperationCanceledException(); - case PromiseState.Rejected: + case RejectedState: throw new TargetInvocationException(m_error); default: - throw new ApplicationException(String.Format("Invalid promise state {0}", State)); + throw new ApplicationException(String.Format("Invalid promise state {0}", m_state)); } } @@ -431,40 +486,45 @@ namespace Implab { return Join(Timeout.Infinite); } - void AddHandler(ResultHandlerInfo handler) { - bool invokeRequired = false; + void AddHandler(ResultHandler success, ErrorHandler error, Action cancel) { + Interlocked.Increment(ref m_childrenCount); + + HandlerDescriptor handler = new HandlerDescriptor { + resultHandler = success, + errorHandler = error, + cancellHandler = cancel + }; - lock (m_lock) { - m_childrenCount++; - if (m_state == PromiseState.Unresolved) { - m_resultHandlers.AddLast(handler); - } else - invokeRequired = true; + bool queued; + + if (!IsResolved) { + m_handlers.Enqueue(handler); + queued = true; + } else { + // the promise is in resolved state, just invoke the handled with minimum overhead + queued = false; + InvokeHandler(handler); } - // обработчики не должны блокировать сам объект - if (invokeRequired) + if (queued && IsResolved && m_handlers.TryDequeue(out handler)) + // if the promise have been resolved while we was adding handler to the queue + // we can't guarantee that someone is still processing it + // therefore we will fetch a handler from the queue and execute it + // note that fetched handler may be not the one we have added InvokeHandler(handler); + } - void InvokeHandler(ResultHandlerInfo handler) { + void InvokeHandler(HandlerDescriptor handler) { switch (m_state) { - case PromiseState.Resolved: - try { - if (handler.resultHandler != null) - handler.resultHandler(m_result); - } catch (Exception e) { - try { - if (handler.errorHandler != null) - handler.errorHandler(e); - } catch { } - } + case ResolvedState: + handler.Resolve(m_result); break; - case PromiseState.Rejected: - try { - if (handler.errorHandler != null) - handler.errorHandler(m_error); - } catch { } + case RejectedState: + handler.Reject(m_error); + break; + case CancelledState: + handler.Cancel(); break; default: // do nothing @@ -473,76 +533,31 @@ namespace Implab { } protected virtual void OnStateChanged() { - switch (m_state) { - case PromiseState.Resolved: - foreach (var resultHandlerInfo in m_resultHandlers) - try { - if (resultHandlerInfo.resultHandler != null) - resultHandlerInfo.resultHandler(m_result); - } catch (Exception e) { - try { - if (resultHandlerInfo.errorHandler != null) - resultHandlerInfo.errorHandler(e); - } catch { } - } - break; - case PromiseState.Cancelled: - foreach (var cancelHandler in m_cancelHandlers) - cancelHandler(); - break; - case PromiseState.Rejected: - foreach (var resultHandlerInfo in m_resultHandlers) - try { - if (resultHandlerInfo.errorHandler != null) - resultHandlerInfo.errorHandler(m_error); - } catch { } - break; - default: - throw new InvalidOperationException(String.Format("Promise entered an invalid state {0}", m_state)); - } - - m_resultHandlers = null; - m_cancelHandlers = null; + HandlerDescriptor handler; + while (m_handlers.TryDequeue(out handler)) + InvokeHandler(handler); } public bool IsExclusive { get { - lock (m_lock) { - return m_childrenCount <= 1; - } - } - } - - public PromiseState State { - get { - lock (m_lock) { - return m_state; - } + return m_childrenCount <= 1; } } protected bool Cancel(bool dependencies) { - bool result; - - lock (m_lock) { - if (m_state == PromiseState.Unresolved) { - m_state = PromiseState.Cancelled; - result = true; - } else { - result = false; - } - } - - if (result) + if (BeginTransit()) { + CompleteTransit(CancelledState); OnStateChanged(); - if (dependencies && m_parent != null && m_parent.IsExclusive) { - m_parent.Cancel(); + if (dependencies && m_parent != null && m_parent.IsExclusive) + m_parent.Cancel(); + + return true; + } else { + return false; } - - return result; } }