From 58b079621fa545c33a5b6c958767a39c628ae875 Mon Sep 17 00:00:00 2001 From: Guillaume Martres Date: Tue, 19 Feb 2019 20:44:23 +0100 Subject: [PATCH] Add barneshut assignment --- .gitignore | 16 + .gitlab-ci.yml | 36 ++ .vscode/settings.json | 8 + assignment.sbt | 4 + build.sbt | 11 + grading-tests.jar | Bin 0 -> 17098 bytes project/MOOCSettings.scala | 46 +++ project/StudentTasks.scala | 318 ++++++++++++++++++ project/build.properties | 1 + project/buildSettings.sbt | 5 + project/plugins.sbt | 2 + src/main/scala/pubsub/Client.scala | 34 ++ src/main/scala/pubsub/Server.scala | 51 +++ .../collection/AbstractBoundedBuffer.scala | 34 ++ .../pubsub/collection/BoundedBuffer.scala | 26 ++ .../collection/ConcurrentMultiMap.scala | 65 ++++ .../scala/pubsub/command/CommandHandler.scala | 43 +++ .../scala/pubsub/command/CommandReader.scala | 45 +++ .../scala/pubsub/command/CommandType.scala | 33 ++ .../instrumentation/MockedMonitor.scala | 72 ++++ .../pubsub/instrumentation/Monitor.scala | 23 ++ .../SchedulableInternalBuffer.scala | 63 ++++ .../pubsub/instrumentation/Scheduler.scala | 305 +++++++++++++++++ .../scala/pubsub/instrumentation/Stats.scala | 23 ++ .../pubsub/instrumentation/TestHelper.scala | 124 +++++++ .../pubsub/instrumentation/TestUtils.scala | 19 ++ src/main/scala/pubsub/network/TCPReader.scala | 32 ++ .../scala/pubsub/BoundedBufferSuite.scala | 43 +++ 28 files changed, 1482 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/pubsub/Client.scala create mode 100644 src/main/scala/pubsub/Server.scala create mode 100644 src/main/scala/pubsub/collection/AbstractBoundedBuffer.scala create mode 100644 src/main/scala/pubsub/collection/BoundedBuffer.scala create mode 100644 src/main/scala/pubsub/collection/ConcurrentMultiMap.scala create mode 100644 src/main/scala/pubsub/command/CommandHandler.scala create mode 100644 src/main/scala/pubsub/command/CommandReader.scala create mode 100644 src/main/scala/pubsub/command/CommandType.scala create mode 100644 src/main/scala/pubsub/instrumentation/MockedMonitor.scala create mode 100644 src/main/scala/pubsub/instrumentation/Monitor.scala create mode 100644 src/main/scala/pubsub/instrumentation/SchedulableInternalBuffer.scala create mode 100644 src/main/scala/pubsub/instrumentation/Scheduler.scala create mode 100644 src/main/scala/pubsub/instrumentation/Stats.scala create mode 100644 src/main/scala/pubsub/instrumentation/TestHelper.scala create mode 100644 src/main/scala/pubsub/instrumentation/TestUtils.scala create mode 100644 src/main/scala/pubsub/network/TCPReader.scala create mode 100644 src/test/scala/pubsub/BoundedBufferSuite.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..4dec89f --- /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-pubsub-2020-03-15 + 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..e27eb6e --- /dev/null +++ b/build.sbt @@ -0,0 +1,11 @@ +course := "concpar" +assignment := "pubsub" + +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 := "pubsub.BoundedBufferSuite" diff --git a/grading-tests.jar b/grading-tests.jar new file mode 100644 index 0000000000000000000000000000000000000000..3bc9c020808dc6f1a4004aa61be6881d28617603 GIT binary patch literal 17098 zcmbTcQ+Os!(Eb_Qwr$(C^Te4vv29LlO>EoF#P-CtZ9D(>+wWla+CAGosp{@J=sIXr z|85m|a0m<#7#J9kg$@T5kpD-+0D%Bel++MqlvR>qRuol|m6BA~U{aL&o&W)foo;{| zU_lwV6HM?JgV!&&;c{vaZ%NqCE!Bc4PD>~j>cvz|py}}ceA{vh+l=|!2dVBJn7le3 z*5n>Zgk5Yr!8NcOB~58dg0C3V9_K|^buP)_EH0Aw+%1#~J?$!SVQ!I!t4O-E?(=jr z-U)g282$0SAzi=BcL0O5eONtYl7zQ_iEfd-1)?Gk4dXn%S{n7=$|XTT{_BH7z=43k zfc$^*f&MST(aqS!&6xRrg8nz@k^fiH#Q%4Yn1h?WshO#mo4L7}v$~s=s~M9k(8blO zCBw%TeWm5bLf$`wLUDnDAm0v+vVf9;BTg_#SZT}V&-Sy+``yvwpaps1YG0RgvtlZ()~szA`Sk>I0_L9`=mx8WD_zIlxs#|SBN%kQ6oz_C zSjwNI4%P1RdOtd4Q|v_Ebnt>z9WoK4QOsNXL9n3?Usdc_zO(rB%9hFbd%mMt;lr?d zsyMCKmNITK&8Ji@gjvjRmm(Brf@Hs;jlWAXi^LRK<=EAwupv=CwwrtVgG^oPd4iwQ zNRF93855DstrXWXqWfGYL#J z&s^7dZ+$5STYZPW&|brFiq6qFIT$DNwWLt_V_y&;O_Yu8KHeCLx)E>na)3`iMcf^5 zI<8bEVPbQ`cZ07rnUC(%6$yVK=!n4Kc3(Xpm`?(X`JLgROlbPR`PtHs_zQx+ zwCk4egrz8Zv&z7amNKGN)TwFbsc6S!;f^GR6>3xiX%1yps@$wGUB#yyEzO!YD2raM zB$s#0!3pZ4z$4yvbP1&7^`I_xlnE$$#9DufIHw2vSB0}nR;G^%HW(3)H+8II!QK}u z6(rAl>8O6Q)Rie^v&h=Y2Xj?FL4?XB!8&0l11v3Q5Of}vqBJyq7kx{oP)WyRk_Nx= z5io`io8rd!fpj725`e=hKca1nrYoDyjw+F5DRF+GFDY;uYrJQKAxz%?p{dl0W-F$1 zZl>zy@xwYy;Zy1otK-Ui#KQ^Y$uZ1=ZN!_6N;knvu+7vB%+bVYfpe5Kau1{B=RXFS z$SVk1^i=)8Rc=#l)TMbnFrHv>O7wRus)%K2 zdrpR79ubKY7Q|9J(P{*SbM`{t*$BO&e;j*^)Kn+*z2im+h~`qQqXm@qX$dvY*OXJ8L>w;b4y#NGPva3Ky;VH(=((G491?!=b5?QPiC@(?q6|e$G zIYu4=53(y!YHR9PhI5rq?1Jsl3n%My5~i}AFO@v%qSG-z);rf{DU;Y3|2$+8xvBEo zOv4LTVZfWb49qWi_^4+f45pkMDX%t)mVj7xwI?bRMAj$Eb=fdDS7Ss;BFlHb_lO7W zM+cOf5TfSSlXw>sY)$ZPW@=PA@2iH@wAiuhY$Zk%bXLvs6vy;LeSRYiTQ6*le35Fo zg-uxlD3BcAc?M4vV0>UhY5C>863DBOa@II5v;-V}N|3+CoI3 z6HKW?xcxhQXIkwn@eR>Z4?A{(0c#NPioO;+6l9QiByjb3W zakS%&pz4fv(2P-GE781+J%vutwY^OadBHDq9+ZP_Y~sK@$J#+jBISDWV96Rwiy0`> z+_))~7hxsHgI@#5kE2%kr-l&)vT3l;QaJ6xFW?a)Lc@;ug3-wcZ8%h}Zmo;!VIioOc-#&<*TRYb&3rm23XJTk6egp)>}<$_V^ zF9n3$c@y=ob5@71Ak&TuOT9A#4*Yb8Appt_M&}S870TKltlGR5o#l?wRIA@5=UjQ(Cb{LQzO~ zc&eVyJL-rqY|j-ysJOm2QDEZ6^_>|E=$r=CQo=bQx#IzxWbPl2`nWPVZ}Gn&un$kR zVJGn4l@+i&pAF`vjvo(~Ovq;|V!w}R2vEvwe4;Css58Ln8g@wNS<_KoF|AQ5x#x;! zT>`}+z1j;`F-SX`Clw}TI^A(22mAu`%U46A1h6RACCHU2ZEDywllT=mjf%o;>aowQ z5pQV_VtyO}203a{d&p=8mp{PTO6IlhJ^$jCy5|8_)er3KQ2|eJI zjge=EPFMage6(kO4DN*(Dz+}1O%h>=YW*KZeZ5@eLgO>z3BgjTd3)=y9k;Nj`qPU| z&Q9sJxmgK!QJXX?La(IgQ!aY5(Z6b!h>`0ep-gIqw4cRU3umtKe1kPhEzZC#><&V{ z$z%hh-kpB@%ghd+^R(T0qR%mV>A)SR`93ozUipdmgnM?A5veLgpR&t?3Ab>e`Rm~{ zJSc$M?=IXug>WFG^jH?&$fA_{Tw`m|x*8fQ(;A%;z~MDaAX}-^RCCv8z*`pp7-!R@ zSO*Z1nsHBVm71ux_79E9@<+NSpJb)}aZhIlVs9okIvAYdg}bjuyW-3K?cRjE`dnq& z%yJCbu=6-yM=V1qE1)cxY#85xA+80PN^XyN1}KF9wJprlTsm-FpAX^`o25_t3KXh| z*#VBIx%{D>@pHch!DbR*sg1n-!2~#IJ(BLhrfW73o~P;1H$;`$wV-vOAQ)xe+`}gw z!0$?Pv)(Cbg(2`dM7RaA6)qLIlkO)?MluIaWHUnP!(c#Jk?g=};fDfQ>CtsXjzMT| z4&l5~#vz)*NYQOym04q_4Ny!a^PDW3zFJ7i=d@ zE@6|V{L@&5QHTG|s(JN)Zp~OG4q1>b?%#3R97UtCu?$-epY{dG4YfsgbXT0U2hf2X zV5yFw3hoCmOviANs7VbG6V!C2hJ^DoIV;Bg02fPKkICsf`+Co^yhvWp`A~}=qe#n2FoRu8Us4r%VDFuGGOROy>G`~#SYDd-fNT@fY zsj3y)+D-ZUZj&}F*RbXyL+qH=gn2Ubnsa6&_k`?DT(3+E@ss?FwIiClAAz_xLJiLJ3 z-0HF3?f^G=g*kFSCwLut!_i9pt(ZDs2R6Dp^nC9HmQs}-=2+5P2XO2gu92E|G=&UC zM-Q0ZT=5OxS^nN1#ud0KS9emM=AQ1OU1@u?a|G83ugg$xrkv(FX2`E@|9a^}+p2q; ziRkz#SFikV`EI|SIh1z`7$I2s%D}6*o{7Lb&G-kNb+DJ<(D!xK+@*?!^u~sS*Bfq7 zXJZVx4c+l-I$H~VW#F9UBRq7@%f2E&0^khBD*F&~a()L+uul-T2~!wt&Hom3kwVZ` zjW5pmylZidp6#h?+5}C>+7UjoUjf|iugQ7)QBFR?kp{EW)oYz6WayDq($@`bDA?LW z*|B?;sR`+^n^*h9U+YcRiZ|b{`QDG#Y^PfEoGc?dIzsYjZ=U3jI*cNbO==81{)Isf zj&5uhykDD~Al~uE@LB$T_HTsUVD~|d&H+mz$la&T3IoM@U1OuDuNfmEcgIcBj*D#R zKyz{zS`nqjl8I0({u_e+6)7nM+Zge*gPq~$ADXRH^$J10-Da)NTSs}gCcLV|)zu4{7_=A*Z z>nSJgfd=FvG=5`QSt37ix%<vcLuK*pR3|}l$_aMNBl(DTD;RW6TS|gXCmjE- z;D?6?e&O^yZ2XUhBFp%@hy7F@VP>M#`~MsqDrR>X*1)@I?NRwwP{O#t&#q=)6n{f& z+|0Z*o9~TRFWPq!Vdj80nNObi95sK#ZR(?wBgRAd9Zx+&YcB|YIEv4%$A2)d4Riti z0W|)9kP5{D^NOLEXz4DCWm9&~>HoRTqrnQx5;5J~gZK~3=QPj8+meaaZJsO7hHy4` zylS1j*1lUeB*UDdcr0mYA)Xo;Pjz?5Pi$80kKZjV{I`RQD0dB>I5U+Jt}9_Zwfq=G zs1X{gsI<3Mc6Z~jTQf`cX#+gp@<*?f+o>yH@%39+vPPehnm|!CC|? z-_c4pR4mWpal8#J88)gU&jzyXHv&NepI#7;2h^w#Vtc^Gi2`4O(!MsFfIFzc}(dtiP35LxKuA^Zb3TXk0}aG{qu5 zp>}^^Plz|TH0hS`d{1a?*)7xtE5(#NE$L= zTp?N@vDc9i?Xfbm9C9)AFb2B|O-NRS6J+^sm$O;4qpUX9-!u63*oE2l&VLYBrH-tU zcDGiuccLpGc>bCRg(OP5tJlZW>;>lEHv7Nw z)Jhj)ap5Z-5dE(awy37+C$C4@+wTwnY$Y^X3Z=t^fA@}!lY9ekgi`aohvcs?A*vXFFhDA1AcXb(`&tCA9XOvuHryUHO`KL21 zfdzv!gtYp_FOKs#Ww(`3K3})%h-4T#Ue2=`J>^+8m?28ubnAY{}U9|d#WXM%7XQHvHEg~yvUK>_$?qn)?`eawwAVg zdH=Acx!3>H>fnz9!i?x}#j??-ST6jig4;&qXCgLQnvY&jUX(=qMM$_%aPS(CRS$Qu zaJ2n+7_Hh1n?tsD1=`O4kcgEt&<2}oSpb`p6U1cP#uOl-BqV@oAgZM<)P@qEGtGJ` zm&1LdNN02b^lK{@3CyA^BzM%>?fl1`jG+qZ?Pt>G-ZrqJv)eh%T^ln1&M6Px$`9GY z?hX(&yI%4Izs5&V8-8WLJwJfp|-;I`ZntiWpWIIuUgG*cqHjz_S#S+5(wB@fg z5zaMJ4;o|UPRrO)kBrLc7j?;F8&vou(-yU-yI+g4sMM!EiJQii#Z?p5DQ$ca${B2< zpvLhf_UsF?Zyn{O#$lBvQ$Y6IO{vf=|M0qv<4)z!90tt#_(yxgmK+|58qO0=CPb)U z(zXR?G&faXz!{1bJfFa3xpgWJ9Rs_AAOpHrH2}jAWut%fL_UqnxwKpyKEzpol?VyR^E>EnbZT@ZFybbL_krvdyTyFz!&&ch(go$r{ax5#1~S1 zX~UF>Zfctu<_Jq|DU?H!pfR65aZ@e5YsxxhiT#w}$(1?HCzIBNkR}>lbLUriFZW+n z)+OWVZd3~U5CCf|bUD29Ix< z7C~ZLDi=ESNnMM!LJrr)!^*=^R<`iF>mIP*#?Rjb{rQcbD2*@vh6{80q?`)L?Y1)9 zISnqZXpohM=-hc#ox_=H$cPz9&{adJs+&&uo>5at(?{^nv*Zb*>%CiQJBU*_4>L!=Y+GVC?Tc*K<~ZsT}v6$$#z)9h0x!SGgi zC>M_*0sD7q!RrTA2)=+jxl2`6%x7<7yT3%pX&N(V(#G4>(LM`s+qTiX3kW?Nf%NyE zhfC7|Z?65a{b|*{`)qTd+hba=Q%qfQp^)ZBkcpu=LF7QG>wJH-V=OxYw6!(RHd}U* z&QtjG|NgrU_nOs~0P&u@GshIQBK{{!s{ZVlOuEQwgdv)G6~20$he2;R zPW3?@x~dBV2-qR{4_vNn8bXsQ4}wY`n{-)rt`0}&(%B9?$3!T_pYrrh#Hu7_&N(DU zUHFB`HnN=VVIA@IaE zd`UO-Hqf=WDur&pZ^H5qZZfXQZKU7)XjdA>Nj9Q_3}2*;Cv`-H4(uU^_!=V`VbiMe z42yE|t9bX49XSOX3I~F8G6W;!1(!tZ+~_rJgwwainYu8;(%TmjZ#>%mFsVhY=wy~< z1EBkKnpT6bdid}J4cJ-@U7SPxXmkWt`RkRDghP8@(iK(u;%oOuZ}Fp1Hk|63juqSc z4dH#fY0;Tp{oy#a?dcY@627pgG%xDclQRmtWhPFcOy zRe1x?7=&DP3E>%93{i@nO~I7x5X9-&Ks&v?v1bqIc=8FSCIGpwv~(FTF+T2%?%EJK z914RsIc$N&A`W3jxV<886$t6^YWp~4$6B*>SWl%UcIuIu;is5=`4x^DO(61XUM;XE zn<%koXnV|8shIf=>1iKm|D8*E_F>b>&kFbWPdM!C0_xq?*Q%<92TNB9_3DumR>O;{ zRAoXN+9a_BL2U)%QVQzqGt(6AR=zxrp`BzuO@(2l zY{**tip=5SH*rk=PCN9EQ=$a#e|Rh`V{30MM%(eC8xgKAd`AsN!L zik1zYkqtca&9Be_G$e1^R;{d5>bxO>%Se|#Qtab=x{-du#9#^=Qz62W{O}W_(Ru<7 zV(mhOWx9cBa$HunY|T(Tt6nL)$f4qzw$_&5}TpvWuLu!sNwiiIZz^ zauMw7|7Mf_GCu|gBi+KM#(LLne+_Z&8rjoM#bdGKJ3D|to5fG{i?UaKsA&}is|b6z5I=>raN&Cu z+>N_itkW&12EGdeipVsADm>s9XBbf`0`{#Now%fa@DjggyxzId zp$pjfeKl60Eea!2CiLVsNsw+n!SM2(tRR+$P^be>y?|}dVAd^a<9nQ7-Sk-D9S`CL z>5Go}$%y^*{t2tp3eEIf%7VGa^!vMPbX`H1&c)Fk*2BDo>s?4!BktP7(?K6j`=S_n z2{S?dc925=mM(7db}Wm2)O{mNw-)Pm1|DEjiJ_%JrJE9M_6=6!E^^Hf_7Kr5SA*JE z(3{7Bg(y{HADOwR7L$KM zAME?w26P(njnaH@0A)IM7@vM!j2={P)XIDKONt#6L{YvoLfoLFpspfno2`lWva%4& zHGXo%3K>|rwY$D$qzEGcVvLodlOc&H>$dG5CPvPUfq}RbSU@OydBApfH-QIW;KTjH zSdwewaTJ6S=K_=;yXqVcy;gl9qGraL%l#!2YA4e5PtMLA6Tv&DWhE3$BZquO)kj)e zopW-JqR&cg*($6BH79Y?XW$$xRHR2hKfNurvOVJdN>m>NWRz{U>%|y)D4!k&qGaV| zSzLelWqFPRbcq9}`kJ8`kZlK(;(V)xrERFJxk+AIzgYP6GQfpYIb^`Dx2kyQ=927R zM?EmH+z`F~u&w*&W!X7ObOmfm^K(Q^;`5E*?qm|7WBR6o@UGUgAL$sTG*T1|~7P!cmfVdE)t;ni2YbYn=Ml$J&! zArSmM@BiyBsR{|CZY_pPz7k6^giN5Cr4_AUJ|lOi*xB%%C~psUXVFr1bK?;QYgpd> zq;63gyTf+Z%eTY!-^s$>5dpK$ClqJZbKtM?g)B190NM>PAgfX~S zpIe7VRom1JqAinN&zWlpr7?J9-`L^$!T_UTZ<K;7-$(d153nj%K2+} zbQ)U)-eO&_rmXc~)c_mS=B`ck9!4BKar!r~^)jsw7<#=zIg-z5A7?d=6H6i3<4YSxV(qeN+)i;{ZeW&sm0cB`(jr}TMttFG#)mHl25=O2j$Zj?{ z3DAR0K%5v7mWFm#)rXd(Bw400>iXASTs751ZWM%ONpML_h>55q2wQ6_E!`r2O4H?T zbU0u#I86*uHl&F@i=0@SSRF2Q-5I2 zeAIRk0xpLr%=O?J`K~}fFB+3I0@O@|`!d{8CK)M#v|VLbm4_=`3-wdA;&%S_#S1qp zx04vEf@7a&dc;AdMBn+<sDNvCwfD3Ayg~<^u zaIX@slJGQ|69OqS9;nwjAl=X*SnT=s7DX1~oRFvTqjVD^wKxz1_heEkrTMc=&3;S4 z9^pBo5+ZeU3J3LMr-oXa5^Sd*+VkIA7d4Q&_mG_BcXP8v0+r{mLi%u5Ay&6>E4=x* zwp1O?K$md^cRl$s+JU*DQ*Z_ORPfsr>N!j``VtO|RQsjA99CnC8b!I^$YSd0wVi73 zF9PmB#nFhSd>(3sMdKwhjbUS4iqi0fc(2}VfP;mEm01|R1>OSsr5jlGSct6zXI3hU`q^pBB}-+%LxZV9eLvklC`F#GB{uD~)u6A1zxFWH z)eEj&M4^_cr0}66I=@^)W^p~|fhub@5f#FXDzagw7>OuOIgU|S;>wZ%yoylx4{7Ul z<(WfLnxj%}DIp9KmNyYC7PnJa|Je2B*fDW0r`Tt{=Pv*jDj7(@%})$ml|trs4+Amp zy*y9&^?M8TPJG4(2)K>7T)J_APIMQ|czz2$H#Q;uY^l@vbeO?H(k~)XlcV~RP|C>G zx!aMsJ(5t+53Wa5Ho;NNWG2c|hJPR}v7nk3R^@#7sx9hJ)Z$C3@@9g9P;Jaq9bR-vW_*Q` z2H7T>9RwfdG{~fBW(=$(3_y{oH$3B>u!y$~gy7YZDR`ak^A5h=w$Lm~bw~c2U$9m4 zn#})wsz1FK$)&%LXmh)7C5=x?V|OkSSe={%6)dMG$Q-QjcauuCQ3p;!MNmM+8kAdI zk}a;bnU0y^O+G;C`sw-AsMh7#h1Nd1(TGw70PjqJWt>CqG!w7mx(7^`bOPbUsI=A_ zyPasJnAy^OYMc32*uRc{%kMga4b!%&u2NbIZZ{y-Qn0kP7nJ#7r#vhnwZqkts^l9i z#@nGSL$=ZHY0iP5k3+4XITo@4pz*;_&Dc9Xg$O9oTcLQmmtl8;F9;n8H-C3O2-*Sj z1Vo3nASjj|z05ZdXtos{o4|XJ+fZ;Q>?yF;cEcf{@VT&Cx=%M6_W>F_gyoSF{#xbG?9G7JCcHKL52SdpPX zd0H(*wB!1VG zlOqQLE!e{2)#mGr7Wp7en)^GifhkwP(yGxRck`Zh`Q8fbSw@9FqE4xmcg)qZ6i_$B4A%znwQafj-a<%V)|C!t8^6T;F?|+rwC95smjN)I~Yy@a@ zoT6UJ@aknw97VfZ=@<~3RL*Lugq`$4G~I)lTVt+ISVOYiQv^u8i{L<(q>HMe8@a(P zoI6qP9=YIZv=RAyGWV*8vn`$29!E4f*cc`2z_Cxvf zE)x*p+7+jLgiU!&{7-!H1V9CP6Vr`3WLd(CPNOkJrOR0O+AZiOF9+1RKyeA!hWF@KK}*wgzK_Sg^g9j@0GmMgVIr!MC*iiy!T9JM zM*A)zZdm=$G7&rq1Yg{nBGt%WBkNxZEm)FAm-SYGYhRHzWI_S$V^SliIA-49MF3H^ z%0gSVTm{F#1=$fa?ge?)b2GL~?e5Up@**=k)%K*mb)moN4lkQ2EHlj2hcgOfVu0t|P{ZO&NYGPtxJ`HPrePWWh2OQZ~M@)4O&7hOQ^gPS;rt+Aym z_;x4Jqvgy_{vguf8cQHU40RW18}t4$md$l4iS;aYx;gbyI|NLF-rq4O1j)JI>8^a^ z3aPP%(&oIfloiIQ8HyrY@{nR*F5=Sxd(DVEpbvX{#*xhMd0*A(}A{ggRPW?oNyttDIo$}_oD z+-fqGxUx@Q<3%;Uw*A2jj*zM#qp_=oq$HT%IoV=;BXo6YW%68k5J8MIyzt8-7f?j_ zYljq7EyXRECvcKs!1U2SEqE1#vO>7(#1u6vGX4~9V=UQ;T$XD<>OZYxEJ0UQ)t^zi z4FNa)@|gZMY;Zz+a7vg>WfmB|t{B;Lvom8fuctJtJp4nL0_eRF{lQlVcDq`nj)MJl zVhT(`6{K6yNGAy{Ym(s)js^W`CP3;pJUz#5>7F?BDR+H$R}9VeF(qo=tV>h&0ZATm z-{`*l)kc%x*6Z}lX>#+1^VNs8aSu}**>CDIKBdrD-!udD$r;kL1bHVbVOjN_PUlp; zEUJUFfw80ZW^Th2`Hp7gTGu|HGiIP{yHEthRD!!gE(r&&s0-KcwQP$fL784()b^r6qX3xbNt8;O* zT9DW7X?^S;E)5YO%`!~6>mT^$;?r;?v^loY4HSg7$%GlU(i%J~5%6 zR&@xWF|MUj9mul6sXz?!FOdUp5ZRYWmdp}e>8p{*sjbx?#Z3FDCFgy7i6dNeL=<$~ zCc%CQ%>Q*CB>N?#ACMYiBERv`aoLsnB~t#ciO7tIehJO}%Dy71OCoA7g~{c-zT$lg zwisY!YwiLd>#$iFme(#Na{=&@zVJ0R$Jac7(*^{5cps`w=FF$lq4imy2FJEj=hj3+LUXilb%tn{3ci${TQTS|{VJ7erIu(C|NfBUnfRsM!PV8 zF_G;go0N(p#ac}LjDpw__TSH*?R_EwIs$C$C`G|rh{##;Bh#Crlh##p(hLm71p*|1 z=!L2zc5AFq)i?0pH5`Aa_P2V*UDo##{sHA**yQNqel`sDc^^Dksd*J#D)LAtDaQWi}mJ|#tGLQPzom$nwfNNB>k5*YRDW|2wa(gK}H)AWJ+a8GC&{&_rnkOGL5aj zVq*1X(x^9CR@H69{}>1ZDV;+0K~_qC0&j*mb-pFMF;eorT?TfRd>K0Yg_S!psySqz&` z2MT#I2ivlEA)#3{OioI-I8irYkxm?woN-PjD{+xYoRd5!KXF>*&#y2%Dcw3Z3<bYlbS$_nq=PRshAe1+!nR})FMZ?Hi!Z?B@=n{^oz+= zP)S6+XsZNP9`jA(WYAsG9%%KWYBcR?PcaXYdO5UWG;|18W(Je63rzgr4M2!*e*$x3Q9-ca*-^Anxi5^?M30E6K}7dA zQ?cOY%5HJ^=^&DKf{<7rE-V)92DuzbwGcMIaC7#-#2>a-HzHL?A>dr3pV+0}Ov#&i zYY^p*U1BI{;K0|WW%O;CaA_q%V+f}RMR~5wEL}W6v~MJ+wXJ095R&_x*>ld~w2mM_ zF~IIt!9BbQE-Em5hfZCKlR=TL3Yx5b=&@fmpvr56c$Pgk4amfv!t$F=^Cvg3{0DCV z757QlkK}yEGx0!D@ux2T`&Y`|E5z4I`FkBf#Csl)k8JU0Uq~V;q`eCALQ^h_t6^?| zF8i^repFQVbfX;mn~Rbwuf#J)tvl&EwrA43y|?L@i2AE+`STZqzV0vi?!?sR&;k_} z)1sS(m%Z)F{F?>jKZZ;dlg34tn4XuE5=!r^8CiQpwFbIPJ@{ffDr%60l>s?FN0{Bc(*+L8{HHZxC4l#nrasbZd_OcVUBa zG>u~y4+o~jy&cS4U1t3<{2x!#x@F0XyacOM&@Lxf0_S%yI9-mV??XIX8pMi9zWwH# zmf)y%?2z%4j`|J&@gpR179p=n#p<$iZ_n0b>R1vZ&0n%9`-qXSjH-KYYJuSRScNv_ zbuOxE-9ua{g*o$C7qoXE&T9M5g+* z;)LYg)rr;#90pIR59zs!ymf55sDqgR#$BiI-=%(u$6T zQQ&?v*e}(t0n}u^`M8bZ35r&AkZgx4>WYx~sI%aR8hEmrbafH?gROf%&@|2*Lc;jNp#^TIg1lPb#?)rh6=`@=pq3QlpnAqmRkua-%GRFU$ zj(n@)SgL*s^^-{_hJoPp5V0U$#1KuBGJWWn4igc9UHZ`Z9A1bkHR4z-WVml|sqIKd zY60$fU7WuuR%P9(`@PW;Kw~iUedmCMeA)A$SmFB94>~{y*HC_tdN5;^HxvArS4gBa zgZ=Qez_H-yUt-q+VbsdkJy~VW|B`+fnQ|+Do`%W&W=bcrps4CbQk6%;2qKh%R|vOM zC`sv(*SRtZ+wQ;X3Km@rY{Bm(dwA!0&^PM+>qkIzP@feEb|6|BgI%1^6)4*?pmC4( ztiG)M7mc#+TJt`ImQQ<q=8&IY-5~zU|u88|9XtZSPHPsMTT8(uZ zIPuCQU@tv_{5I*%cBae3fzsS%DL<5yRRK z2Qi3%J4b(T)A!>zqM6-r0NP%x)8i_v?-b7pmnasPMIN~EW1K+OSI-s|-bBx;r23it z1ERlVJDA>D)v)LcVt*M#oUHq1f%VgDmG`k5H_y~AFVJD+pdS&L5E~E<_Ce@~d!G*y zX3~xum0u?LmHiK0`gr3`)E3y4Eo)gXbx4d#728kuWF#CT?b#IU1mp) zN|+d1kzLu@`V%-WrwBm>+lT67oE8`uHEqISoq-74gAmwnBMYYx-vs5Qx%M-6?5-Hx z5}<4S0qpHq8NCrufwuG$-wJy8yUj&hVE1Z2_rK*_S^if zJjtd^O+PkNs>j5D70AAYs|>vJ)I34W`%iB#$k{qz({*=o3+2#-UU9jpP|7>`b<6dR zC+7^Z%{ezCobKN&N2QAgPlR&oQERcq_M8(43e42sErd7%Eb2Q&-O=*8~5MtgTC+Qx&{kU{N1UABuo@+ zkg(<-*3GBwjw_k9jSTCfO$(Wed-=zmMA{PF|I1IJT=B znST@5J+sFo1+k5o#bskfL)TS?Y5CF=sM*sSS8paE}i#?};PaBt_2mTah|*c#?&8EYQ=BSnxm?f;KW#Tqd;UmrJEO_$pM zY3TISQvBHHJ^+q;9?|b#8seUKV9Uxz6p6=)%0@U=&52n3*VSU^)9qD2glbWN8cF(U zbxGGEoZvojC31Y#)j5_p&pG3%o#QjsOANsSpx&6&62=_3 zU^F5}KP#?P7*LDd<=jd%g>`0#dzK_@%dyHf@vosp`P=BQA@*T4H}}7x*FmK1*C}!< zUn~b-uC;Oey2B@`Vd^Ak!he6vyqbnPx}vk|=)O?8wE9M*dJ=JdV+4Lt+$*|5VD&LY zywO>SA802&C(migH-E!kEiP;{e((!}O6!Ulr zlwEUy#2MfocVsY$vi zZa;)JY<5BjQ?~{4w<g0n>}o?(?q5_4YRSneg|X?T7lGuj;O3a8y_`O4`PgIk+f4k z{4&@t=n@Hh(|;i6yFf^EdF+7X#Az%<8jvYKbSyw5i~QFDFX4m6!S_`L?I{bS9s(yy40k`1ifg?X)D^cX<%r#$_v|))E zaaOv8NFngcnEY8d9JcR|&;7;B9BXy@VO5z5uNVgpzBXMR{D~0<4|Y+%my9rRQU3}H zo-bihKQ}9${_nY6F&-Ff*ty+&BE;wOxm_heyv5VGU8{S9c&M37so)c|qJAAueOo+4 zUvG|AqIjr_e|nePb~KyNC^SO;kA>9r|;AbpLIgRVe|+M zdpp^qK2!p>R`;HH53eH9UhH*RSn*p@$=nue^CIs|1zBu_XT;}B&jNhJ@-kn(pi_z4 zir>NXAGF;@AL@h~dGIVP*KyQ!$&Wm+Cqa+294z&S3)H{tC_ORjhs~KT2cEd6?;Qy_ zNZM2T5qX|2vDjACy`cCpmaFoOx39}I#KC;(q}YIwO&> zwJj;~I$-raIMYuMrL4_PM!Q6<`C`RAdy2Tk=tHi zJ!v(`JPF5#C%UYv%|wuXa@Siy>ujQ@xT^UKW?r*Fb+3Gjr8W&?sojxTq9q1xiFNXi z!}z>LBmBs*9)qqS`f<`5SCu-NI4o9di?WW`Ad=Pw$q%7orz^ek13kyZB%Gdt2PgB&Pf}GQ3O5d zLi;@nQM?F}KvX>s0{yO+U@cF`XRA&;J&;i9fhR6Jt^MB9$O%z7@>DA%5V;IBuFBFo z-**^fqNeMPtNRXtNzOjf1_FCHWU^}iNWUY;WeqrDX4;`YZK{v**Wzr<50Y^#fq%3; zl~bW80&WnK27v%7J37LO0}4-zh|3+(!sQEwYb74zf+^)Vj_IoKWD~+hUb^ZtK_}Ks zE*Gik4EDCyaNqZmE}> z3KHVIceR~kyIsQt4)dlYvkd3XUpC%z#sYq$Cc?Qj>Fk%T6FgY<|*J{?Ii+Hh{R-1Tn&o&ETI<$s{(ZuEPBrma~opd zmff2+m1g=eeqr*KVPii>ocSc7cL3vFi&E^HQ{dYK?(N|pI*iRZ#Qr+YzC#4*cPY+( zfF;N~VW~&}YT3_2vJiQG>#jsHmQZ~jPp`{9xOu%?W01&;;HOMbIkd>g|h>Um?V*`vIbP$d3R z5uJv&ofIwupS4L$#n_oowlmP9`%q)>m;?Su z%&AlT=ztYt%hS=tKwyxY`bGK(Ehw_%Fx_@pMogs7mBL-GeDu^Le^Zo4P?-16`H=tq zay2f%b#CUa*^JzTOt$NfBNg+9l;dU6rNN(vT|>dlDT|A!1JI(;lhbE+e0{SbV$^5C zAM}CR^t}wuguxk0>P44y=h>OpC-$c(*8VYZsKN0e$de?}%K221q~1n>>j! zUW{B*znpgI{AE2TRwakK*0BBfa7w2h9@#kEw|f;=aV?np0$JOhe-FmPb&{03Y!Q;o zb%}cOU8PYn2j-AW)*o3$C$&w)M!1a(a)q;0Dk`4=91J`dI9Y;S5?+=}o`P(IAa4R! z^u#AO09SPqRl(_bi*H3b@+iIHvTx{?xmDwVb<9KI%~51hmBxh0QK)xy7q;e@Mgx`q z_yFH1g>Kw=>Z>=ryl;OfWzYV=AK*A8CDY%b7Kc%Deidl@W@gU>xu_jJYo z^cL6v^}$Gipkt9>>tCe_oC2wQ#e5aH$VN1IjJA|p^J>3+H2ic*sN6!fUVW)TlW~I3oskcLHWDBLF`AOh2b#b(chW6T%)beI4p4KE%6JfnAo$GyGA4 z$~IH-4C=%V$^`c4#E{%&d51PHHy{j5O-9P;okC5F>h9jy8BC{o#Q}Ximnj(>EH7dD zM9i7>uS%UuxXKSxL6wV4uRrtRPmzkhEKM&}@(zY)5BcpU&vX(TUx8A= zCECQ`zVGQ64>6zWt~%3V9ohtY$h~dgj6mQFT;TuNc}=I<^#Nfmn}CLK*ld4cv$kY< z!1Jnwe`**0VP5$5?Z$0iIDZxYYKg0J)Yvbk@BKJAHEHUyqm^EVZm1@{nPgDrDZ9Ib zNzE=<(7ISa{j`9v^o9SoFI;ziasBxM{`j7Mv%2?1Hr3r|vlDmtze}R_rQVm>dS520 zUzGpXv8|r>?H|6MGj6S%yz7O5)g-?qPq%cPTd?`1&90_-PJ2TZ+TC*8ej)Zt)vwxL z%vIKXA6b9RGKrg|7V2rd#oZ+4I?^l=Qj!IU`a&!IX2QBt2E+jk$VVN34ro9HxQ=T8nZm_@`&a~oDHniD_#p`ht3k&k zAg@OO9g~0nUx7@hCR~RmAo~Mk7_uKx4O<4};x-Jtpe4YYl?`M&3lNF|&wQBg2;u<% DCn7!A 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/pubsub/Client.scala b/src/main/scala/pubsub/Client.scala new file mode 100644 index 0000000..dec6784 --- /dev/null +++ b/src/main/scala/pubsub/Client.scala @@ -0,0 +1,34 @@ +package pubsub + +import java.net.Socket +import java.nio.charset.Charset + +case class Client(socket: Socket, id: Int) + (implicit charset: Charset = Charset.forName("UTF-8")) { + private var name_ = "client_" + id + val outStream = socket.getOutputStream() + + def name = name_ + def name_=(newName: String) = name_ = newName + + def isConnected: Boolean = socket.isConnected() + + def close(): Unit = socket.close() + + def send(message: String): Unit = { + val payload = message.getBytes(charset) + outStream.write(payload) + outStream.flush() + } + + def sendAck(ackType: String, message: String): Unit = + send(s"${ackType}_ack $message\n") + + def sayHello(): Unit = sendAck("connection", name_) + def sayGoodbye(): Unit = send(s"Bye Bye dear $name_!\n") + + def invalidPreviousCommand(): Unit = send("! previous command was invalid\n") + + def sendMessage(sender: String, topic: String, message: String): Unit = + send(s"$sender@$topic $message\n") +} diff --git a/src/main/scala/pubsub/Server.scala b/src/main/scala/pubsub/Server.scala new file mode 100644 index 0000000..eb04961 --- /dev/null +++ b/src/main/scala/pubsub/Server.scala @@ -0,0 +1,51 @@ +package pubsub + +import java.net.ServerSocket +import java.net.Socket +import java.io.BufferedReader +import java.io.InputStreamReader +import java.net.URL +import java.util.concurrent.Executors + +import scala.concurrent.JavaConversions._ +import scala.concurrent.{ExecutionContext, Future} +import scala.concurrent.ExecutionContext.Implicits.global +import pubsub.collection._ +import pubsub.command._ +import pubsub.network.TCPReader + +object Server extends App { + val port = 7676 + val maxWorkers = 12 + val bufferSize = 20 + val socket = new ServerSocket(port) + try { + val whatismyip = new URL("http://checkip.amazonaws.com") + val in = new BufferedReader(new InputStreamReader(whatismyip.openStream())); + val serverIP = in.readLine() + println(s"Connect to $serverIP (or `localhost`), port $port with `telnet` to join this server") + } catch { + case e: Exception => + println("There is a problem with your internet connection, you can only access it via localhost") + } + + val buffer = new BoundedBuffer[Command](20) + val commandHandlers = for{ + i <- 0 until maxWorkers + } yield { + Future { + new CommandHandler(buffer).handle() + } + } + val threadPool = Executors.newFixedThreadPool(maxWorkers) + + var clientId = 0 + while(true) { + val client = socket.accept(); + val cid = clientId + clientId += 1 + Future{ + new TCPReader(clientId, client, buffer).read() + }(ExecutionContext.fromExecutorService(threadPool)) + } +} diff --git a/src/main/scala/pubsub/collection/AbstractBoundedBuffer.scala b/src/main/scala/pubsub/collection/AbstractBoundedBuffer.scala new file mode 100644 index 0000000..64c077e --- /dev/null +++ b/src/main/scala/pubsub/collection/AbstractBoundedBuffer.scala @@ -0,0 +1,34 @@ +package pubsub.collection + +import instrumentation.Monitor + +trait InternalBuffer[T] { + def update(index: Int, elem: T): Unit + def apply(index: Int): T + def delete(index: Int): Unit + val size: Int +} + + +abstract class AbstractBoundedBuffer[T](bufferSize: Int) extends Monitor { + require(bufferSize > 0) + + def put(element: T): Unit + def take(): T + + val buffer: InternalBuffer[T] = new InternalBuffer[T] { + private val buffer: Array[Option[T]] = new Array(bufferSize) + def update(index: Int, elem: T): Unit = buffer(index) = Some(elem) + def apply(index: Int): T = buffer(index).get + def delete(index: Int): Unit = buffer(index) = None + val size = bufferSize + } + + def head: Int = _head + def head_=(e: Int): Unit = _head = e + def count: Int = _count + def count_=(e: Int): Unit = _count = e + + private var _head = 0; + private var _count = 0; +} diff --git a/src/main/scala/pubsub/collection/BoundedBuffer.scala b/src/main/scala/pubsub/collection/BoundedBuffer.scala new file mode 100644 index 0000000..26b69be --- /dev/null +++ b/src/main/scala/pubsub/collection/BoundedBuffer.scala @@ -0,0 +1,26 @@ +package pubsub.collection + +class BoundedBuffer[T](size: Int) extends AbstractBoundedBuffer[T](size) { + + // You have at your disposition the following two variables: + // - count : Int + // - head : Int + // In addition, you have access to an array-like internal buffer: + // - buffer + // You can access elements of this buffer using: + // - buffer(i) + // Similarly, you can set elements using: + // - buffer(i) = e + // + // You do not need to create those variables yourself! + // They are inherited from the AbstractBoundedBuffer class. + + override def put(e: T): Unit = ??? + + override def take(): T = ??? + + // You may want to add methods to: + // - check whether the buffer is empty + // - check whether the buffer is full + // - get the index of tail +} diff --git a/src/main/scala/pubsub/collection/ConcurrentMultiMap.scala b/src/main/scala/pubsub/collection/ConcurrentMultiMap.scala new file mode 100644 index 0000000..f23d50a --- /dev/null +++ b/src/main/scala/pubsub/collection/ConcurrentMultiMap.scala @@ -0,0 +1,65 @@ +package pubsub.collection + +import scala.collection.mutable +import java.util.concurrent.locks.ReentrantReadWriteLock + +class ConcurrentMultiMap[K,V] { + + private val lockRW = new ReentrantReadWriteLock() + val map = mutable.HashMap[K, Set[V]]() + + def lock(): Unit = lockRW.writeLock().lock() + + def unlock(): Unit = lockRW.writeLock().unlock() + + def add(key: K, value: V): Unit = { + try { + lockRW.writeLock().lock() + map.get(key) match { + case Some(set) => + if (! set.contains(value)) { + map += ((key, set + value)) + } + case None => + map += ((key, Set(value))) + } + } finally { + lockRW.writeLock().unlock() + } + } + + def get(key: K): Option[Set[V]] = { + try { + lockRW.readLock().lock() + val v = map.get(key) + v + } finally { + lockRW.readLock().unlock() + } + } + + def remove(key: K, value: V): Unit = { + try { + lockRW.writeLock().lock() + map.get(key) match { + case Some(set) => + if (set.contains(value)) { + map += ((key, set - value)) + } + case None => + } + } finally { + lockRW.writeLock().unlock() + } + } + + def removeValueFromAll(value: V): Unit = { + try { + lockRW.writeLock().lock() + map.keys.foreach(remove(_, value)) + } finally { + lockRW.writeLock().unlock() + } + } + +} diff --git a/src/main/scala/pubsub/command/CommandHandler.scala b/src/main/scala/pubsub/command/CommandHandler.scala new file mode 100644 index 0000000..c6b3afb --- /dev/null +++ b/src/main/scala/pubsub/command/CommandHandler.scala @@ -0,0 +1,43 @@ +package pubsub.command + +import pubsub.Client +import pubsub.collection._ + +class CommandHandler(buffer: BoundedBuffer[Command]) { + import CommandHandler._ + + def handle(): Unit = { + val command = buffer.take() + + command match { + case Subscribe(topic, client) => + multiMap.add(topic, client) + client.sendAck("subscribe", topic) + + case Unsubscribe(topic, client) => + multiMap.remove(topic, client) + client.sendAck("unsubscribe", topic) + + case Publish(topic, message, sender) => + for { + subscribers <- multiMap.get(topic) + client <- subscribers + } client.sendMessage(sender.name, topic, message) + + case EndOfClient(client) => + multiMap.removeValueFromAll(client) + + case Rename(newName,client) => + client.name = newName + client.sendAck("rename", newName) + + case _ => + // nothing should happen + } + } +} + + +object CommandHandler { + val multiMap = new ConcurrentMultiMap[String, Client]() +} diff --git a/src/main/scala/pubsub/command/CommandReader.scala b/src/main/scala/pubsub/command/CommandReader.scala new file mode 100644 index 0000000..580715d --- /dev/null +++ b/src/main/scala/pubsub/command/CommandReader.scala @@ -0,0 +1,45 @@ +package pubsub.command + +import java.io.BufferedReader +import java.io.InputStreamReader +import java.io.InputStream + +import pubsub.Client + +class CommandReader(inStream: InputStream, client: Client) { + val inputBuffer = new BufferedReader(new InputStreamReader(inStream)) + + def fetchCommand(): Command = { + val line = inputBuffer.readLine() + + if (line == null || line.startsWith("leave")) { + EndOfClient(client) + } + else { + val quoteIndex = line.indexOf('\'') + val hasPayload = quoteIndex != -1 + val parts = + if(!hasPayload) { + line.split(" ").toList + } else { + val (command, payload) = line.splitAt(quoteIndex) + command.split(" ").toList :+ payload + } + + parts match { + case "subscribe" :: topic :: Nil => Subscribe(topic, client) + case "unsubscribe" :: topic :: Nil => Unsubscribe(topic, client) + case "rename" :: newName :: Nil => Rename(newName, client) + + case "publish" :: topic :: msg :: Nil if hasPayload && msg != "\'" => + var message = msg + while(!message.endsWith("\'")) { + message += "\n" + inputBuffer.readLine() + } + Publish(topic, message, client) + + case _ => MalformedCommand(client) + } + } + } +} diff --git a/src/main/scala/pubsub/command/CommandType.scala b/src/main/scala/pubsub/command/CommandType.scala new file mode 100644 index 0000000..cfd35c6 --- /dev/null +++ b/src/main/scala/pubsub/command/CommandType.scala @@ -0,0 +1,33 @@ +package pubsub.command + +import pubsub.Client + +sealed trait Topic { + val topic: String +} + +sealed abstract class Command(from: Client) { + def toString(): String +} +case class EndOfClient(from: Client) extends Command(from) { + override def toString(): String = s"${from.name}: End of client" +} +case class MalformedCommand(from: Client) extends Command(from) { + override def toString(): String = s"${from.name}: Invalid Command" +} +case class Subscribe(topic: String, + from: Client) extends Command(from) with Topic { + override def toString(): String = s"${from.name}: Subscribe @ $topic" +} +case class Unsubscribe(topic: String, + from: Client) extends Command(from) with Topic { + override def toString(): String = s"${from.name}: Unsubscribe @ $topic" +} +case class Publish(topic: String, message: String, + from: Client) extends Command(from) with Topic { + override def toString(): String = s"${from.name}: Publish @ $topic -> $message" +} + +case class Rename(newName: String, from: Client) extends Command(from) { + override def toString(): String = s"${from.name}: Renamed to $newName" +} diff --git a/src/main/scala/pubsub/instrumentation/MockedMonitor.scala b/src/main/scala/pubsub/instrumentation/MockedMonitor.scala new file mode 100644 index 0000000..a93723c --- /dev/null +++ b/src/main/scala/pubsub/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/pubsub/instrumentation/Monitor.scala b/src/main/scala/pubsub/instrumentation/Monitor.scala new file mode 100644 index 0000000..82c6cf2 --- /dev/null +++ b/src/main/scala/pubsub/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/pubsub/instrumentation/SchedulableInternalBuffer.scala b/src/main/scala/pubsub/instrumentation/SchedulableInternalBuffer.scala new file mode 100644 index 0000000..4c12d7e --- /dev/null +++ b/src/main/scala/pubsub/instrumentation/SchedulableInternalBuffer.scala @@ -0,0 +1,63 @@ +package instrumentation + +import java.util.concurrent._; +import scala.concurrent.duration._ +import scala.collection.mutable._ +import Stats._ + +import java.util.concurrent.atomic.AtomicInteger + +import pubsub.collection._ + +class SchedulableInternalBuffer[T](val size: Int, scheduler: Scheduler) extends InternalBuffer[T] { + private val buffer = new Array[Option[T]](size) + private val threadBuffer = new Array[Option[Int]](size) // Who last wrote in the array. + + def update(index: Int, elem: T): Unit = { + scheduler.exec { + buffer(index) = Some(elem) + threadBuffer(index) = Some(scheduler.threadId) + }(s"Write buffer($index) = $elem") + } + + def apply(index: Int): T = scheduler.exec { + buffer(index).fold { + threadBuffer(index).fold { + throw new Exception(s"buffer($index) was never set ! ") + } { tid => + throw new Exception(s"buffer($index) was deleted by thread $tid ! ") + } + }(identity) + }(s"Read buffer($index)") + + def delete(index: Int): Unit = { + scheduler.exec { + buffer(index) = None + threadBuffer(index) = Some(scheduler.threadId) + }(s"Delete buffer($index)") + } +} + +trait MockedInternals[T] { self: SchedulableBoundedBuffer[T] => + override val buffer = new SchedulableInternalBuffer[T](self.size, self.scheduler) + + var h: Int = 0 + var c: Int = 0 + + override def head_=(i: Int) = scheduler.exec { + h = i + }(s"Write head = $i") + override def head: Int = scheduler.exec { h }(s"Read head -> $h") + + override def count_=(i: Int) = scheduler.exec { + c = i + }(s"Write count = $i") + + override def count: Int = scheduler.exec { c }(s"Read count -> $c") +} + +class SchedulableBoundedBuffer[T](val size: Int, val scheduler: Scheduler) + extends BoundedBuffer[T](size) with MockedMonitor with MockedInternals[T] { + +} + diff --git a/src/main/scala/pubsub/instrumentation/Scheduler.scala b/src/main/scala/pubsub/instrumentation/Scheduler.scala new file mode 100644 index 0000000..67d4aea --- /dev/null +++ b/src/main/scala/pubsub/instrumentation/Scheduler.scala @@ -0,0 +1,305 @@ +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/pubsub/instrumentation/Stats.scala b/src/main/scala/pubsub/instrumentation/Stats.scala new file mode 100644 index 0000000..ab221bf --- /dev/null +++ b/src/main/scala/pubsub/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/pubsub/instrumentation/TestHelper.scala b/src/main/scala/pubsub/instrumentation/TestHelper.scala new file mode 100644 index 0000000..57b7215 --- /dev/null +++ b/src/main/scala/pubsub/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/pubsub/instrumentation/TestUtils.scala b/src/main/scala/pubsub/instrumentation/TestUtils.scala new file mode 100644 index 0000000..3ab651a --- /dev/null +++ b/src/main/scala/pubsub/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/main/scala/pubsub/network/TCPReader.scala b/src/main/scala/pubsub/network/TCPReader.scala new file mode 100644 index 0000000..b7528dd --- /dev/null +++ b/src/main/scala/pubsub/network/TCPReader.scala @@ -0,0 +1,32 @@ +package pubsub.network + +import java.net.Socket + +import pubsub.Client +import pubsub.collection.BoundedBuffer +import pubsub.command._ + +class TCPReader(id: Int, socket: Socket, buffer: BoundedBuffer[Command]) { + val client = new Client(socket, id) + val reader = new CommandReader(socket.getInputStream(), client) + + def read(): Unit = { + client.sayHello() + println(s"New client: ${client.name}") + while(client.isConnected) { + + reader.fetchCommand() match { + case c: EndOfClient => + buffer.put(c) + println(c) + client.sayGoodbye() + client.close() + case _: MalformedCommand => + client.invalidPreviousCommand() + case command => + println(command) + buffer.put(command) + } + } + } +} diff --git a/src/test/scala/pubsub/BoundedBufferSuite.scala b/src/test/scala/pubsub/BoundedBufferSuite.scala new file mode 100644 index 0000000..ebc93fd --- /dev/null +++ b/src/test/scala/pubsub/BoundedBufferSuite.scala @@ -0,0 +1,43 @@ +package pubsub + +import scala.concurrent._ +import scala.concurrent.duration._ +import scala.concurrent.ExecutionContext.Implicits.global +import scala.collection.mutable.HashMap +import pubsub.collection._ +import org.junit._ +import org.junit.Assert.assertEquals + +import instrumentation._ +import instrumentation.Stats._ +import TestHelper._ +import TestUtils._ + +class BoundedBufferSuite { + @Test def `Should work in a sequential setting`: Unit = { + testSequential[(Int, Int, Int, Int)]{ sched => + val buffer = new SchedulableBoundedBuffer[Int](4, sched) + buffer.put(1) + buffer.put(2) + buffer.put(3) + buffer.put(4) + (buffer.take(), + buffer.take(), + buffer.take(), + buffer.take()) + }{ tuple => + (tuple == (1, 2, 3, 4), s"Expected (1, 2, 3, 4) got $tuple") + } + } + + @Test def `Should work when Thread 1: 'put(1)', Thread 2: 'take' and a buffer of size 1`: Unit = { + testManySchedules(2, sched => { + val prodCons = new SchedulableBoundedBuffer[Int](1, sched) + (List(() => prodCons.put(1), () => prodCons.take()), + args => (args(1) == 1, s"expected 1 your 'take' implementation returned ${args(1)}")) + }) + } + + + @Rule def individualTestTimeout = new org.junit.rules.Timeout(400 * 1000) +}