/* Ċ Ġ* ĠCopyright Ġ( c ) Ġ 2 0 2 0 ĠAt Large ĠResearch Ċ Ġ* Ċ Ġ* ĠPermission Ġis Ġhereby Ġgranted , Ġfree Ġof Ġcharge , Ġto Ġany Ġperson Ġobtaining Ġa Ġcopy Ċ Ġ* Ġof Ġthis Ġsoftware Ġand Ġassociated Ġdocumentation Ġfiles Ġ( the Ġ" Software "), Ġto Ġdeal Ċ Ġ* Ġin Ġthe ĠSoftware Ġwithout Ġrestriction , Ġincluding Ġwithout Ġlimitation Ġthe Ġrights Ċ Ġ* Ġto Ġuse , Ġcopy , Ġmodify , Ġmerge , Ġpublish , Ġdistribute , Ġsublicense , Ġand / or Ġsell Ċ Ġ* Ġcopies Ġof Ġthe ĠSoftware , Ġand Ġto Ġpermit Ġpersons Ġto Ġwhom Ġthe ĠSoftware Ġis Ċ Ġ* Ġfurnished Ġto Ġdo Ġso , Ġsubject Ġto Ġthe Ġfollowing Ġconditions : Ċ Ġ* Ċ Ġ* ĠThe Ġabove Ġcopyright Ġnotice Ġand Ġthis Ġpermission Ġnotice Ġshall Ġbe Ġincluded Ġin Ġall Ċ Ġ* Ġcopies Ġor Ġsubstantial Ġportions Ġof Ġthe ĠSoftware . Ċ Ġ* Ċ Ġ* ĠTHE ĠSOFTWARE ĠIS ĠPROVIDED Ġ" AS ĠIS ", ĠWITHOUT ĠWARRANTY ĠOF ĠANY ĠKIND , ĠEXPRESS ĠOR Ċ Ġ* ĠIMPLIED , ĠINCLUDING ĠBUT ĠNOT ĠLIMITED ĠTO ĠTHE ĠWARRANTIES ĠOF ĠMERCHANTABILITY , Ċ Ġ* ĠFITNESS ĠFOR ĠA ĠPARTICULAR ĠPURPOSE ĠAND ĠNONINFRINGEMENT . ĠIN ĠNO ĠEVENT ĠSHALL ĠTHE Ċ Ġ* ĠAUTHORS ĠOR ĠCOPYRIGHT ĠHOLDERS ĠBE ĠLIABLE ĠFOR ĠANY ĠCLAIM , ĠDAMAGES ĠOR ĠOTHER Ċ Ġ* ĠLIABILITY , ĠWHETHER ĠIN ĠAN ĠACTION ĠOF ĠCONTRACT , ĠTORT ĠOR ĠOTHERWISE , ĠARISING ĠFROM , Ċ Ġ* ĠOUT ĠOF ĠOR ĠIN ĠCONNECTION ĠWITH ĠTHE ĠSOFTWARE ĠOR ĠTHE ĠUSE ĠOR ĠOTHER ĠDEALINGS ĠIN ĠTHE Ċ Ġ* ĠSOFTWARE . Ċ Ġ*/ Ċ Ċ package Ġorg . opend c . experiments . allocate am . policies Ċ Ċ import Ġmu . Kotlin Logging Ċ import Ġorg . opend c . workflows . service . Job State Ċ import Ġorg . opend c . workflows . service . Stage Workflow Scheduler Listener Ċ import Ġorg . opend c . workflows . service . Stage Workflow Service Ċ import Ġorg . opend c . workflows . service . Task State Ċ import Ġorg . opend c . workflows . service . stage . task . Task El ig ibility Policy Ċ import Ġorg . opend c . workflows . workload . Task Ċ import Ġjava . util .* Ċ Ċ private Ġval Ġlogger Ġ= ĠKotlin Logging . logger Ġ{} ĊĊ Ċ /** Ċ Ġ* ĠA Ġ[ Task El ig ibility Policy ] Ġthat Ġlimits Ġthe Ġnumber Ġof Ġactive Ġtasks Ġof Ġa Ġjob Ġin Ġthe Ġsystem . Ċ Ġ*/ Ċ public Ġdata Ġclass ĠRound Rob in Policy ( public Ġval Ġquant a : ĠInt ) Ġ: ĠTask El ig ibility Policy Ġ{ ĊĊĠĠĠ Ġpublic Ġdata Ġclass ĠQue ued Job ( val Ġjob Id : ĠUUID , Ġvar Ġtasks : ĠMutable Set < Task >, Ġvar Ġtotal NumberOf Tasks : ĠInt ) ĊĊĠĠĠ Ġoverride Ġfun Ġinvoke ( scheduler : ĠStage Workflow Service ): ĠTask El ig ibility Policy . Logic Ġ= ĊĠĠĠĠĠĠĠ Ġobject Ġ: ĠTask El ig ibility Policy . Logic , ĠStage Workflow Scheduler Listener Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġvar Ġnum Scheduled So Far Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġactive Queue : ĠQueue < Queued Job > Ġ= ĠLinkedList () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġvar Ġnum Jobs Started Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġvar Ġnum Jobs Finished Ġ= Ġ 0 ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscheduler . addListener ( this ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġjob Started ( job : ĠJob State ) Ġ{ Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Job Ġstarted : Ġ${ job . job . uid } Ġnumber Ġof Ġjobs Ġstarted : Ġ${ ++ num Jobs Started }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġqueued Job Ġ= ĠQue ued Job ( job . job . uid , Ġmutable Set Of (), Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġactive Queue . add ( queued Job ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġjob Finished ( job : ĠJob State ) Ġ{ Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Job Ġfinished : Ġ${ job . job . uid } Ġnumber Ġof Ġjobs Ġfinished : Ġ${ ++ num Jobs Finished }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġtask Ready ( task : ĠTask State ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġqueued Jobs Ġ= Ġactive Queue . filter Ġ{ Ġit . jobId . equals ( task . job . job . uid ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif ( queued Jobs . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġqueued Jobs [ 0 ]. tasks . add ( task . task ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġqueued Jobs [ 0 ]. total NumberOf Tasks Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġqueued Job Ġ= ĠQue ued Job ( task . job . job . uid , Ġmutable Set Of ( task . task ), Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġactive Queue . add ( queued Job ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġinvoke ( task : ĠTask State ): ĠTask El ig ibility Policy . Ad vice Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġwe Ġalready Ġscheduled Ġa Ġquant a Ġamount Ġduring Ġthis Ġbatch , Ġwe Ġstop Ġscheduling Ġany Ġmore Ġtasks Ġfor Ġthis ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġbatch ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif ( num Scheduled So Far Ġ+ Ġ 1 Ġ> Ġquant a ) Ġ{ Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Quant a Ġof Ġ$ quant a Ġexceeded ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnum Scheduled So Far Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠTask El ig ibility Policy . Ad vice . STOP ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjob : ĠQue ued Job Ġ= Ġactive Queue . peek () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġcheck Ġto Ġsee Ġif Ġthe Ġcurrent Ġtask Ġis Ġpart Ġof Ġthe Ġsame Ġjob Ġas Ġthe Ġtop Ġof Ġthe Ġjob Ġqueue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif (! task . job . job . uid . equals ( job . jobId )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠTask El ig ibility Policy . Ad vice . DEN Y ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġtask Is Queued Ġ= Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġcheck Ġto Ġsee Ġif Ġthe Ġtask Ġis Ġin Ġthe Ġqueued Ġjob Ġset ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfor ( queued Task Ġin Ġjob . tasks ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif ( queued Task . equals ( task . task )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtask Is Queued Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif (! task Is Queued ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠTask El ig ibility Policy . Ad vice . DEN Y ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġremove Ġthe Ġtask Ġfrom Ġthe Ġlist Ġof Ġtasks Ġof Ġthe Ġqueued Ġjob Ġto Ġexecute ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjob . tasks . remove ( task . task ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġwe Ġfinished Ġall Ġthe Ġtasks Ġfrom Ġthe Ġqueued Ġjob Ġset , Ġwe Ġpop Ġit Ġfrom Ġthe Ġqueue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif ( job . tasks . size Ġ== Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġactive Queue . remove () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠOtherwise Ġwe Ġcreate Ġa Ġnew Ġqueued Ġjob Ġwith Ġremaining Ġof Ġthe Ġtasks Ġto Ġbe Ġexecuted ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġif ( job . total NumberOf Tasks Ġ- Ġjob . tasks . size Ġ== Ġquant a ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġremaining Job Ġ= Ġactive Queue . remove () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġremaining Job . total NumberOf Tasks Ġ= Ġremaining Job . tasks . size ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġactive Queue . add ( remaining Job ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnum Scheduled So Far ++ Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Scheduling Ġjob : Ġ${ task . job . job . uid }, Ġtask : Ġ${ task . task . uid }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠTask El ig ibility Policy . Ad vice . AD MIT ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠtoString (): ĠString Ġ= Ġ" Round - Rob in - Quant a - ($ quant a )" Ċ } Ċ