/* Ċ Ġ* ĠWork Ste aling Queue . kt Ċ Ġ* ĠCopyright ĠÂ© Ġ 1 9 9 3 - 2 0 2 1 , ĠThe ĠAv ail ĠFoundation , ĠLLC . Ċ Ġ* ĠAll Ġrights Ġreserved . Ċ Ġ* Ċ Ġ* ĠRedistribution Ġand Ġuse Ġin Ġsource Ġand Ġbinary Ġforms , Ġwith Ġor Ġwithout Ċ Ġ* Ġmodification , Ġare Ġpermitted Ġprovided Ġthat Ġthe Ġfollowing Ġconditions Ġare Ġmet : Ċ Ġ* Ċ Ġ* Ġ Ġ* ĠRedistributions Ġof Ġsource Ġcode Ġmust Ġretain Ġthe Ġabove Ġcopyright Ġnotice , Ċ Ġ* ĠĠĠ Ġthis Ġlist Ġof Ġconditions Ġand Ġthe Ġfollowing Ġdisclaimer . Ċ Ġ* Ċ Ġ* Ġ Ġ* ĠRedistributions Ġin Ġbinary Ġform Ġmust Ġreproduce Ġthe Ġabove Ġcopyright Ġnotice , Ċ Ġ* ĠĠĠ Ġthis Ġlist Ġof Ġconditions Ġand Ġthe Ġfollowing Ġdisclaimer Ġin Ġthe Ġdocumentation Ċ Ġ* ĠĠĠ Ġand / or Ġother Ġmaterials Ġprovided Ġwith Ġthe Ġdistribution . Ċ Ġ* Ċ Ġ* Ġ Ġ* ĠNeither Ġthe Ġname Ġof Ġthe Ġcopyright Ġholder Ġnor Ġthe Ġnames Ġof Ġthe Ġcontributors Ċ Ġ* ĠĠĠ Ġmay Ġbe Ġused Ġto Ġendorse Ġor Ġpromote Ġproducts Ġderived Ġfrom Ġthis Ġsoftware Ċ Ġ* ĠĠĠ Ġwithout Ġspecific Ġprior Ġwritten Ġpermission . Ċ Ġ* Ċ Ġ* ĠTHIS ĠSOFTWARE ĠIS ĠPROVIDED ĠBY ĠTHE ĠCOPYRIGHT ĠHOLDERS ĠAND ĠCONTRIBUTORS Ġ" AS ĠIS " Ċ Ġ* ĠAND ĠANY ĠEXPRESS ĠOR ĠIMPLIED ĠWARRANTIES , ĠINCLUDING , ĠBUT ĠNOT ĠLIMITED ĠTO , ĠTHE Ċ Ġ* ĠIMPLIED ĠWARRANTIES ĠOF ĠMERCHANTABILITY ĠAND ĠFITNESS ĠFOR ĠA ĠPARTICULAR ĠPURPOSE Ċ Ġ* ĠARE ĠDISCLAIMED . ĠIN ĠNO ĠEVENT ĠSHALL ĠTHE ĠCOPYRIGHT ĠHOLDER ĠOR ĠCONTRIBUTORS ĠBE Ċ Ġ* ĠLIABLE ĠFOR ĠANY ĠDIRECT , ĠINDIRECT , ĠINCIDENTAL , ĠSPECIAL , ĠEXEMPLARY , ĠOR Ċ Ġ* ĠCONSEQUENTIAL ĠDAMAGES Ġ( INCLUDING , ĠBUT ĠNOT ĠLIMITED ĠTO , ĠPROCUREMENT ĠOF Ċ Ġ* ĠSUBSTITUTE ĠGOODS ĠOR ĠSERVICES ; ĠLOSS ĠOF ĠUSE , ĠDATA , ĠOR ĠPROFITS ; ĠOR ĠBUSINESS Ċ Ġ* ĠINTERRUPTION ) ĠHOWEVER ĠCAUSED ĠAND ĠON ĠANY ĠTHEORY ĠOF ĠLIABILITY , ĠWHETHER ĠIN Ċ Ġ* ĠCONTRACT , ĠSTRICT ĠLIABILITY , ĠOR ĠTORT Ġ( INCLUDING ĠNEGLIGENCE ĠOR ĠOTHERWISE ) Ċ Ġ* ĠARISING ĠIN ĠANY ĠWAY ĠOUT ĠOF ĠTHE ĠUSE ĠOF ĠTHIS ĠSOFTWARE , ĠEVEN ĠIF ĠADVISED ĠOF ĠTHE Ċ Ġ* ĠPOSSIBILITY ĠOF ĠSUCH ĠDAMAGE . Ċ Ġ*/ Ċ Ċ package Ġcom . avail Ċ Ċ import Ġcom . avail . interpreter . execution . Interpreter Ċ import Ġjava . util . Abstract Queue Ċ import Ġjava . util . concurrent . BlockingQueue Ċ import Ġjava . util . concurrent . Priority BlockingQueue Ċ import Ġjava . util . concurrent . TimeUnit Ċ import Ġkotlin . math . max Ċ Ċ /** Ċ Ġ* ĠWIP ĠâĢĵ ĠThe Ġintent Ġis Ġto Ġuse Ġthis Ġqueue Ġto Ġstore Ġtasks Ġin Ġa Ġway Ġthat Ġa Ċ Ġ* ĠThreadPool Executor Ġcan Ġfetch Ġthem Ġin Ġa Ġmostly - priority Ġorder , Ġbut Ġwith Ġa Ċ Ġ* Ġpreference Ġfor Ġkeeping Ġwork Ġlocal Ġto Ġa Ġthread Ġif Ġthe Ġother Ġthreads Ġalready Ċ Ġ* Ġhave Ġtheir Ġown Ġwork Ġto Ġdo . Ċ Ġ*/ Ċ class ĠWork Ste aling Queue < E > Ċ constructor ( Ċ ĉ private Ġval Ġparallel ism : ĠInt Ċ ) Ġ: ĠBlock ing Queue < E >, ĠAbstract Queue < E >() Ċ { Ċ ĉ /** Ċĉ Ġ* ĠAn Ġarray Ġof Ġregular Ġ[ Priority BlockingQueue ] s , Ġone Ġdedicated Ġto Ġeach Ċĉ Ġ* Ġworker Ġthread Ġof Ġthe Ġthread Ġpool Ġthat Ġthis Ġinstance Ġis Ġplug ged Ġinto . Ċĉ Ġ* ĠNew Ġtasks Ġfrom Ġthread Ġ# N Ġalways Ġgets Ġadded Ġto Ġqueue Ġ# N , Ġbut Ġother Ġthreads Ċĉ Ġ* Ġmay Ġste al Ġthe Ġtask Ġ( and Ġrun Ġit ) Ġif Ġthey 're Ġotherwise Ġidle . Ċĉ Ġ*/ Ċ ĉ private Ġval Ġqueues Ġ= ĠArray ( parallel ism ) Ġ{ ĠPriority BlockingQueue < E >() Ġ} ĊĊ ĉ /** Ċĉ Ġ* ĠA ĠJava Ġ[ Object ] Ġto Ġuse Ġas Ġa Ġmonitor Ġfor Ġthreads Ġto Ġblock Ġon , Ġwaiting Ġfor Ċĉ Ġ* Ġa Ġsuitable Ġtask Ġto Ġbecome Ġavailable . Ċĉ Ġ*/ Ċ ĉ private Ġval Ġmonitor Ġ= ĠObject () ĊĊ ĉ /** ĠChoose Ġthe Ġqueue Ġon Ġwhich Ġto Ġadd Ġa Ġtask . Ġ*/ Ċ ĉ private Ġval Ġlocal Queue Ġget () Ġ= Ġqueues [ Interpreter . currentIndex Or Zero ()] ĊĊ ĉ override Ġval Ġsize : ĠInt Ċĉ ĉ get () Ġ= Ġqueues . sum Of ( Priority BlockingQueue < E >:: size ) ĊĊ ĉ override Ġfun Ġdrain To ( c : ĠMutable Collection < in ĠE >): ĠInt Ġ= Ċĉ ĉ queues . sum Of Ġ{ Ġit . drain To ( c ) Ġ} ĊĊ ĉ override Ġfun Ġdrain To ( c : ĠMutable Collection < in ĠE >, Ġmax Elements : ĠInt ): ĠInt Ċ ĉ { Ċĉ ĉ var Ġremaining Ġ= Ġmax Elements Ċĉ ĉ for Ġ( queue Ġin Ġqueues ) Ċĉ ĉ { Ċĉĉ ĉ remaining Ġ-= Ġqueue . drain To ( c , Ġremaining ) Ċĉĉ ĉ if Ġ( remaining Ġ<= Ġ 0 ) Ġbreak Ċĉ ĉ } Ċĉ ĉ return Ġmax Elements Ġ- Ġremaining Ċ ĉ } ĊĊ ĉ override Ġfun Ġoffer ( element : ĠE ): ĠBoolean Ċ ĉ { Ċĉ ĉ val Ġqueue Ġ= Ġlocal Queue Ċĉ ĉ val Ġchanged Ġ= Ġqueue . add ( element ) Ċĉ ĉ // ĠWhen Ġthe Ġlocal Ġqueue Ġreaches Ġspecific Ġsizes , Ġallow Ġanother Ġthread Ġto Ċĉ ĉ // Ġwake Ġup , Ġif Ġit 's Ġwaiting . Ġ ĠIf Ġno Ġthread Ġis Ġwaiting , Ġthis Ġhas Ġno Ċĉ ĉ // Ġeffect . Ċĉ ĉ // Ċĉ ĉ // ĠThe Ġnotify Ġis Ġnecessary Ġwhen Ġthe Ġqueue Ġreaches Ġsize Ġ 1 , Ġsince Ġall Ċĉ ĉ // Ġthreads Ġmight Ġbe Ġblocked , Ġand Ġwe Ġneed Ġto Ġwake Ġup Ġone Ġof Ġthem Ġto Ġscan Ċĉ ĉ // Ġfor Ġthis Ġnewly Ġadded Ġtask . Ċĉ ĉ // Ċĉ ĉ // ĠIf Ġa Ġtask Ġadds Ġitems Ġfast Ġenough Ġthat Ġa Ġqueue Ġgets Ġto Ġ 2 , Ġwe Ġwake Ċĉ ĉ // Ġanother Ġthread Ġin Ġthe Ġhope Ġthat Ġit Ġcan Ġste al Ġone Ġof Ġthese Ġtasks Ġand Ċĉ ĉ // Ġstart Ġprocessing Ġthe Ġresulting Ġtree Ġof Ġtasks Ġgenerated Ġfrom Ġit . Ċĉ ĉ // Ċĉ ĉ // ĠIf Ġthe Ġlocal Ġqueue Ġmanages Ġto Ġget Ġ 5 Ġelements , Ġwe Ġwake Ġup Ġtwo Ġother Ċĉ ĉ // Ġthreads , Ġto Ġaccel erate Ġprocessing . Ċĉ ĉ // Ċĉ ĉ // ĠIf Ġwe 've Ġqueued Ġabout Ġas Ġmany Ġitems Ġin Ġthe Ġlocal Ġqueue Ġas Ġthere Ġare Ċĉ ĉ // Ġthreads , Ġwake Ġthem Ġall Ġup . Ċĉ ĉ when Ġ( queue . size ) Ċĉ ĉ { Ċĉĉĉ 1 , Ġ 2 Ġ-> Ġsynchronized ( monitor ) Ġ{ Ġmonitor . notify () Ġ} Ċĉĉĉ 5 Ġ-> Ġsynchronized ( monitor ) Ġ{ Ċĉĉĉ ĉ monitor . notify () Ċĉĉĉ ĉ monitor . notify () Ċĉĉ ĉ } Ċĉĉ ĉ parallel ism Ġ+ Ġ 1 Ġ-> Ġsynchronized ( monitor ) Ġ{ Ġmonitor . notify All () Ġ} Ċĉ ĉ } Ċĉ ĉ return Ġchanged Ċ ĉ } ĊĊ ĉ override Ġfun Ġoffer ( e : ĠE , Ġtimeout : ĠLong , Ġunit : ĠTimeUnit ): ĠBoolean Ġ= Ġoffer ( e ) ĊĊ ĉ override Ġfun Ġpoll (): ĠE ? Ċ ĉ { Ċĉ ĉ // ĠFirst Ġtry Ġthe Ġqueue Ġdedicated Ġto Ġthe Ġcurrent Ġthread . Ġ ĠWhen Ġthe Ġnumber Ċĉ ĉ // Ġof Ġthreads Ġis Ġstable Ġ( due Ġto Ġlow , Ġhigh , Ġor Ġconstant Ġload ), Ġthe Ġqueue Ċĉ ĉ // Ġfor Ġthe Ġcurrent Ġthread Ġwon 't Ġbe Ġunder Ġsignificant Ġcontent ion . Ċĉ ĉ local Queue . poll ()?. let Ġ{ Ġreturn Ġit Ġ} Ċĉ ĉ // ĠLocal Ġqueue Ġdidn 't Ġhave Ġanything . Ġ ĠLook Ġfor Ġsomething Ġto Ġste al . Ċĉ ĉ for Ġ( q Ġin Ġqueues ) Ċĉ ĉ { Ċĉĉ ĉ q . poll ()?. let Ġ{ Ġreturn Ġit Ġ} Ċĉ ĉ } Ċĉ ĉ return Ġnull Ċ ĉ } ĊĊ ĉ override Ġfun Ġpoll ( timeout : ĠLong , Ġunit : ĠTimeUnit ): ĠE ? Ċ ĉ { Ċĉ ĉ // ĠFirst Ġtry Ġit Ġwithout Ġtimeout . Ċĉ ĉ poll ()?. let Ġ{ Ġreturn Ġit Ġ} Ċĉ ĉ when Ġ( timeout ) Ċĉ ĉ { Ċĉĉĉ 0 L Ġ-> Ċĉĉ ĉ { Ċĉĉĉ ĉ synchronized ( monitor ) Ġ{ Ċĉĉĉĉ ĉ while Ġ( true ) Ċĉĉĉĉ ĉ { Ċĉĉĉĉĉ ĉ monitor . wait () Ċĉĉĉĉĉ ĉ // ĠRe - check , Ġbecause Ġsomeone Ġwo ke Ġus Ġup . Ċĉĉĉĉĉ ĉ poll ()?. let Ġ{ Ġreturn Ġit Ġ} Ċĉĉĉĉ ĉ } Ċĉĉĉ ĉ } Ċĉĉ ĉ } Ċĉĉ ĉ else Ġ-> Ċĉĉ ĉ { Ċĉĉĉ ĉ val Ġexpiry Ġ= ĠSystem . nano Time () Ġ+ Ġunit . to Nan os ( timeout ) Ċĉĉĉ ĉ synchronized ( monitor ) Ġ{ Ċĉĉĉĉ ĉ while Ġ( true ) Ċĉĉĉĉ ĉ { Ċĉĉĉĉĉ ĉ val Ġnan os Ġ= Ġwhen Ġ( timeout ) Ċĉĉĉĉĉ ĉ { Ċĉĉĉĉĉĉĉ 0 L Ġ-> Ġ 0 L Ċĉĉĉĉĉĉ ĉ else Ġ-> Ġmax ( 1 L , Ġexpiry Ġ- ĠSystem . nano Time ()) Ċĉĉĉĉĉ ĉ } Ċĉĉĉĉĉ ĉ if Ġ( nan os Ġ== Ġ 0 L Ġ&& Ġtimeout Ġ> Ġ 0 L ) Ċĉĉĉĉĉ ĉ { Ċĉĉĉĉĉĉ ĉ // ĠActually Ġtimed Ġout . Ċĉĉĉĉĉĉ ĉ return Ġnull Ċĉĉĉĉĉ ĉ } Ċĉĉĉĉĉ ĉ monitor . wait ( Ċĉĉĉĉĉĉ ĉ nan os Ġ/ Ġ 1 _ 0 0 0 _ 0 0 0 , Ċĉĉĉĉĉĉ ĉ ( nan os Ġ% Ġ 1 _ 0 0 0 _ 0 0 0 ). toInt ()) Ċĉĉĉĉĉ ĉ // ĠRe - check , Ġeven Ġif Ġit Ġjust Ġtimed Ġout . Ċĉĉĉĉĉ ĉ poll ()?. let Ġ{ Ġreturn Ġit Ġ} Ċĉĉĉĉ ĉ } Ċĉĉĉ ĉ } Ċĉĉ ĉ } Ċĉ ĉ } Ċ ĉ } ĊĊ ĉ override Ġfun Ġput ( e : ĠE ) Ċ ĉ { Ċĉ ĉ offer ( e ) Ċ ĉ } ĊĊ ĉ override Ġfun Ġtake (): ĠE Ċ ĉ { Ċĉ ĉ // ĠFirst Ġtry Ġit Ġwithout Ġtimeout . Ċĉ ĉ poll ()?. let Ġ{ Ġreturn Ġit Ġ} Ċĉ ĉ synchronized ( monitor ) Ġ{ Ċĉĉ ĉ while Ġ( true ) Ċĉĉ ĉ { Ċĉĉĉ ĉ monitor . wait () Ċĉĉĉ ĉ // ĠRe - check , Ġbecause Ġsomeone Ġwo ke Ġus Ġup . Ċĉĉĉ ĉ poll ()?. let Ġ{ Ġreturn Ġit Ġ} Ċĉĉ ĉ } Ċĉ ĉ } Ċ ĉ } ĊĊ ĉ override Ġfun Ġremaining Capacity (): ĠInt Ġ= ĠInt . MAX _ VALUE ĊĊ ĉ override Ġfun Ġremove ( element : ĠE ): ĠBoolean Ġ= Ġqueues . any Ġ{ Ġit . remove ( element ) Ġ} ĊĊ ĉ override Ġfun Ġiterator (): ĠMutable Iterator < E > Ċ ĉ { Ċĉ ĉ // ĠNot Ġefficient , Ġbut Ġtechnically Ġcorrect . Ċĉ ĉ val Ġlist Ġ= Ġmutable ListOf < E >() Ċĉ ĉ queues . forEach Ġ{ Ġq Ġ-> Ġq . forEach Ġ{ Ġe Ġ-> Ġlist . add ( e ) Ġ} Ġ} Ċĉ ĉ return Ġlist . iterator () Ċ ĉ } ĊĊ ĉ override Ġfun Ġpeek (): ĠE ? Ġ= Ċĉ ĉ local Queue . peek () Ġ?: Ġqueues . first NotNull Of OrNull Ġ{ Ġit . peek () Ġ} Ċ } ĊĊ