package Ġnet . just m achinery . f utility . execution Ċ Ċ import Ġjava . lang . ref . Weak Reference Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . Future Ċ import Ġjava . util . concurrent . Linked Blocking Deque Ċ import Ġjava . util . concurrent . Semaphore Ċ import Ġjava . util . concurrent . atomic . Atomic Integer Ċ Ċ /** Ċ Ġ* ĠProcess Ġitems Ġin Ġa Ġnumber Ġof Ġbackground Ġthreads Ġin Ġthe Ġdefault Ġexecutor Ġ[ pools ]. Ċ Ġ* ĠProbably Ġa Ġbad Ġidea Ġto Ġcall Ġthis Ġfrom Ġa Ġbackground Ġthread ; Ġyou Ġcould Ġtheoret ically Ġblock Ġall Ġthreads Ġin Ġthe Ġpool Ċ Ġ* Ġwith Ġparallel Maps Ġwhich Ġwill Ġnever Ġprogress . Ċ Ġ* Ġ@ param Ġmax Concurrent ĠMaximum Ġnumber Ġof Ġbackground Ġthreads Ġrunning Ġfor Ġthis Ġcall Ġat Ġany Ġgiven Ġtime , Ġor Ġun limited Ġif Ġnull . Ċ Ġ* Ġ@ param Ġmax Buffer ĠMaximum Ġnumber Ġof Ġun processed Ġresults Ġat Ġany Ġtime . Ġ( Should Ġbe Ġlarger Ġthan Ġmax Concurrent ). Ċ Ġ* Ġ@ param Ġstop On First Exception ĠWhether Ġto Ġstop Ġthe Ġentire Ġsequence Ġfrom Ġgenerating Ġfuture Ġitems Ġwhen Ġthe Ġfirst Ġexception Ġis Ġencountered Ċ Ġ* Ġ@ param Ġtransform ĠFunction Ġto Ġapply Ġto Ġall Ġitems Ġof Ġsequence Ċ Ġ*/ Ċ public Ġfun Ġ< T , ĠR > ĠSequence < T >. parallel Map ( ĊĠĠĠ Ġmax Concurrent Ġ: ĠInt ? Ġ= Ġnull , ĊĠĠĠ Ġmax Buffer Ġ: ĠInt ? Ġ= Ġnull , ĊĠĠĠ Ġstop On First Exception Ġ: ĠBoolean Ġ= Ġtrue , ĊĠĠĠ Ġtransform : Ġ( T ) Ġ-> ĠR Ċ ): ĠSequence < R > Ġ{ ĊĠĠĠ Ġreturn Ġthis . parallel Map Internal ( ĊĠĠĠĠĠĠĠ Ġmax Concurrent Ġ= Ġmax Concurrent , ĊĠĠĠĠĠĠĠ Ġmax Buffer Ġ= Ġmax Buffer , ĊĠĠĠĠĠĠĠ Ġstop On First Exception Ġ= Ġstop On First Exception , ĊĠĠĠĠĠĠĠ Ġtransform Ġ= Ġtransform ĊĠĠĠ Ġ). map Ġ{ Ġit . get () Ġ} Ċ } Ċ Ċ public Ġfun Ġ< T , ĠR > ĠIterable < T >. parallel Map ( ĊĠĠĠ Ġmax Concurrent Ġ: ĠInt ? Ġ= Ġnull , ĊĠĠĠ Ġmax Buffer Ġ: ĠInt ? Ġ= Ġnull , ĊĠĠĠ Ġstop On First Exception Ġ: ĠBoolean Ġ= Ġtrue , ĊĠĠĠ Ġtransform : Ġ( T ) Ġ-> ĠR Ċ ): ĠIterable < R > Ġ{ ĊĠĠĠ Ġreturn Ġthis . as Sequence (). parallel Map Internal ( ĊĠĠĠĠĠĠĠ Ġmax Concurrent Ġ= Ġmax Concurrent , ĊĠĠĠĠĠĠĠ Ġmax Buffer Ġ= Ġmax Buffer , ĊĠĠĠĠĠĠĠ Ġstop On First Exception Ġ= Ġstop On First Exception , ĊĠĠĠĠĠĠĠ Ġtransform Ġ= Ġtransform ĊĠĠĠ Ġ). as Iterable (). map Ġ{ Ġit . get () Ġ} Ċ } Ċ Ċ private Ġclass ĠParallel Map State < T , ĠR >( ĊĠĠĠ Ġmax Buffer Ġ: ĠInt ?, ĊĠĠĠ Ġval Ġiter Ġ: ĠIterator < T > Ċ ){ ĊĠĠĠ Ġval Ġavailable Buffer Space Ġ= ĠSem aphore ( max Buffer Ġ?: ĠInt . MAX _ VALUE , Ġfalse ) ĊĠĠĠ Ġval Ġfutures Queue Ġ= ĠLinked Blocking Deque < Optional < Future < R >> >() Ġ// Deque Ġdoes Ġnot Ġlike Ġnull s ĊĠĠĠ Ġvar Ġcurrent Queue Number Ġ= Ġ 0 ĊĠĠĠ Ġvar Ġhad Exception Ġ= Ġfalse ĊĠĠĠ Ġvar Ġavailable Concurrent F utures Ġ= ĠAtomicInteger ( 0 ) Ċ } Ċ Ċ private Ġfun Ġ< T , ĠR > ĠSequence < T >. parallel Map Internal ( ĊĠĠĠ Ġmax Concurrent Ġ: ĠInt ? Ġ= Ġnull , ĊĠĠĠ Ġmax Buffer Ġ: ĠInt ? Ġ= Ġnull , ĊĠĠĠ Ġstop On First Exception Ġ: ĠBoolean Ġ= Ġtrue , ĊĠĠĠ Ġtransform : Ġ( T ) Ġ-> ĠR Ċ ): ĠSequence < Future < R >> Ġ{ ĊĠĠĠ Ġ// There Ġare Ġa Ġnumber Ġof Ġtricky Ġconstraints Ġinvolved Ġin Ġgetting Ġthis Ġright . ĊĠĠĠ Ġ// The Ġideal Ġbehavior Ġis : ĊĠĠĠ Ġ// - ĠUp Ġto Ġmax Concurrent Ġbackground Ġtasks Ġare Ġactively Ġcomputing Ġthe Ġtransformation Ġon Ġelements Ġof Ġthe Ġlist Ġat Ġany Ġtime ĊĠĠĠ Ġ// - ĠElements Ġmust Ġbe Ġdelivered Ġin Ġorder Ġas Ġif Ġthey Ġwere Ġcomputed Ġser ially ĊĠĠĠ Ġ// - ĠNo Ġmore Ġthan Ġmax Buffer Ġfutures Ġshould Ġbe Ġoutstanding Ġat Ġany Ġtime ĊĠĠĠ Ġ// - ĠIf Ġany Ġitem Ġin Ġthe Ġsequence Ġthrows Ġan Ġexception , Ġno Ġmore Ġwork Ġshould Ġbe Ġdone Ġon Ġitems Ġafter Ġit . ĊĠĠĠ Ġ// - ĠIf Ġthe Ġreference Ġto Ġthe Ġreturned Ġsequence Ġis Ġlost , Ġno Ġfurther Ġbackground Ġtasks Ġshould Ġbe Ġspawn ed Ġto Ġcompute Ġit . ĊĠĠĠ Ġreturn Ġif ( max Concurrent Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstate Ġ= ĠParallel Map State < T , ĠR >( max Buffer , Ġthis . iterator ()) ĊĠĠĠĠĠĠĠ Ġ// Access ing Ġthe Ġstate Ġthrough Ġa Ġweak Ġreference Ġallows Ġit Ġto Ġbe Ġgarbage Ġwhen Ġthe Ġmain Ġthread Ġis Ġdone Ġwith Ġit , ĊĠĠĠĠĠĠĠ Ġ// th us Ġsignal ing Ġbackground Ġthreads Ġnot Ġto Ġspawn Ġmore Ġwork Ġprocessors . ĊĠĠĠĠĠĠĠ Ġval Ġref Ġ= ĠWeak Reference ( state ) ĊĠĠĠĠĠĠĠ Ġfun Ġstart Next Future (){ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpar State Ġ= Ġref . get () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif ( par State Ġ== Ġnull Ġ|| Ġ( stop On First Exception Ġ&& Ġpar State . had Exception )){ Ġreturn Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif ( par State . available Buffer Space . try Acquire ()){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( par State ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif ( par State . iter . hasNext ()){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnext Ġ= Ġpar State . iter . next () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfuture Ġ= Ġfuture Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtransform ( next ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch ( t Ġ: ĠThrowable ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpar State . had Exception Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġt ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart Next Future () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpar State . futures Queue . add ( Optional . of ( future )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpar State . current Queue Number Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpar State . futures Queue . add ( Optional . empty ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Since Ġno Ġnew Ġfuture Ġwas Ġspawn ed Ġdue Ġto Ġa Ġlack Ġof Ġbuffer Ġspace , Ġmain Ġsequence Ġmay Ġspawn Ġmore Ġfutures Ġlater . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// This Ġbackground Ġthread Ġwill Ġbe Ġreleased Ġimmediately . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpar State . available Concurrent F utures . increment AndGet () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġsynchronized ( state ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfor ( i Ġin Ġ 0 Ġuntil Ġmax Concurrent ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart Next Future () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġsequence Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile ( true ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnext Ġ= Ġstate . futures Queue . take First (). orElse ( null ) Ġ?: Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate . available Buffer Space . release () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġyield ( next ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif ( state . available Concurrent F utures . get () Ġ> Ġ 0 ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate . available Concurrent F utures . decrement AndGet () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart Next Future () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Force Ġall Ġto Ġbegin Ġcalculating ĊĠĠĠĠĠĠĠ Ġthis . map Ġ{ Ġfuture Ġ{ Ġtransform ( it ) Ġ} Ġ}. toList (). as Sequence () ĊĠĠĠ Ġ} Ċ } ĊĊ