From a125de585e1f6b5dee114acb12c788f47bffa51a Mon Sep 17 00:00:00 2001 From: Guillaume Martres Date: Tue, 19 Feb 2019 20:44:23 +0100 Subject: [PATCH] Add lockfree assignment --- .gitignore | 16 + .gitlab-ci.yml | 36 ++ .vscode/settings.json | 8 + assignment.sbt | 4 + build.sbt | 11 + grading-tests.jar | Bin 0 -> 16090 bytes project/MOOCSettings.scala | 46 +++ project/StudentTasks.scala | 318 ++++++++++++++++++ project/build.properties | 1 + project/buildSettings.sbt | 5 + project/plugins.sbt | 2 + .../scala/lockfree/AbstractSortedList.scala | 32 ++ src/main/scala/lockfree/AtomicVariable.scala | 25 ++ src/main/scala/lockfree/Node.scala | 22 ++ src/main/scala/lockfree/SortedList.scala | 26 ++ .../scala/lockfree/SortingBenchmarks.scala | 83 +++++ .../instrumentation/MockedMonitor.scala | 72 ++++ .../lockfree/instrumentation/Monitor.scala | 23 ++ .../SchedulableSortedList.scala | 35 ++ .../lockfree/instrumentation/Scheduler.scala | 304 +++++++++++++++++ .../lockfree/instrumentation/Stats.scala | 23 ++ .../lockfree/instrumentation/TestHelper.scala | 124 +++++++ .../lockfree/instrumentation/TestUtils.scala | 19 ++ src/test/scala/lockfree/SortedListSuite.scala | 96 ++++++ 24 files changed, 1331 insertions(+) create mode 100644 .gitignore create mode 100644 .gitlab-ci.yml create mode 100644 .vscode/settings.json create mode 100644 assignment.sbt create mode 100644 build.sbt create mode 100644 grading-tests.jar create mode 100644 project/MOOCSettings.scala create mode 100644 project/StudentTasks.scala create mode 100644 project/build.properties create mode 100644 project/buildSettings.sbt create mode 100644 project/plugins.sbt create mode 100644 src/main/scala/lockfree/AbstractSortedList.scala create mode 100644 src/main/scala/lockfree/AtomicVariable.scala create mode 100644 src/main/scala/lockfree/Node.scala create mode 100644 src/main/scala/lockfree/SortedList.scala create mode 100644 src/main/scala/lockfree/SortingBenchmarks.scala create mode 100644 src/main/scala/lockfree/instrumentation/MockedMonitor.scala create mode 100644 src/main/scala/lockfree/instrumentation/Monitor.scala create mode 100644 src/main/scala/lockfree/instrumentation/SchedulableSortedList.scala create mode 100644 src/main/scala/lockfree/instrumentation/Scheduler.scala create mode 100644 src/main/scala/lockfree/instrumentation/Stats.scala create mode 100644 src/main/scala/lockfree/instrumentation/TestHelper.scala create mode 100644 src/main/scala/lockfree/instrumentation/TestUtils.scala create mode 100644 src/test/scala/lockfree/SortedListSuite.scala diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..349d2e8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ +# General +*.DS_Store +*.swp +*~ + +# Dotty +*.class +*.tasty +*.hasTasty + +# sbt +target/ + +# Dotty IDE +/.dotty-ide-artifact +/.dotty-ide.json diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml new file mode 100644 index 0000000..584562a --- /dev/null +++ b/.gitlab-ci.yml @@ -0,0 +1,36 @@ +# DO NOT EDIT THIS FILE + +stages: + - build + - grade + +compile: + stage: build + image: lampepfl/moocs:dotty-2020-02-12 + except: + - tags + tags: + - cs206 + script: + - sbt packageSubmission + artifacts: + expire_in: 1 day + paths: + - submission.jar + +grade: + stage: grade + except: + - tags + tags: + - cs206 + image: + name: smarter3/moocs:concpar-lockfree-2020-03-24 + entrypoint: [""] + allow_failure: true + before_script: + - mkdir -p /shared/submission/ + - cp submission.jar /shared/submission/submission.jar + script: + - cd /grader + - /grader/grade | /grader/feedback-printer diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..a35362b --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,8 @@ +{ + "dotty": { + "trace": { + "remoteTracingUrl": "wss://lamppc36.epfl.ch/dotty-remote-tracer/upload/lsp.log", + "server": { "format": "JSON", "verbosity": "verbose" } + } + } +} diff --git a/assignment.sbt b/assignment.sbt new file mode 100644 index 0000000..c6705a8 --- /dev/null +++ b/assignment.sbt @@ -0,0 +1,4 @@ +// Student tasks (i.e. submit, packageSubmission) +enablePlugins(StudentTasks) + + diff --git a/build.sbt b/build.sbt new file mode 100644 index 0000000..088b0a9 --- /dev/null +++ b/build.sbt @@ -0,0 +1,11 @@ +course := "concpar" +assignment := "lockfree" + +scalaVersion := "0.23.0-bin-20200211-5b006fb-NIGHTLY" + +scalacOptions ++= Seq("-language:implicitConversions", "-deprecation") + +libraryDependencies += "com.novocode" % "junit-interface" % "0.11" % Test + +testOptions in Test += Tests.Argument(TestFrameworks.JUnit, "-a", "-v", "-s") +testSuite := "lockfree.SortedListSuite" diff --git a/grading-tests.jar b/grading-tests.jar new file mode 100644 index 0000000000000000000000000000000000000000..bb247679a492d0912cf52a780763d4dd6916ed0b GIT binary patch literal 16090 zcmbWeV~{RU(=FJxZQHhO+qR9?Sa*Lrj8g6iMOG}xwvC!AWzQk@FET+@1(cEqW7}Ui@DTi4v zDC=q8+i$^-$3krX6hpx+vX0(0l;OqA+%}!xQ=w{+j*w=a1@_s^XexX_O0tkpu#)2} z$^Q`O0R#Q>LBU{vfS`c>AGo0ZMc6r-+FH1nn=}3=?0>U`!+H zGXc9K)39Xrr)16bH$5*uKOcTSKVL6@1=(kJ8OtNXrV-e_V5O1lJYJ$NF-Yto4|aZi z;KQOc$7Ka%TdrLu(Jos}v}VovgrINGXLZ}*7oQRt*U#3rSYX3P&Npeh-&(}LwzNED zEgj9fr@sv>WFc9$SeTQJJGzB;jp3R7l*}fYi`OoG61enQH=WID94$ue3i|DqPqz~p zrUOjaH(dsE%=(Y6XWrhs+U>3m>`OMP3qA_8F=$=-N;gPGmXD6b@2fzgz{mdpe z+v#cuxR+W64+n^E61u^8M$9kaZmTkrRgPX)djdkW)a5GaLZ}%ew;Y{~E_d=BTg~TJ z>DaP=O0~HR9T$2z^dH{c;Kdq<*?C$FK=;p^AmhY;`r(6Tmdq_rtFbt80q6|$jGD{@ zoXwYV6wb+#bG3Nb1n8=$;&=#hqbqB~&-Py*OjZ+n)t9%N3PqQ!nce1*oaZb*rM`*@ z>m5M$n%&xtOP7P21H=)lkuMfZbx)CV67@;0)mT30c*g85QcXd-=-5};Wwyu>&dSV5 z9%et%9EETZ*MUf}SRl{CaP?ka;c<~aWs5w5({}`gT)TF9R4m*C5G)A8wus@t_Bb7m z^|PsYY0#1jXDv^eYSW_ap3|GNJ~gv;3!mR@=YP%xzeCKd@;5R=EDZ`2hVVj*o8~w# z-MyUSor88rB&Z;)MyJUkG2$?|Pl7F_z`l3m#_am$fFyU501mrd7}o@tbqb~!4{9m! zoi?!#Yz=+y7r&vmjFC5l>O0_EpJE~hF$foF30(g6OyeCS=wv1se(^72J(@bM=`dRt)a}No4GtfKbD=?;nfQxf7WMs&3!QI6GOjgBRqraXx7-_nNEsR@e~L5Jhv@zl*aKnV%0d*WVCAp zq3hRx%zcs>Cvk~~w#vvNya?*o>Eu+Z;+?{Ebe{bZs{vn(1S^m@ovdUeAo#1^Vm$^q zhO5=lEjzkgsQGc3{UMP1i;X&iiRhtuCNMI$QiVWAq8ZJF3YA&Vi!?-(53iJlAh&h3 za5^2S4Tt^qg1yqZo8q?>tF1vYTwm$r6ti>{HB1>Ef}JEkbb6yLEXUtMtATcP;U*s8 zeIvp5PEKxZe>!7Y0!$SI1#^Os(TS{5dc(>zM0 z`DQrR0PEbEAN_Q90ksK`m8W5E({eV9^Ik81HE1mOjV>I}+Sa`k8+PpnHds_P5fxf8 zp{_5sakJT3BV{K_lapSA?eS1jl_okG@ON1ysJC&S*~XH4MyIZM?{8#`FzehKTjacu zGq-gVz#$=^5JD!$9~h#IsWKk)aY@USaelG)Z#YDnqObYfWdjv*Uuqzc%LMwKzxWKw ztVo`Ch*yrVORQn+Ttv=v!32pUe6+L(wbpSswS#7u&J zD4(PDlf{;wzxl*U%{mU+v{T{Q`-c^IHJ)aB10~v3zbDF7)JzDNP+FWjqOU&VH!5uJhx`PNtER00`x=tA!#tW zFz-Z6`d$Fk5jVJ?fX&Gr)a0R4%frLYYI%_{b! z_C<_3nKXIkjLo&>WOn(Gml!SR8-2mN^4{=|%fcjd_5~L33gbeK>mg;V=5$Y&sk(@T zB-lvg2W53sfr{nAnKMMpmi+Ht+mLqgoisue5JfJ@-k(dP0_grC=7+GqEyVK;2|>E8KC=^zK# zq5WlLw$>TJf0)nZf-Rr!q4{fYm#dsXDFnQvppG|3kZuHC&-A+ai33Ydph2Z>%QCkO{7d4xz#sCla1#-3u+xaW?$*f7c9+;%qd=>z|2}Y;P&2 z>~2e3eJ?4;>!92X))ov57~9;k^26Zy6-IEApc(_z}pzxC0%Va~U zA6PO&pbx+6nq9!KxW_?mgyAYQoKs}^01chDY!ms09hU=NA8{0M5SNIM8&Fq`y2zZS z)}pLs_oG$IiIxka^15P0?}B3_GtPAj!YVIiPlnYCK_fyvCZ+fsRq_0Zu!4?GMNo~+ zZAFWu6^@-1T&DDrpAgtQ>v)@#|Dz^6({1Nx@$FZA#^WRtO zz=G$t=H#{>(9wb()Q}lWP5~cW09Be3nbV#CUr$b%)(cZQPPv?r>_|>+b%bqO z-z%98ZFC8g?$Pf@;r&HkRTfBb{cMNE5$oKBrr2k|s&NtkUiHh%3Ppg3xypRLM@J)UTLA2IaZC2@SayC6MjkvhQf!gdbj^ ztL}dsZ)G~B-2^>I_krKm!@LOiZU`O%)d^O}DWLeU(ufdxi&KP$qs%vM#V^ai-aLdp zuFC?{gVEv@636G06AH$lmmu5Nfrv_B}U%2S?Lg`c@z_Jqk_ z1*ctne_;NBmB!Bu?EuT<9WJb?yxLlc_(!I|8PWk~40kXZ5;rmce}$jhxZrQ-#SnTOZn4R;w;mSm2#Y(k2xha#;TT4jM>e2$w;@^CUy& zZ-KT&*FORGldMC3nX6iGRVaCqd5V%rnB6Zdi^8=loiS{ZT`RlE=uQ)lB7R-CqUyx| zj7?C|5|yZ-o(!Q*3w2D%XHb@smjPPPfIjOZsMW2KHs4dw|J=TpmkJka4)3_2>M$n- zxj|?p$+^X40u_T*YT6wzS)3E!KM>m_9aZQlWthA*`L2W+Imn=wc#x6h+XyM)2bxAk zoeoEx29Mab9t!LRjY6?=_sCnJlkF35LlYl(?N;nbj8o+WOl2JM1Y~m-JqYsG=COnx zQdHQ1`0CG9J3k90jMCCL_~GYc*cHWCE5(-EKkc2QdBj5-JtZ$l!CEgP$i6@$2=>my zk(b&HIC-e)K*2j@c)kA-J5FdACw+;8d@Nio@m*M6O(Z0}PsYe*QZJVR$;C+Sfr2CW z!>8t{lrxckQCWDA?f_ry=hSJbz*2o3bybI~*1F8tMA_ceT8G$Kz>LK|v(&OmN(Vr* zIN#u?!gq_rjJnkyh6g*`2rO`CuB1^0{be8~9rxvXDuKg07>Ud4Pz4j3kB>BZmaNCm zTxn}j)P}WUBu!i|Ms?$HztOL4#UFSHEugkgu+2w&FrSMY_d;HyZXFH%u3d%8 zS%`3n{TO_)isqrXIoPx)x^}sU5z)cy0H&C{)!g@p=UR}bVroUD;8RD692*YZV+V5= zDF9S6>~WO*-Pa;?3d$tn9eiTA(u6Y~xz#)!Tfs#=8CxMKGR<}JjD_O#r+ot`l#>Z} zI-kh(5et9rZBc(l%I!c|jxZ*e-7`oPZL7fKoe`%X!tftz?a=dv=~zQn2yPA5%6Wyv z9^7aG>qE0@Qg+H^$PrF%2XzvC36jStYZme+YiuE_b(qz?;??p~GF8V!=>Y3Adq0Y! zu2OEXDQiRDAAIa;^ms4o$C{tIk95y@ow?!M=v62aO-19Dju+w9P+7Nv}i8 z=AUNKCx^K!CB}%p< z)c}1eN$bl_^#S`4cHc?%)YD7;DvMG24LLc#qiwaT_2&I8EA4M}Dnp%>-@|Xr!hh;r znHUiMEPHf?td_^Z{p#{e(i>1GOID#Qq$!c&$`+KkbunKc9h{wo1Iu zcB>0qG9!a(aoK{r?lJZ9_oDUTR3p3YV31Xbo1$s=!s`I=fl?6$ z;Pl@aVXusOfSv=n!eAJ0T@m7MRhlElG8x%RqdG!d7~1|%z~Ds zcVG8I%Y0-%irY|LH(x+aM){i4#6Y7TY*U`iX%w*xzAKto7maTKD6Z9hW#&91SVZQY zKFxuf3bDZ5pdq1{m(rR*^|tOr8@v*2lmRbjzV_di0tydQ{aRc`g&lMu$t`MJv0K{C zay>$ioyfeJce7LVfVv0&NpC3^0|`9b%U>BseAco+Pd%!%_qb%IF@K^IOc=c+{bktq zsP4Y#cg9Ch4s`heut>6jwmsYw7@ySZQmEr;fyr-yHf_F3UA+$~La^^qDI?f;3lwmY z-(3TOg$+Np%D;byGM8aWe#OC%+sN`dM{)(1?8p}ZOMW}o%XIDxzME_@k&RH)Pr0I< z87d(6JjwYRV;-^SQVcPir28 zQW^s`pvuJX%))7JQrX#IxiX_a-tvMQjev&zk+m82NBr6@oLwl$7ME9#8 z+S~JbWZN=$DkD2sknmhg7ovegEDsuT87O(g?aLoD)lf5rao0l6cwbj2_g9R6+3>2z zD&oVxNL5h8r`JUqKx0s5QaLfXeVe;m1Ff!{AS$7I)_4gF5mTmhkE$!pvfpV2hMV;I z>3nvffY5nA zC5hYt64`@c{DmsJ@}bY54oPu4@&oBj%j|8~86x_N6$^DRZ5X-|)e~*X1}%gz1|lb@ zSzkyyL9_U3&ClFam=<)EGo4Si=C?iX#KX_RrQHcm^^Ld|f*8qD{YP1`xPR36&kp$Z z0oh%q4&C3(OOTl%wMP_Yy)2g&jt+5Sma-h5G0G!XvN6v=F#jlXTrwkfW8QeyuAlKLS)z{xwGmT zAb9GpFp}>O_-Hha1DQY!0n9kgFsS=}k<(o~?-nAquC~Tsb9Z;vaU+=`uY1m=1wh}M zrZvx-QPR7{b#k}YR4$lc{m+$~YQM&`>a{{pbk~Y8OdIBj@`AcpCTUo;s#@&) z=3r<<6;uZrVVz1_iU@Il03`Jqp{ZZ}iLhjBzhuO%DsnjTvF_*V;+u*E{ZJ}6 z&VsI{&I769t{-Obnx0bHtw+Vh5EspV$iHL|sz6sf3yj#3$Rv5z!9kCB znBX?C^Q{!I3G{nAmA*uM-L|j)bfb-B)Mq#TRK;IAqw10xIbMp|!zZ+{_mk>rXo4$A zcLu7vfYoPHGiw5(!L%psYl*s@g&i4aZPRtF_k=TuJZULfGtTNuNmg^U&LzVuX|JZl zhZ%l<4DS}GC`t_Q#(&@cA+I<%`>nG3hvZ28a~7Y?cd+W- zl;I{F7}oIRof+Az%zs5SXMo@TYh6s<#`D-vI?&G86dWda34@EQ%-z*qltoX|YHK|b zv{{BA5w-|?*dz7~A^0R+i63j)v;fgJI7H722;m$&>B#Cpr_?L1*O_bpQ;!~+p?m6K z7oK}KFrF+G4+uRq!d_2wk1n+qx0>6B_oS}0__rzGDCg$3;G8Zi4X%t6I8ljjlrPkE^$Rv{x5XQ4@-Q6JJzfL_^im?3eOoz+g)aD82R*=RZ)Ke zqZ4&)67Cae(0UGM9pt5Yk_SQX^9bpZo) z>MqWERLqE>6Fl#6M>tAZxO#@|1BuK{<(!HlJI{kc++?2MGiG(Qleq!lg*y8K>8KJ; zy-QOvAvC8*T)&v+h7OZxF8U6dI}HYRuPlIz8Ip&a!%)hGp5j^JCHyy_Ua?dk67muq z84kUEsAsR(1+h|Q9q;f;ZHH<6=mWE%zzv&Qk4h^?5U;^8TxR2d!7A_uD`P|oiwzcQ z3n+PlSqtTwq24}|W^*D9xu-KObG}c^6CT}~w%i9x?qro%({=FF$Kax&`3UhQ!4H!;x@g9KMU1~BrXb^7}& zZTT`-E*z6h4@``**TfFvwr%lj)Cnt_){o!V8QAfK{BvetilusXH0i9ak=GeVhP8n@ zYABOsAH=YCRxg6%kRj3;K3*63e3ZMRDe??Nt^>5x14Hz}%P0eIJDriKIHD^`PdR;z z)C+ z@$)~4Lq2|r!ZBPRpdr@(MI8SBU<6I=j9p!~^&x+uO0ESsn#s1)4c4C12WJ}Dh47L; z#zz6baG{VuhIB%L>9}c9*_L3T&(qYFq+Cjqj0Q^+6?mY}`(Aoldt|%xUU#nIHF|2& z=o#K_u0H*1*?D;%p3{DOA1pjPyx%)cAb|_F(&H9}32Sw=36qbDvyG@od~1H~OUbWb zCgBG{ZAkiRct?Nc58fAFm`hH(BSWqub!TTI!2kCCBnbU=+)>GJ_M)n4$fR^+YZWQM zH~s0dh*wN}ssVBFx!CB4uG~S|scU;12p1|>hw5DOY;`B5fuoS3sgt+4ia_VmPaBXOuWAZ$}gdMK%Q0 z9@Fd)QTL{n=5F!1->RlA-6Bw5o0aG)#G_#;f(uWASmh{nZqQRilSs_2fpeF65M?zc zOGNRh%BT*mv5fcyNMvkoZ!TKAt$7e1Lcl6nr)3d5vbIYVHH+(&Vff&}x(#h&3X(0H z5P$}tL!V69lDH7b;Yd==)zso`C6mh;k>{5Ws!>KU(c_M#fVg(qySTdY2N@L#;7rgv z1K|91AtKye{_z^QF}apPTi(Bd)B$NO%4rqGS-)9TOiZJd3-*R_3?+W zz15-hs{ql3o;+F8j1jIi!GN^IG39OxcI5bH5kdizvK`f8S*he|*cK85$^H_p9H z^1D;!Vd#d^@+3l=8J|jPfW3pLzPY(WC$h1QfqL_vuv;HM%B(INVkx=$2wPRWicBK% zh86KdwzP#5Y5<;~_?K0FWBi(zTFmp3cRVb{5DBH(P!hu#Fh6%_0*RfN#_PnF54j~) zQIUl>)5Xh)L`;lqj3%Tgt+;ge zl?h=uwA@hcGsvIg&c(izm&36j;Q{I{TvZ&0KCxCEnxm+GQ_!jg`GFJFn~ZJ%awlrf zk7=Z}wsJ*)vK02wYIDIys|w5-EG!1*&-jJMScFjMjW*<=UG@7qZ* zOgnNkvaJk|8ke3AsA*$yCW4!~O6~DoiuHW(O(V#mEFak#h^VyHT?xfVI+BkRgmq|9 z_gc!g0&MK5UF1<85}Q3NX;R>Aew1jJLN7GpHz)!9XH^Or(0v90r1k|7<#o_3CqceQ zr$}+Njs;PUD~Zm^II2OQx8PndZQBctiJeMA1Z!_y`Y?~KRImUrf@6g{EN;Nz#vRqr zU@sa`J>0M_StCF-H9%4-jg>#Bt!Z;%eVx5w31wh@utj0%!4)cl92ckB1SEk7+pMqk ziS24i#6(qYaYg+WcJ#3<;nQXTffD|?iAg1*vC5vsB6H_Tz3!Pr`x0P&vBZcYrxikOqIa4!YN&)Z-5q$1H=VXA!%3Qe z6rvE_V=w-x6(Ot0d_LtnrEhAoc=Fx9&g08Qxt~#5m!~!;!X-)Tzw6;J!6vMS%fbE* z`1yu7^lQ35fQID{$RnrRe$@%F?1_zg%dNp$wNg zi;G;EHIWd%e%9M4G6`M$#y}X8PHFWpDu=hjAA1<*r_DVR!wevLAU!Q1m0tzD z&xVrk(61~t3tPtq_G7C!xy1QJVDa!UKt;Os=cGKUjOVzV%lim;wlg?*55^X~tL)QR z$Hq^ZQ^3E)_D`s=ttXvzUP+yx} zrIy!5_jMWD{NR1Ifh8UjyoA6;)XRP?bPU@I(nsg4WvmsIERMaIWHeg-m z^%0{5Mqd!9?R$(1%ri0mt|2YF@j8xtk>0n-tC7li#f@N4N*-sFf6aKJ_N>9#vPoJ7 zw+ysyS=ep!u$y~8*v&B@k{8sv-sm);H~>&*@F zWPaFdo+dO_N~v$+cLZvW(x$f7?B<+Cetp@!+f*+>C&DS$BoRAWi6~##?QZpcTu^ze zW1G?m>x#FL%(}}SNN9*oR~#rN@uRJ?b;|AZ;HZLLq-2*9(uuR6zD&{YVN7;aXp9Xl zPoRCuQFfSA^0UCFU#s78nCZscmeZ1U_;E^02dGN49a9nmVE9e>2O4_kQ0K;TB*y~u zXOPCSx`rEc>_;TjhyxhY4-!i09iSU=RT23g(H?_-UB+tX72v8i*Y&5HEnB}*T_l5V z%vr#f*rB|`ZrT7=6Z}$rn?xNAMap@YH02&ODkxEKeCbOb{rtjW^Vbl)FgNJ{? zD8>GYO?A7~Qrq(9Qa%~j9G&{mxzYZ?&E<)qBo>0m7a``6*tHrq{!}}iK;R1r7gYsv z+0#B5B=G$cdZ~{!#FT)4s&chnXg4Lj$~)s6FIo0Lein+^$yANf7a3_lC(L0$)E+#O zbQzY9INP#SnVNmTI6U2T(BQOh!+vjHimOj9Y)LTWU{j+A73*B-YXWheC)*+pX%Laf z;i+LxFEU?E7AFOna~CexOW?>xV&a9HF6d2%vl!mRF-FAyn{<=uclWQ}Ev_mmYKLJv z#AWG1*pmx;-RM+LqDT2ie6%SuMX4*f!JgF9@wJ?RyL`G{+>Y|z51A~kt?>%rO{8*b zfMJ>>=0{p!flu$Q1B-eTY=b(fH7j_hK8h4y=T5N(zmz(nbmDN0gL(}?QMDxb*-(vx zW=(!!H6{7kR84y&;NH60Mp(Mx7RO9+nCzfnnUkm$nTZKbBuEWdw1m;Rk!j66*oI$ z=v_~`mcIoxXT1FERlt7_3|mmlNLEJwmAT&YAnP?M-B^S3 zJ|5vSwpkg)Oh(kytnQu$WORt}PL|R?-Z0Bay9le{i@V$(ZEB2B{}rcy!u|<=Lpj>~;nz*zt>x`Mg*+2< zRy!y!napfD-{A}6T!oai&dg2+WZ|jOhy>lGZdPy|A#S2>5G~F;u)5_bbLZ>GS0;<` z=0s-j>L(sItNBdoT*wP}nbO&EO?IQ^7rFmhiAH&U@;v_)=+Ur0bw_A&EYbp7Ve_;J z)LOq4Jl3v(dXIoQhMEab9?kgQrioJbn#^tAn)P59zhhmR1Y(IYC*QlA8V+`6`0$`9Xbk|27$LmR(iBTSnMdf`j zHdp70O8+%{1M#Y2#nJc)9Xog(eHMVPVh17_J4Zp4Ch?r)p3T!)yb@+!}n^??e zj3TnfV)w7H`_R|28kI+uP{8}p|9Rb|)mx)umB^kTMV`fAUo+MU*3P~X4dP>>2)qX3 zw`{0BlC72{7cRgcmIa{|d1c+FxmQdnanh8et4U56fI^+%kXI=#^}1|Pb#|1_HxluJ zBgZbbaC|$00P(6tLU`H$q$#@mfOEr`reDBe_y7#Erni687qFIDsJd@LyE@^G2mIm%jc@{Bs|;^;S94DW>*7nr-eLIIz0OzYPuAs` zX$71IjGJl8e8yX`%B(Whbg!S=5b?&u2DF2Lq87s+&*E-E?4qXwAIh;Cv+#` zdhQHycVA4L@s2BV(d=3e`-FENt0#8TpY|-kKQozlLcts2AYQ)=;p=Mi=gewu!rhQp z;7i_)WtvPT_iQxSKtzFiOQ+Qp3GG(X;CE_Sx;!_*UD5KPRNp2SCP*m5tZ8zyrG)I!37DO=jT++)o5Vt zi?-M~3wS)Q(-zCO>_d1)6M1&wCW`pw{t2pOa=J2s;gQ$g>>SNMobN7@${(>{2>X&B zsC$Aap4#UYc4NLMMQ7*-97JE9_*N`^YL<>abp!Iu0zRynivF-WU|`==5Zgx~OzgWo zN}&9&p}u`X4;_A2&2PiypAq;sA57HS=3}(r-H+T9!sg)tvL?VGDh`)S>Y!P$XJl;F z!r{!t+zidBYVQgnbw5&i=mKgrd<(r4OmlKq=P7};RYgs?_^S9$dX z1?{0)Q62IfQvFLc80AKYyURd^N>m`BU`gzW*P{z+n7&_46V>)uJxBXCU0?!hW2HRh zu*0b_K>47H4lDK!(SGsq_26en@-N*)`)3&6v*UA4%GcDhXY(ddQOhn}`1II1BTDKY zJ2Lq=NAE$=#-m-AY_!Lz&o4AXp?q|EEs6*m@0CNwCopN|2fmepIdSuO?&|E5w;_S* zOeCT~Dr8o<{SJzQDqZg7WvV~jG@xvBa`z945sT)Q@=Q3@DHZRpZC6pue$^yPPmh0~ z+1p^xG~ZNkA0Z*Ef%br+0<4o|P=4l0?{i4~dm?x17v$fr8RmS26oqto(y-GIDE%IT zchpQPvZo_ymTs>;uNOE4O>yy33$E88z8z`XYdj{rxd>Ra^06wp4VEgqifo=R0IaUjLOpZla{=pmUOd5wk)lZ>Ci8lZ%{v}Y z2&Ofh0SQ)aJ+swe?*<#g@{=A9R|dyHa=hEYN`8Lsvj5-@fmlQi*R^6U0+9mZ$?j{Y z0F%=AwO8EcdHR+?t+Q{$o`(a9FQ!@K((!{g`IoyE&4%!mC7C5ZVeTVmLOfu^N1(|- zs1_6DQZ(|Zh)fDupT4{WB@c_0o%uv3+BuA34Bkjp&!gHYx4LuK%y&{BCVSs!RNn%I z+Y%kYI(!q@V`v)=remm&*O)Sdcx{iWq9pj0KthI40VTGuB?+eB%}oiCDSiTd~Yf{r?K1=KI73JZ)7PuC(wk(U4q@;O7;sGA2Vg* zcPq`k=49++9l7xkgB{6`XPb;x^NM*u3qyy_1rG&KZkN2p zgh2amK{}6hu_>?i-@h7l>*)Tid$nf{sFsR4E^FH=wJ0rGzI41Vfd!#j?{l&>4W#x@ivmQqYL1+XeQm{`IM+C8bi z*f$TU?Xeb>0J0MgoHIA017p-u@ukmlO>m2>DcI$tDe?NLW(gT4W)6Y4;8~*Jmd~F% zY(hJx#jGNdek8>p`NPM$WROEKOpGQzLkE>X|gdF2WbMIAU*1^eL7F z?5?!&bSo@Tv05al6ez^LPaqMZ(^MApUMS5ch~6MPcH!}&*-kTQds3Tl*6sLQ#+V8^ zjrf$Odn**V%Il%5Id^vywa0dU>=eW+tUs(16cz6UQ4A|OwX-#Sx`uHI*Ae42oD;|?>T`6sRs{U zU=ca-gF7jlX`>@CGnI}nO?mC~<0h2mUN5i>1=;$~w^AA%Sxj7HNl2|jiAwuH5Y>_s zAxT-B>^Vb7*+Fy(d8V*+OVgL_w8?iWbWIkW)4V-W7DlEhYEGPHPZr3D!McgHR0-p- zPXJWmJ8Qtl{bg{L4;xj!f$w)ezfwUrguJzQbq`*&;Ap(qX!T`Z4K?YYk5^k0OaSAm z9)Ws2gWwuJikMgm<#V)$!m)Acike~ zP}r29<(dPP-d0HEjcA8!Vizd_kTnt)1hV7t(8iaog zwx|mO8X&$j+0P8g3k~ap4HMiY;VuzJ8o!wsOI7!s4H{MVssGKvvlbcVjXi{#VHXX= z_b$mk7-WHn^B;4LpCt;ai~6^Pf1aw&pvAMB!7y571yAU`=KIs~WaO|G9>Td9$P~69 z8e+IvK=Ekz`RX%5oYH5zF9(^&OX6?4^K@&!MmPT&lWl#kMGrbO^JPtIqU>TpkU^~#Ty10XZ-CXEKjPqM~mX6oh7s5 z%Q4wN$Q?nY9xh|MM)CLnzv7`3$ck5$qOSn?AtdqvoGU?DF06dgY5n8n%FP#?^o3b@`EE4FuroRZIQ{)7bw?nb+DBY2+?ns%zU4Yomwr!E`Ssyp z09LyJA5!*r>82r^yS(&Tj2PSn`dL0%0IwK^{gxJUo1t+}sFWb)WvH?E z{mN z@IE45jY!;;#+SD6K6n!jL4Zwoszd%#1+A)&N3p-myh4*494Bx-^5D9w>EPD>4N4TIxR(fV11rmuI zcr1QS;1b+qlgH0=Df4Dn6eb_yH>zlGHTB|JgLX8Pm54{!lq-__Vig8bh?SDG^RyfGzycPZAMAWEvnIm>$GMH($Yez&yYRZx%OSvCN8+jDwF4{&?H@~ndu5U|U$df$ZqGA)__3n{WmnleQ2kEqMc zT(4!u%ogd%#5W^f<-Gj3C0M8vIOp#rMvoyIf{ zq+X*&>(LDLj|f`Vhke0?jdgM@p=u0q`R^*FLwcR#Z`%On`5QrBS)P042lin?488y(UCo~S(Dk1r{@HZ(;%SP;`EJ+7NovN+7d@WG|~4VTA)>uRVzB)(=dY zVll}-xq@GWYW7Dj7rA>U$e(~XcZU30ZAgyCBI=%9#~bySE=TOm*b5|oW$MH2nA7bT z(e0Sg?by-nIN)q*V01bYgVt2=D&ed3-TZ8Iind9(l^b$?*tPYZyosf)OXv5Azkbek z_q}QNJ%RqNWm|v4q3&xpRX~j3UI%kt;|+7(-ld&TEj-`M{lTwhkvIiG4mt{0|7i+q zHv)?m_iJ1COXPf5NTJw$05Z zA0})xQ^Gt_*q;+mc83VHJUIf7n?yKb3O49TTs_0?f84*?TtNKKGs36oS$iY zbNc4{jRhMC?eT*gWpw>ET|45bK)F^29H%{v8~@hX+y(qc)@%%8U!_6>1Vr%bKhp`o zAZS4UCD{J&lIQ;jw*PngKgqZMlLZFo|16LGuPp2Tmj6vG{a>5^M=t%}$0q(Gm;NvN z`M3PP9{Yb4Pycry|H<{=LH_T#kn#UNa{bS-3nf`_i2v*j_U}Ufr~Pi=|Bv@S08<89 AkpKVy literal 0 HcmV?d00001 diff --git a/project/MOOCSettings.scala b/project/MOOCSettings.scala new file mode 100644 index 0000000..171244f --- /dev/null +++ b/project/MOOCSettings.scala @@ -0,0 +1,46 @@ +package ch.epfl.lamp + +import sbt._ +import sbt.Keys._ + +/** + * Coursera uses two versions of each assignment. They both have the same assignment key and part id but have + * different item ids. + * + * @param key Assignment key + * @param partId Assignment partId + * @param itemId Item id of the non premium version + * @param premiumItemId Item id of the premium version (`None` if the assignment is optional) + */ +case class CourseraId(key: String, partId: String, itemId: String, premiumItemId: Option[String]) + +/** + * Settings shared by all assignments, reused in various tasks. + */ +object MOOCSettings extends AutoPlugin { + + object autoImport { + val course = SettingKey[String]("course") + val assignment = SettingKey[String]("assignment") + val options = SettingKey[Map[String, Map[String, String]]]("options") + val courseraId = settingKey[CourseraId]("Coursera-specific information identifying the assignment") + val testSuite = settingKey[String]("Fully qualified name of the test suite of this assignment") + // Convenient alias + type CourseraId = ch.epfl.lamp.CourseraId + val CourseraId = ch.epfl.lamp.CourseraId + } + + import autoImport._ + + override val globalSettings: Seq[Def.Setting[_]] = Seq( + // supershell is verbose, buggy and useless. + useSuperShell := false + ) + + override val projectSettings: Seq[Def.Setting[_]] = Seq( + parallelExecution in Test := false, + // Report test result after each test instead of waiting for every test to finish + logBuffered in Test := false, + name := s"${course.value}-${assignment.value}" + ) +} diff --git a/project/StudentTasks.scala b/project/StudentTasks.scala new file mode 100644 index 0000000..7604830 --- /dev/null +++ b/project/StudentTasks.scala @@ -0,0 +1,318 @@ +package ch.epfl.lamp + +import sbt._ +import Keys._ + +// import scalaj.http._ +import java.io.{File, FileInputStream, IOException} +import org.apache.commons.codec.binary.Base64 +// import play.api.libs.json.{Json, JsObject, JsPath} +import scala.util.{Failure, Success, Try} + +/** + * Provides tasks for submitting the assignment + */ +object StudentTasks extends AutoPlugin { + + override def requires = super.requires && MOOCSettings + + object autoImport { + val packageSourcesOnly = TaskKey[File]("packageSourcesOnly", "Package the sources of the project") + val packageBinWithoutResources = TaskKey[File]("packageBinWithoutResources", "Like packageBin, but without the resources") + val packageSubmissionZip = TaskKey[File]("packageSubmissionZip") + val packageSubmission = inputKey[Unit]("package solution as an archive file") + val runGradingTests = taskKey[Unit]("run black-box tests used for final grading") + } + + + import autoImport._ + import MOOCSettings.autoImport._ + + override lazy val projectSettings = Seq( + packageSubmissionSetting, + // submitSetting, + runGradingTestsSettings, + + fork := true, + connectInput in run := true, + outputStrategy := Some(StdoutOutput), + ) ++ packageSubmissionZipSettings + + lazy val runGradingTestsSettings = runGradingTests := { + val testSuiteJar = "grading-tests.jar" + if (!new File(testSuiteJar).exists) { + throw new MessageOnlyException(s"Could not find tests JarFile: $testSuiteJar") + } + + val classPath = s"${(Test / dependencyClasspath).value.map(_.data).mkString(File.pathSeparator)}${File.pathSeparator}$testSuiteJar" + val junitProcess = + Fork.java.fork( + ForkOptions(), + "-cp" :: classPath :: + "org.junit.runner.JUnitCore" :: + (Test / testSuite).value :: + Nil + ) + + // Wait for tests to complete. + junitProcess.exitValue() + } + + + /** ********************************************************** + * SUBMITTING A SOLUTION TO COURSERA + */ + + val packageSubmissionZipSettings = Seq( + packageSubmissionZip := { + val submission = crossTarget.value / "submission.zip" + val sources = (packageSourcesOnly in Compile).value + val binaries = (packageBinWithoutResources in Compile).value + IO.zip(Seq(sources -> "sources.zip", binaries -> "binaries.jar"), submission) + submission + }, + artifactClassifier in packageSourcesOnly := Some("sources"), + artifact in (Compile, packageBinWithoutResources) ~= (art => art.withName(art.name + "-without-resources")) + ) ++ + inConfig(Compile)( + Defaults.packageTaskSettings(packageSourcesOnly, Defaults.sourceMappings) ++ + Defaults.packageTaskSettings(packageBinWithoutResources, Def.task { + val relativePaths = + (unmanagedResources in Compile).value.flatMap(Path.relativeTo((unmanagedResourceDirectories in Compile).value)(_)) + (mappings in (Compile, packageBin)).value.filterNot { case (_, path) => relativePaths.contains(path) } + }) + ) + + val maxSubmitFileSize = { + val mb = 1024 * 1024 + 10 * mb + } + + /** Check that the jar exists, isn't empty, isn't crazy big, and can be read + * If so, encode jar as base64 so we can send it to Coursera + */ + def prepareJar(jar: File, s: TaskStreams): String = { + val errPrefix = "Error submitting assignment jar: " + val fileLength = jar.length() + if (!jar.exists()) { + s.log.error(errPrefix + "jar archive does not exist\n" + jar.getAbsolutePath) + failSubmit() + } else if (fileLength == 0L) { + s.log.error(errPrefix + "jar archive is empty\n" + jar.getAbsolutePath) + failSubmit() + } else if (fileLength > maxSubmitFileSize) { + s.log.error(errPrefix + "jar archive is too big. Allowed size: " + + maxSubmitFileSize + " bytes, found " + fileLength + " bytes.\n" + + jar.getAbsolutePath) + failSubmit() + } else { + val bytes = new Array[Byte](fileLength.toInt) + val sizeRead = try { + val is = new FileInputStream(jar) + val read = is.read(bytes) + is.close() + read + } catch { + case ex: IOException => + s.log.error(errPrefix + "failed to read sources jar archive\n" + ex.toString) + failSubmit() + } + if (sizeRead != bytes.length) { + s.log.error(errPrefix + "failed to read the sources jar archive, size read: " + sizeRead) + failSubmit() + } else encodeBase64(bytes) + } + } + + /** Task to package solution to a given file path */ + lazy val packageSubmissionSetting = packageSubmission := { + val args: Seq[String] = Def.spaceDelimited("[path]").parsed + val s: TaskStreams = streams.value // for logging + val jar = (packageSubmissionZip in Compile).value + + val base64Jar = prepareJar(jar, s) + + val path = args.headOption.getOrElse((baseDirectory.value / "submission.jar").absolutePath) + scala.tools.nsc.io.File(path).writeAll(base64Jar) + } + +/* + /** Task to submit a solution to coursera */ + val submit = inputKey[Unit]("submit solution to Coursera") + lazy val submitSetting = submit := { + // Fail if scalafix linting does not pass. + scalafixLinting.value + + val args: Seq[String] = Def.spaceDelimited("").parsed + val s: TaskStreams = streams.value // for logging + val jar = (packageSubmissionZip in Compile).value + + val assignmentDetails = + courseraId.?.value.getOrElse(throw new MessageOnlyException("This assignment can not be submitted to Coursera because the `courseraId` setting is undefined")) + val assignmentKey = assignmentDetails.key + val courseName = + course.value match { + case "capstone" => "scala-capstone" + case "bigdata" => "scala-spark-big-data" + case other => other + } + + val partId = assignmentDetails.partId + val itemId = assignmentDetails.itemId + val premiumItemId = assignmentDetails.premiumItemId + + val (email, secret) = args match { + case email :: secret :: Nil => + (email, secret) + case _ => + val inputErr = + s"""|Invalid input to `submit`. The required syntax for `submit` is: + |submit + | + |The submit token is NOT YOUR LOGIN PASSWORD. + |It can be obtained from the assignment page: + |https://www.coursera.org/learn/$courseName/programming/$itemId + |${ + premiumItemId.fold("") { id => + s"""or (for premium learners): + |https://www.coursera.org/learn/$courseName/programming/$id + """.stripMargin + } + } + """.stripMargin + s.log.error(inputErr) + failSubmit() + } + + val base64Jar = prepareJar(jar, s) + val json = + s"""|{ + | "assignmentKey":"$assignmentKey", + | "submitterEmail":"$email", + | "secret":"$secret", + | "parts":{ + | "$partId":{ + | "output":"$base64Jar" + | } + | } + |}""".stripMargin + + def postSubmission[T](data: String): Try[HttpResponse[String]] = { + val http = Http("https://www.coursera.org/api/onDemandProgrammingScriptSubmissions.v1") + val hs = List( + ("Cache-Control", "no-cache"), + ("Content-Type", "application/json") + ) + s.log.info("Connecting to Coursera...") + val response = Try(http.postData(data) + .headers(hs) + .option(HttpOptions.connTimeout(10000)) // scalaj default timeout is only 100ms, changing that to 10s + .asString) // kick off HTTP POST + response + } + + val connectMsg = + s"""|Attempting to submit "${assignment.value}" assignment in "$courseName" course + |Using: + |- email: $email + |- submit token: $secret""".stripMargin + s.log.info(connectMsg) + + def reportCourseraResponse(response: HttpResponse[String]): Unit = { + val code = response.code + val respBody = response.body + + /* Sample JSON response from Coursera + { + "message": "Invalid email or token.", + "details": { + "learnerMessage": "Invalid email or token." + } + } + */ + + // Success, Coursera responds with 2xx HTTP status code + if (response.is2xx) { + val successfulSubmitMsg = + s"""|Successfully connected to Coursera. (Status $code) + | + |Assignment submitted successfully! + | + |You can see how you scored by going to: + |https://www.coursera.org/learn/$courseName/programming/$itemId/ + |${ + premiumItemId.fold("") { id => + s"""or (for premium learners): + |https://www.coursera.org/learn/$courseName/programming/$id + """.stripMargin + } + } + |and clicking on "My Submission".""".stripMargin + s.log.info(successfulSubmitMsg) + } + + // Failure, Coursera responds with 4xx HTTP status code (client-side failure) + else if (response.is4xx) { + val result = Try(Json.parse(respBody)).toOption + val learnerMsg = result match { + case Some(resp: JsObject) => + (JsPath \ "details" \ "learnerMessage").read[String].reads(resp).get + case Some(x) => // shouldn't happen + "Could not parse Coursera's response:\n" + x + case None => + "Could not parse Coursera's response:\n" + respBody + } + val failedSubmitMsg = + s"""|Submission failed. + |There was something wrong while attempting to submit. + |Coursera says: + |$learnerMsg (Status $code)""".stripMargin + s.log.error(failedSubmitMsg) + } + + // Failure, Coursera responds with 5xx HTTP status code (server-side failure) + else if (response.is5xx) { + val failedSubmitMsg = + s"""|Submission failed. + |Coursera seems to be unavailable at the moment (Status $code) + |Check https://status.coursera.org/ and try again in a few minutes. + """.stripMargin + s.log.error(failedSubmitMsg) + } + + // Failure, Coursera repsonds with an unexpected status code + else { + val failedSubmitMsg = + s"""|Submission failed. + |Coursera replied with an unexpected code (Status $code) + """.stripMargin + s.log.error(failedSubmitMsg) + } + } + + // kick it all off, actually make request + postSubmission(json) match { + case Success(resp) => reportCourseraResponse(resp) + case Failure(e) => + val failedConnectMsg = + s"""|Connection to Coursera failed. + |There was something wrong while attempting to connect to Coursera. + |Check your internet connection. + |${e.toString}""".stripMargin + s.log.error(failedConnectMsg) + } + + } +*/ + + def failSubmit(): Nothing = { + sys.error("Submission failed") + } + + /** + * ***************** + * DEALING WITH JARS + */ + def encodeBase64(bytes: Array[Byte]): String = + new String(Base64.encodeBase64(bytes)) +} diff --git a/project/build.properties b/project/build.properties new file mode 100644 index 0000000..a919a9b --- /dev/null +++ b/project/build.properties @@ -0,0 +1 @@ +sbt.version=1.3.8 diff --git a/project/buildSettings.sbt b/project/buildSettings.sbt new file mode 100644 index 0000000..8fac702 --- /dev/null +++ b/project/buildSettings.sbt @@ -0,0 +1,5 @@ +// Used for Coursera submission (StudentPlugin) +// libraryDependencies += "org.scalaj" %% "scalaj-http" % "2.4.2" +// libraryDependencies += "com.typesafe.play" %% "play-json" % "2.7.4" +// Used for Base64 (StudentPlugin) +libraryDependencies += "commons-codec" % "commons-codec" % "1.10" diff --git a/project/plugins.sbt b/project/plugins.sbt new file mode 100644 index 0000000..017735d --- /dev/null +++ b/project/plugins.sbt @@ -0,0 +1,2 @@ +addSbtPlugin("org.scala-js" % "sbt-scalajs" % "0.6.28") +addSbtPlugin("ch.epfl.lamp" % "sbt-dotty" % "0.4.0") diff --git a/src/main/scala/lockfree/AbstractSortedList.scala b/src/main/scala/lockfree/AbstractSortedList.scala new file mode 100644 index 0000000..ef25f5b --- /dev/null +++ b/src/main/scala/lockfree/AbstractSortedList.scala @@ -0,0 +1,32 @@ +package lockfree + +import instrumentation.LockFreeMonitor +import scala.collection._ +import java.util.concurrent.atomic._ + + +abstract class AbstractSortedList extends LockFreeMonitor { + + def createNode(value: Int, tail: Option[Node], isHead: Boolean = false) = new Node(value, tail) { + override def toString = if(isHead) "HEAD" else super.toString + } + + def firstNode: Option[Node] + + def findNodeWithPrev(pred: Int => Boolean): (Node, Option[Node]) + + def insert(e: Int): Unit + + def delete(e: Int): Boolean + + def toList: List[Int] = { + var curr = firstNode + var list = List[Int]() + while(curr.nonEmpty) { + if(!curr.get.deleted) list :+= curr.get.value + curr = curr.get.next + } + list + } +} + diff --git a/src/main/scala/lockfree/AtomicVariable.scala b/src/main/scala/lockfree/AtomicVariable.scala new file mode 100644 index 0000000..18779d8 --- /dev/null +++ b/src/main/scala/lockfree/AtomicVariable.scala @@ -0,0 +1,25 @@ +package lockfree + +import java.util.concurrent.atomic._ + +abstract class AbstractAtomicVariable[T] { + def get: T + def compareAndSet(expect: T, newval: T) : Boolean +} + +class AtomicVariable[T](initial: T) extends AbstractAtomicVariable[T] { + + private val atomic = new AtomicReference[T](initial) + + override def get: T = atomic.get() + + override def compareAndSet(expected: T, value: T): Boolean = { + val current = atomic.get + if (current == expected) { + atomic.compareAndSet(current, value) + } + else { + false + } + } +} \ No newline at end of file diff --git a/src/main/scala/lockfree/Node.scala b/src/main/scala/lockfree/Node.scala new file mode 100644 index 0000000..6e7a84e --- /dev/null +++ b/src/main/scala/lockfree/Node.scala @@ -0,0 +1,22 @@ +package lockfree + +abstract class Node(val value: Int, initTail: Option[Node]) { + + // The type of mutable state held by node. + type State = Option[Node] + + // The initial mutable state of the node. + def initialState: State = initTail + + // The atomic variable that holds the state. + val atomicState: AbstractAtomicVariable[State] = new AtomicVariable[State](initialState) + + // Function to read the next node from the state. + def next: Option[Node] = atomicState.get + + // Should return true if the node was marked as deleted. + def deleted: Boolean = false + + // Should mark the node as deleted. + def mark: Boolean = ??? +} diff --git a/src/main/scala/lockfree/SortedList.scala b/src/main/scala/lockfree/SortedList.scala new file mode 100644 index 0000000..c88bd4c --- /dev/null +++ b/src/main/scala/lockfree/SortedList.scala @@ -0,0 +1,26 @@ +package lockfree + +import scala.annotation.tailrec + +class SortedList extends AbstractSortedList { + + // The sentinel node at the head. + private val _head: Node = createNode(0, None, isHead = true) + + // The first logical node is referenced by the head. + def firstNode: Option[Node] = _head.next + + // Finds the first node whose value satisfies the predicate. + // Returns the predecessor of the node and the node. + def findNodeWithPrev(pred: Int => Boolean): (Node, Option[Node]) = ??? + + // Insert an element in the list. + def insert(e: Int): Unit = ??? + + // Checks if the list contains an element. + def contains(e: Int): Boolean = ??? + + // Delete an element from the list. + // Should only delete one element when multiple occurences are present. + def delete(e: Int): Boolean = ??? +} diff --git a/src/main/scala/lockfree/SortingBenchmarks.scala b/src/main/scala/lockfree/SortingBenchmarks.scala new file mode 100644 index 0000000..bc93c98 --- /dev/null +++ b/src/main/scala/lockfree/SortingBenchmarks.scala @@ -0,0 +1,83 @@ +package lockfree + +import scala.util.Random +import instrumentation.Stats._ + +object SortingBenchmarks extends App { + + val items = 2500 + val threads = 12 + + val insertions = List.fill(items)(Random.nextInt) + val deletions = Random.shuffle(insertions) + val parInsertions = parBuckets(threads, insertions)_ + val parDeletions = parBuckets(threads, deletions)_ + + private class SynchronizedSortedList extends SortedList { + val lock = new Object + def syncTraverse(stop: Int => Boolean) = lock.synchronized(super.findNodeWithPrev(stop)) + def syncInsert(e: Int) = lock.synchronized(super.insert(e)) + def syncDelete(e: Int) = lock.synchronized(super.delete(e)) + } + + val syncSeqTime = { + val l = new SynchronizedSortedList + withTime { + insertions.foreach(l.syncInsert) + deletions.foreach(l.syncDelete) + }._2 + } + + val syncParTime = { + val l = new SynchronizedSortedList + withTime { + val ops = parInsertions(l.syncInsert) ++ parDeletions(l.syncDelete) + ops.foreach(_.start) + ops.foreach(_.join) + }._2 + } + + val fineSeqTime = { + val l = new SortedList + withTime { + insertions.foreach(l.insert) + deletions.foreach(l.delete) + }._2 + } + + val fineParTime = { + val l = new SortedList + withTime { + val ops = parInsertions(l.insert) ++ parDeletions(l.delete) + ops.foreach(_.start) + ops.foreach(_.join) + }._2 + } + + println( + s""" + |***** Results ***** + | + |synchronized: + | - sequential: $syncSeqTime + | - parallel: $syncParTime + | - speedup: ${syncSeqTime / syncParTime} + | + |fine-grained: + | - sequential: $fineSeqTime + | - parallel: $fineParTime + | - speedup: ${fineSeqTime / fineParTime} + | + |synchronized/fine-grained: + | - speedup: ${syncParTime / fineParTime} + | + """.stripMargin) + + def parBuckets(threadCount: Int, items: List[Int])(op: Int => Any): List[Thread] = + items.grouped(items.size / threadCount).toList.map { bucket => + new Thread { + override def run(): Unit = + bucket.foreach(op) + } + } +} diff --git a/src/main/scala/lockfree/instrumentation/MockedMonitor.scala b/src/main/scala/lockfree/instrumentation/MockedMonitor.scala new file mode 100644 index 0000000..2ad9bc5 --- /dev/null +++ b/src/main/scala/lockfree/instrumentation/MockedMonitor.scala @@ -0,0 +1,72 @@ +package instrumentation + +trait MockedMonitor extends Monitor { + def scheduler: Scheduler + + // Can be overriden. + override def waitDefault() = { + scheduler.log("wait") + scheduler updateThreadState Wait(this, scheduler.threadLocks.tail) + } + override def synchronizedDefault[T](toExecute: =>T): T = { + scheduler.log("synchronized check") + val prevLocks = scheduler.threadLocks + scheduler updateThreadState Sync(this, prevLocks) // If this belongs to prevLocks, should just continue. + scheduler.log("synchronized -> enter") + try { + toExecute + } finally { + scheduler updateThreadState Running(prevLocks) + scheduler.log("synchronized -> out") + } + } + override def notifyDefault() = { + scheduler mapOtherStates { + state => state match { + case Wait(lockToAquire, locks) if lockToAquire == this => SyncUnique(this, state.locks) + case e => e + } + } + scheduler.log("notify") + } + override def notifyAllDefault() = { + scheduler mapOtherStates { + state => state match { + case Wait(lockToAquire, locks) if lockToAquire == this => Sync(this, state.locks) + case SyncUnique(lockToAquire, locks) if lockToAquire == this => Sync(this, state.locks) + case e => e + } + } + scheduler.log("notifyAll") + } +} + +trait LockFreeMonitor extends Monitor { + override def waitDefault() = { + throw new Exception("Please use lock-free structures and do not use wait()") + } + override def synchronizedDefault[T](toExecute: =>T): T = { + throw new Exception("Please use lock-free structures and do not use synchronized()") + } + override def notifyDefault() = { + throw new Exception("Please use lock-free structures and do not use notify()") + } + override def notifyAllDefault() = { + throw new Exception("Please use lock-free structures and do not use notifyAll()") + } +} + + +abstract class ThreadState { + def locks: Seq[AnyRef] +} +trait CanContinueIfAcquiresLock extends ThreadState { + def lockToAquire: AnyRef +} +case object Start extends ThreadState { def locks: Seq[AnyRef] = Seq.empty } +case object End extends ThreadState { def locks: Seq[AnyRef] = Seq.empty } +case class Wait(lockToAquire: AnyRef, locks: Seq[AnyRef]) extends ThreadState +case class SyncUnique(lockToAquire: AnyRef, locks: Seq[AnyRef]) extends ThreadState with CanContinueIfAcquiresLock +case class Sync(lockToAquire: AnyRef, locks: Seq[AnyRef]) extends ThreadState with CanContinueIfAcquiresLock +case class Running(locks: Seq[AnyRef]) extends ThreadState +case class VariableReadWrite(locks: Seq[AnyRef]) extends ThreadState diff --git a/src/main/scala/lockfree/instrumentation/Monitor.scala b/src/main/scala/lockfree/instrumentation/Monitor.scala new file mode 100644 index 0000000..82c6cf2 --- /dev/null +++ b/src/main/scala/lockfree/instrumentation/Monitor.scala @@ -0,0 +1,23 @@ +package instrumentation + +class Dummy + +trait Monitor { + implicit val dummy: Dummy = new Dummy + + def wait()(implicit i: Dummy) = waitDefault() + + def synchronized[T](e: => T)(implicit i: Dummy) = synchronizedDefault(e) + + def notify()(implicit i: Dummy) = notifyDefault() + + def notifyAll()(implicit i: Dummy) = notifyAllDefault() + + private val lock = new AnyRef + + // Can be overriden. + def waitDefault(): Unit = lock.wait() + def synchronizedDefault[T](toExecute: =>T): T = lock.synchronized(toExecute) + def notifyDefault(): Unit = lock.notify() + def notifyAllDefault(): Unit = lock.notifyAll() +} diff --git a/src/main/scala/lockfree/instrumentation/SchedulableSortedList.scala b/src/main/scala/lockfree/instrumentation/SchedulableSortedList.scala new file mode 100644 index 0000000..5811443 --- /dev/null +++ b/src/main/scala/lockfree/instrumentation/SchedulableSortedList.scala @@ -0,0 +1,35 @@ +package instrumentation + +import scala.annotation.tailrec +import lockfree._ +import java.util.concurrent.atomic._ + +class SchedulableAtomicVariable[T](initial: T, scheduler: Scheduler, self: Node) extends AbstractAtomicVariable[T] { + private val proxied: AtomicVariable[T] = new AtomicVariable[T](initial) + + override def get: T = scheduler.exec { + proxied.get + } ( s"", Some(res => s"$self: get $res") ) + + override def compareAndSet(expect: T, newval: T): Boolean = { + scheduler.exec { + proxied.compareAndSet(expect, newval) + } (s"$self: compareAndSet expect = $expect, newval = $newval", Some(res => s"$self: Did it set? $res") ) + } +} + +class SchedulableNode(value: Int, initTail: Option[Node], val scheduler: Scheduler) extends Node(value, initTail) with LockFreeMonitor { self => + + override val atomicState: AbstractAtomicVariable[State] = new SchedulableAtomicVariable[State](initialState, scheduler, this) + + override def toString: String = + String.format("Node(%s)#%02d", value.toString, new Integer(## % 100)) +} + +class SchedulableSortedList(val scheduler: Scheduler) extends SortedList with LockFreeMonitor { + + override def createNode(value: Int, tail: Option[Node], isHead: Boolean) = new SchedulableNode(value, tail, scheduler) { + override def toString = if(isHead) "HEAD" else super.toString + } +} + diff --git a/src/main/scala/lockfree/instrumentation/Scheduler.scala b/src/main/scala/lockfree/instrumentation/Scheduler.scala new file mode 100644 index 0000000..4d30606 --- /dev/null +++ b/src/main/scala/lockfree/instrumentation/Scheduler.scala @@ -0,0 +1,304 @@ +package instrumentation + +import java.util.concurrent._; +import scala.concurrent.duration._ +import scala.collection.mutable._ +import Stats._ + +import java.util.concurrent.atomic.AtomicInteger + +sealed abstract class Result +case class RetVal(rets: List[Any]) extends Result +case class Except(msg: String, stackTrace: Array[StackTraceElement]) extends Result +case class Timeout(msg: String) extends Result + +/** + * A class that maintains schedule and a set of thread ids. + * The schedules are advanced after an operation of a SchedulableBuffer is performed. + * Note: the real schedule that is executed may deviate from the input schedule + * due to the adjustments that had to be made for locks + */ +class Scheduler(sched: List[Int]) { + val maxOps = 500 // a limit on the maximum number of operations the code is allowed to perform + + private var schedule = sched + private var numThreads = 0 + private val realToFakeThreadId = Map[Long, Int]() + private val opLog = ListBuffer[String]() // a mutable list (used for efficient concat) + private val threadStates = Map[Int, ThreadState]() + + /** + * Runs a set of operations in parallel as per the schedule. + * Each operation may consist of many primitive operations like reads or writes + * to shared data structure each of which should be executed using the function `exec`. + * @timeout in milliseconds + * @return true - all threads completed on time, false -some tests timed out. + */ + def runInParallel(timeout: Long, ops: List[() => Any]): Result = { + numThreads = ops.length + val threadRes = Array.fill(numThreads) { None: Any } + var exception: Option[Except] = None + val syncObject = new Object() + var completed = new AtomicInteger(0) + // create threads + val threads = ops.zipWithIndex.map { + case (op, i) => + new Thread(new Runnable() { + def run(): Unit = { + val fakeId = i + 1 + setThreadId(fakeId) + try { + updateThreadState(Start) + val res = op() + updateThreadState(End) + threadRes(i) = res + // notify the master thread if all threads have completed + if (completed.incrementAndGet() == ops.length) { + syncObject.synchronized { syncObject.notifyAll() } + } + } catch { + case e: Throwable if exception != None => // do nothing here and silently fail + case e: Throwable => + log(s"throw ${e.toString}") + exception = Some(Except(s"Thread $fakeId crashed on the following schedule: \n" + opLog.mkString("\n"), + e.getStackTrace)) + syncObject.synchronized { syncObject.notifyAll() } + //println(s"$fakeId: ${e.toString}") + //Runtime.getRuntime().halt(0) //exit the JVM and all running threads (no other way to kill other threads) + } + } + }) + } + // start all threads + threads.foreach(_.start()) + // wait for all threads to complete, or for an exception to be thrown, or for the time out to expire + var remTime = timeout + syncObject.synchronized { + timed { if(completed.get() != ops.length) syncObject.wait(timeout) } { time => remTime -= time } + } + if (exception.isDefined) { + exception.get + } else if (remTime <= 1) { // timeout ? using 1 instead of zero to allow for some errors + Timeout(opLog.mkString("\n")) + } else { + // every thing executed normally + RetVal(threadRes.toList) + } + } + + // Updates the state of the current thread + def updateThreadState(state: ThreadState): Unit = { + val tid = threadId + synchronized { + threadStates(tid) = state + } + state match { + case Sync(lockToAquire, locks) => + if (locks.indexOf(lockToAquire) < 0) waitForTurn else { + // Re-aqcuiring the same lock + updateThreadState(Running(lockToAquire +: locks)) + } + case Start => waitStart() + case End => removeFromSchedule(tid) + case Running(_) => + case _ => waitForTurn // Wait, SyncUnique, VariableReadWrite + } + } + + def waitStart(): Unit = { + //while (threadStates.size < numThreads) { + //Thread.sleep(1) + //} + synchronized { + if (threadStates.size < numThreads) { + wait() + } else { + notifyAll() + } + } + } + + def threadLocks = { + synchronized { + threadStates(threadId).locks + } + } + + def threadState = { + synchronized { + threadStates(threadId) + } + } + + def mapOtherStates(f: ThreadState => ThreadState) = { + val exception = threadId + synchronized { + for (k <- threadStates.keys if k != exception) { + threadStates(k) = f(threadStates(k)) + } + } + } + + def log(str: String) = { + if((realToFakeThreadId contains Thread.currentThread().getId())) { + val space = (" " * ((threadId - 1) * 2)) + val s = space + threadId + ":" + "\n".r.replaceAllIn(str, "\n" + space + " ") + opLog += s + } + } + + /** + * Executes a read or write operation to a global data structure as per the given schedule + * @param msg a message corresponding to the operation that will be logged + */ + def exec[T](primop: => T)(msg: => String, postMsg: => Option[T => String] = None): T = { + if(! (realToFakeThreadId contains Thread.currentThread().getId())) { + primop + } else { + updateThreadState(VariableReadWrite(threadLocks)) + val m = msg + if(m != "") log(m) + if (opLog.size > maxOps) + throw new Exception(s"Total number of reads/writes performed by threads exceed $maxOps. A possible deadlock!") + val res = primop + postMsg match { + case Some(m) => log(m(res)) + case None => + } + res + } + } + + private def setThreadId(fakeId: Int) = synchronized { + realToFakeThreadId(Thread.currentThread.getId) = fakeId + } + + def threadId = + try { + realToFakeThreadId(Thread.currentThread().getId()) + } catch { + case e: NoSuchElementException => + throw new Exception("You are accessing shared variables in the constructor. This is not allowed. The variables are already initialized!") + } + + private def isTurn(tid: Int) = synchronized { + (!schedule.isEmpty && schedule.head != tid) + } + + def canProceed(): Boolean = { + val tid = threadId + canContinue match { + case Some((i, state)) if i == tid => + //println(s"$tid: Runs ! Was in state $state") + canContinue = None + state match { + case Sync(lockToAquire, locks) => updateThreadState(Running(lockToAquire +: locks)) + case SyncUnique(lockToAquire, locks) => + mapOtherStates { + _ match { + case SyncUnique(lockToAquire2, locks2) if lockToAquire2 == lockToAquire => Wait(lockToAquire2, locks2) + case e => e + } + } + updateThreadState(Running(lockToAquire +: locks)) + case VariableReadWrite(locks) => updateThreadState(Running(locks)) + } + true + case Some((i, state)) => + //println(s"$tid: not my turn but $i !") + false + case None => + false + } + } + + var threadPreference = 0 // In the case the schedule is over, which thread should have the preference to execute. + + /** returns true if the thread can continue to execute, and false otherwise */ + def decide(): Option[(Int, ThreadState)] = { + if (!threadStates.isEmpty) { // The last thread who enters the decision loop takes the decision. + //println(s"$threadId: I'm taking a decision") + if (threadStates.values.forall { case e: Wait => true case _ => false }) { + val waiting = threadStates.keys.map(_.toString).mkString(", ") + val s = if (threadStates.size > 1) "s" else "" + val are = if (threadStates.size > 1) "are" else "is" + throw new Exception(s"Deadlock: Thread$s $waiting $are waiting but all others have ended and cannot notify them.") + } else { + // Threads can be in Wait, Sync, SyncUnique, and VariableReadWrite mode. + // Let's determine which ones can continue. + val notFree = threadStates.collect { case (id, state) => state.locks }.flatten.toSet + val threadsNotBlocked = threadStates.toSeq.filter { + case (id, v: VariableReadWrite) => true + case (id, v: CanContinueIfAcquiresLock) => !notFree(v.lockToAquire) || (v.locks contains v.lockToAquire) + case _ => false + } + if (threadsNotBlocked.isEmpty) { + val waiting = threadStates.keys.map(_.toString).mkString(", ") + val s = if (threadStates.size > 1) "s" else "" + val are = if (threadStates.size > 1) "are" else "is" + val whoHasLock = threadStates.toSeq.flatMap { case (id, state) => state.locks.map(lock => (lock, id)) }.toMap + val reason = threadStates.collect { + case (id, state: CanContinueIfAcquiresLock) if !notFree(state.lockToAquire) => + s"Thread $id is waiting on lock ${state.lockToAquire} held by thread ${whoHasLock(state.lockToAquire)}" + }.mkString("\n") + throw new Exception(s"Deadlock: Thread$s $waiting are interlocked. Indeed:\n$reason") + } else if (threadsNotBlocked.size == 1) { // Do not consume the schedule if only one thread can execute. + Some(threadsNotBlocked(0)) + } else { + val next = schedule.indexWhere(t => threadsNotBlocked.exists { case (id, state) => id == t }) + if (next != -1) { + //println(s"$threadId: schedule is $schedule, next chosen is ${schedule(next)}") + val chosenOne = schedule(next) // TODO: Make schedule a mutable list. + schedule = schedule.take(next) ++ schedule.drop(next + 1) + Some((chosenOne, threadStates(chosenOne))) + } else { + threadPreference = (threadPreference + 1) % threadsNotBlocked.size + val chosenOne = threadsNotBlocked(threadPreference) // Maybe another strategy + Some(chosenOne) + //threadsNotBlocked.indexOf(threadId) >= 0 + /* + val tnb = threadsNotBlocked.map(_._1).mkString(",") + val s = if (schedule.isEmpty) "empty" else schedule.mkString(",") + val only = if (schedule.isEmpty) "" else " only" + throw new Exception(s"The schedule is $s but$only threads ${tnb} can continue")*/ + } + } + } + } else canContinue + } + + /** + * This will be called before a schedulable operation begins. + * This should not use synchronized + */ + var numThreadsWaiting = new AtomicInteger(0) + //var waitingForDecision = Map[Int, Option[Int]]() // Mapping from thread ids to a number indicating who is going to make the choice. + var canContinue: Option[(Int, ThreadState)] = None // The result of the decision thread Id of the thread authorized to continue. + private def waitForTurn = { + synchronized { + if (numThreadsWaiting.incrementAndGet() == threadStates.size) { + canContinue = decide() + notifyAll() + } + //waitingForDecision(threadId) = Some(numThreadsWaiting) + //println(s"$threadId Entering waiting with ticket number $numThreadsWaiting/${waitingForDecision.size}") + while (!canProceed()) wait() + } + numThreadsWaiting.decrementAndGet() + } + + /** + * To be invoked when a thread is about to complete + */ + private def removeFromSchedule(fakeid: Int) = synchronized { + //println(s"$fakeid: I'm taking a decision because I finished") + schedule = schedule.filterNot(_ == fakeid) + threadStates -= fakeid + if (numThreadsWaiting.get() == threadStates.size) { + canContinue = decide() + notifyAll() + } + } + + def getOperationLog() = opLog +} diff --git a/src/main/scala/lockfree/instrumentation/Stats.scala b/src/main/scala/lockfree/instrumentation/Stats.scala new file mode 100644 index 0000000..ab221bf --- /dev/null +++ b/src/main/scala/lockfree/instrumentation/Stats.scala @@ -0,0 +1,23 @@ +/* Copyright 2009-2015 EPFL, Lausanne */ +package instrumentation + +import java.lang.management._ + +/** + * A collection of methods that can be used to collect run-time statistics about Leon programs. + * This is mostly used to test the resources properties of Leon programs + */ +object Stats { + def timed[T](code: => T)(cont: Long => Unit): T = { + var t1 = System.currentTimeMillis() + val r = code + cont((System.currentTimeMillis() - t1)) + r + } + + def withTime[T](code: => T): (T, Long) = { + var t1 = System.currentTimeMillis() + val r = code + (r, (System.currentTimeMillis() - t1)) + } +} diff --git a/src/main/scala/lockfree/instrumentation/TestHelper.scala b/src/main/scala/lockfree/instrumentation/TestHelper.scala new file mode 100644 index 0000000..f47df4f --- /dev/null +++ b/src/main/scala/lockfree/instrumentation/TestHelper.scala @@ -0,0 +1,124 @@ +package instrumentation + +import scala.util.Random +import scala.collection.mutable.{Map => MutableMap} + +import Stats._ + +object TestHelper { + val noOfSchedules = 10000 // set this to 100k during deployment + val readWritesPerThread = 20 // maximum number of read/writes possible in one thread + val contextSwitchBound = 10 + val testTimeout = 240 // the total time out for a test in seconds + val schedTimeout = 15 // the total time out for execution of a schedule in secs + + // Helpers + /*def testManySchedules(op1: => Any): Unit = testManySchedules(List(() => op1)) + def testManySchedules(op1: => Any, op2: => Any): Unit = testManySchedules(List(() => op1, () => op2)) + def testManySchedules(op1: => Any, op2: => Any, op3: => Any): Unit = testManySchedules(List(() => op1, () => op2, () => op3)) + def testManySchedules(op1: => Any, op2: => Any, op3: => Any, op4: => Any): Unit = testManySchedules(List(() => op1, () => op2, () => op3, () => op4))*/ + + def testSequential[T](ops: Scheduler => Any)(assertions: T => (Boolean, String)) = + testManySchedules(1, + (sched: Scheduler) => { + (List(() => ops(sched)), + (res: List[Any]) => assertions(res.head.asInstanceOf[T])) + }) + + /** + * @numThreads number of threads + * @ops operations to be executed, one per thread + * @assertion as condition that will executed after all threads have completed (without exceptions) + * the arguments are the results of the threads + */ + def testManySchedules(numThreads: Int, + ops: Scheduler => + (List[() => Any], // Threads + List[Any] => (Boolean, String)) // Assertion + ) = { + var timeout = testTimeout * 1000L + val threadIds = (1 to numThreads) + //(1 to scheduleLength).flatMap(_ => threadIds).toList.permutations.take(noOfSchedules).foreach { + val schedules = (new ScheduleGenerator(numThreads)).schedules() + var schedsExplored = 0 + schedules.takeWhile(_ => schedsExplored <= noOfSchedules && timeout > 0).foreach { + //case _ if timeout <= 0 => // break + case schedule => + schedsExplored += 1 + val schedr = new Scheduler(schedule) + //println("Exploring Sched: "+schedule) + val (threadOps, assertion) = ops(schedr) + if (threadOps.size != numThreads) + throw new IllegalStateException(s"Number of threads: $numThreads, do not match operations of threads: $threadOps") + timed { schedr.runInParallel(schedTimeout * 1000, threadOps) } { t => timeout -= t } match { + case Timeout(msg) => + throw new java.lang.AssertionError("assertion failed\n"+"The schedule took too long to complete. A possible deadlock! \n"+msg) + case Except(msg, stkTrace) => + val traceStr = "Thread Stack trace: \n"+stkTrace.map(" at "+_.toString).mkString("\n") + throw new java.lang.AssertionError("assertion failed\n"+msg+"\n"+traceStr) + case RetVal(threadRes) => + // check the assertion + val (success, custom_msg) = assertion(threadRes) + if (!success) { + val msg = "The following schedule resulted in wrong results: \n" + custom_msg + "\n" + schedr.getOperationLog().mkString("\n") + throw new java.lang.AssertionError("Assertion failed: "+msg) + } + } + } + if (timeout <= 0) { + throw new java.lang.AssertionError("Test took too long to complete! Cannot check all schedules as your code is too slow!") + } + } + + /** + * A schedule generator that is based on the context bound + */ + class ScheduleGenerator(numThreads: Int) { + val scheduleLength = readWritesPerThread * numThreads + val rands = (1 to scheduleLength).map(i => new Random(0xcafe * i)) // random numbers for choosing a thread at each position + def schedules(): LazyList[List[Int]] = { + var contextSwitches = 0 + var contexts = List[Int]() // a stack of thread ids in the order of context-switches + val remainingOps = MutableMap[Int, Int]() + remainingOps ++= (1 to numThreads).map(i => (i, readWritesPerThread)) // num ops remaining in each thread + val liveThreads = (1 to numThreads).toSeq.toBuffer + + /** + * Updates remainingOps and liveThreads once a thread is chosen for a position in the schedule + */ + def updateState(tid: Int): Unit = { + val remOps = remainingOps(tid) + if (remOps == 0) { + liveThreads -= tid + } else { + remainingOps += (tid -> (remOps - 1)) + } + } + val schedule = rands.foldLeft(List[Int]()) { + case (acc, r) if contextSwitches < contextSwitchBound => + val tid = liveThreads(r.nextInt(liveThreads.size)) + contexts match { + case prev :: tail if prev != tid => // we have a new context switch here + contexts +:= tid + contextSwitches += 1 + case prev :: tail => + case _ => // init case + contexts +:= tid + } + updateState(tid) + acc :+ tid + case (acc, _) => // here context-bound has been reached so complete the schedule without any more context switches + if (!contexts.isEmpty) { + contexts = contexts.dropWhile(remainingOps(_) == 0) + } + val tid = contexts match { + case top :: tail => top + case _ => liveThreads(0) // here, there has to be threads that have not even started + } + updateState(tid) + acc :+ tid + } + schedule #:: schedules() + } + } +} diff --git a/src/main/scala/lockfree/instrumentation/TestUtils.scala b/src/main/scala/lockfree/instrumentation/TestUtils.scala new file mode 100644 index 0000000..3ab651a --- /dev/null +++ b/src/main/scala/lockfree/instrumentation/TestUtils.scala @@ -0,0 +1,19 @@ +package instrumentation + +import scala.concurrent._ +import scala.concurrent.duration._ +import scala.concurrent.ExecutionContext.Implicits.global + +object TestUtils { + def failsOrTimesOut[T](action: => T): Boolean = { + val asyncAction = Future { + action + } + try { + Await.result(asyncAction, 2000.millisecond) + } catch { + case _: Throwable => return true + } + return false + } +} diff --git a/src/test/scala/lockfree/SortedListSuite.scala b/src/test/scala/lockfree/SortedListSuite.scala new file mode 100644 index 0000000..6ce36e0 --- /dev/null +++ b/src/test/scala/lockfree/SortedListSuite.scala @@ -0,0 +1,96 @@ +package lockfree + +import scala.concurrent._ +import scala.concurrent.duration._ +import scala.collection.mutable.HashMap +import scala.util.Random +import org.junit._ +import org.junit.Assert.assertEquals +import instrumentation.SchedulableSortedList +import instrumentation.TestHelper._ +import instrumentation.TestUtils._ + +class SortedListSuite { + + @Test def `Should work when 1, 2, and 3 are inserted sequentially`: Unit = { + val l = new SortedList + l.insert(1) + l.insert(2) + l.insert(3) + assertEquals(List(1, 2, 3), l.toList) + } + + @Test def `Should work when 3, 2, and 1 are inserted sequentially`: Unit = { + val l = new SortedList + l.insert(3) + l.insert(2) + l.insert(1) + assertEquals(List(1, 2, 3), l.toList) + } + + @Test def `Should work when duplicate elements are inserted sequentially`: Unit = { + val l = new SortedList + l.insert(0) + l.insert(0) + l.insert(2) + l.insert(2) + assertEquals(List(0, 0, 2, 2), l.toList) + } + + @Test def `Should return [1,4,5] when from [1,2,3,5], a thread removes 3, 2 and then inserts 4`: Unit = { + val l = new SortedList + l.insert(1) + l.insert(2) + l.insert(3) + l.insert(5) + l.delete(3) + l.delete(2) + l.insert(4) + assertEquals(List(1, 4, 5), l.toList) + } + + @Test def `Should work when a random list of 100 elements are inserted sequentially`: Unit = { + val rand = new Random() + val randvals = for(i <- 1 to 100) yield rand.nextInt() + val l = new SortedList + randvals.foreach { l.insert } + assertEquals(randvals.sorted, l.toList) + } + + @Test def `Should insert in parallel 1, 2 and 3 in the list (0, 4)`: Unit = { + testManySchedules(3, sched => { + val sortedList = new SchedulableSortedList(sched) + sortedList.insert(0) + sortedList.insert(4) + ((for(i <- 1 to 3) yield () => sortedList.insert(i)).toList, + results => { + val res = sortedList.toList + (res == List(0, 1, 2, 3, 4), + s"expected List(0, 1, 2, 3, 4), got $res") + }) + }) + } + + @Test def `Should return List(true, false) when the first thread deletes 2 and the sec`: Unit = { + testManySchedules(2, sched => { + val sortedList = new SchedulableSortedList(sched) + sortedList.insert(1) + sortedList.insert(2) + sortedList.insert(3) + (List(() => sortedList.delete(2), + () => sortedList.delete(4)), + results => { + val res = sortedList.toList + val expected = List(1, 3) + if(res != expected) { + (false, s"expected the final list to be $expected, your 'delete' implementation returned ${res}") + } else if(results != List(true, false)) { + (false, s"expected threads to return List(true, false), your 'delete' implementation returned ${results}") + } else (true, "") + }) + }) + } + + + @Rule def individualTestTimeout = new org.junit.rules.Timeout(200 * 1000) +}