/* Ċ package Ġtech . pr ong h orn . coroutines . await able . queue Ċ Ċ import Ġorg . junit . jupiter . api . Assertions .* Ċ import Ġorg . junit . jupiter . api . Repeated Test Ċ import Ġtech . pr ong h orn . coroutines . await able . future . Coroutine Future Ċ import Ġtech . pr ong h orn . coroutines . core .* Ċ import Ġtech . pr ong h orn . test .* Ċ import Ġjava . time . Duration Ċ import Ġjava . util . concurrent . atomic . Atomic Integer Ċ Ċ class ĠS pmc Pre allocated Queue Tests Ġ: ĠPr ong h orn Test () Ġ{ ĊĠĠĠ Ġclass ĠData Holder ( var Ġlong A : ĠLong Ġ= Ġ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġlong B : ĠLong Ġ= Ġ 0 ) ĊĊĠĠĠ Ġ@ Repeated Test ( light Repeat Count ) ĊĠĠĠ Ġfun Ġsingle Thread St ress Test () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġbuffer Ġ= ĠS pmc Pre allocated Queue ( 1 0 2 4 , Ġ 2 , Ġ{ ĠData Holder () Ġ}) ĊĠĠĠĠĠĠĠ Ġval Ġwriter Ġ= Ġbuffer . writer ĊĠĠĠĠĠĠĠ Ġval Ġreader A Ġ= Ġbuffer . get Reader ( 0 ) ĊĠĠĠĠĠĠĠ Ġval Ġreader B Ġ= Ġbuffer . get Reader ( 1 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġcount Ġ= Ġ 1 0 0 0 0 0 0 0 0 L ĊĠĠĠĠĠĠĠ Ġval Ġpre Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠĠĠĠĠ Ġvar Ġx Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġvar Ġtotal Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġwhile Ġ( x Ġ< Ġcount ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( x Ġ< Ġcount Ġ&& Ġwriter . offer Ġ{ Ġit . long A Ġ= Ġx ; Ġit . long B Ġ= Ġx Ġ}) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġx Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwriter . publish () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġresult Ġ= Ġreader A . poll () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( result Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtotal Ġ+= Ġresult . long A Ġ+ Ġresult . long B ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġreader A . poll () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreader A . commit () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġreader B . poll () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( result Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtotal Ġ+= Ġresult . long A Ġ+ Ġresult . long B ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġreader B . poll () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreader B . commit () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġpost Ġ= ĠSystem . currentTimeMillis () ĊĊĠĠĠĠĠĠĠ Ġvar Ġz Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġvar Ġexpected Total Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġwhile Ġ( z Ġ< Ġcount ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpected Total Ġ+= Ġz Ġ* Ġ 2 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġz Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprintln ("$ total Ġ: Ġ$ expected Total ") ĊĠĠĠĠĠĠĠ Ġlogger . info Ġ{ Ġ" T ook Ġ${ post Ġ- Ġpre } ms Ġfor Ġ$ count , Ġ${ ( count Ġ/ ĠMath . max ( 1 , Ġ( post Ġ- Ġpre ))) Ġ/ Ġ 1 0 0 0 . 0 } Ġmillion Ġper Ġsecond " Ġ} ĊĠĠĠĠĠĠĠ ĠassertEquals ( expected Total , Ġtotal ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġclass ĠWrite Thread ( val Ġbuffer : ĠS pmc Pre allocated Queue < Data Holder >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcount : ĠLong ) Ġ: ĠThread () Ġ{ ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġa 0 0 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 1 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 2 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 3 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġa 0 4 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 5 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 6 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 7 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġa 0 8 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 9 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 1 0 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 1 1 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġa 1 2 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 1 3 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 1 4 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 1 5 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġb 0 0 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 1 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 2 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 3 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġb 0 4 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 5 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 6 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 7 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġb 0 8 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 9 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 1 0 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 1 1 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġb 1 2 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 1 3 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 1 4 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 1 5 Ġ= Ġ 0 L ĊĊĠĠĠĠĠĠĠ Ġpublic Ġfun ĠgetAll (): ĠLong Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġa 0 0 Ġ+ Ġa 0 1 Ġ+ Ġa 0 2 Ġ+ Ġa 0 3 Ġ+ Ġa 0 4 Ġ+ Ġa 0 5 Ġ+ Ġa 0 6 Ġ+ Ġa 0 7 Ġ+ Ġa 0 8 Ġ+ Ġa 0 9 Ġ+ Ġa 1 0 Ġ+ Ġa 1 1 Ġ+ Ġa 1 2 Ġ+ Ġa 1 3 Ġ+ Ġa 1 4 Ġ+ Ġa 1 5 Ġ+ Ġb 0 0 Ġ+ Ġb 0 1 Ġ+ Ġb 0 2 Ġ+ Ġb 0 3 Ġ+ Ġb 0 4 Ġ+ Ġb 0 5 Ġ+ Ġb 0 6 Ġ+ Ġb 0 7 Ġ+ Ġb 0 8 Ġ+ Ġb 0 9 Ġ+ Ġb 1 0 Ġ+ Ġb 1 1 Ġ+ Ġb 1 2 Ġ+ Ġb 1 3 Ġ+ Ġb 1 4 Ġ+ Ġb 1 5 ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġpublic Ġfun Ġset All ( v : ĠLong ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġa 0 0 Ġ= Ġv ; Ġa 0 1 Ġ= Ġv ; Ġa 0 2 Ġ= Ġv ; Ġa 0 3 Ġ= Ġv ; Ġa 0 4 Ġ= Ġv ; Ġa 0 5 Ġ= Ġv ; Ġa 0 6 Ġ= Ġv ; Ġa 0 7 Ġ= Ġv ; Ġa 0 8 Ġ= Ġv ; Ġa 0 9 Ġ= Ġv ; Ġa 1 0 Ġ= Ġv ; Ġa 1 1 Ġ= Ġv ; Ġa 1 2 Ġ= Ġv ; Ġa 1 3 Ġ= Ġv ; Ġa 1 4 Ġ= Ġv ; Ġa 1 5 Ġ= Ġv ; Ġb 0 0 Ġ= Ġv ; Ġb 0 1 Ġ= Ġv ; Ġb 0 2 Ġ= Ġv ; Ġb 0 3 Ġ= Ġv ; Ġb 0 4 Ġ= Ġv ; Ġb 0 5 Ġ= Ġv ; Ġb 0 6 Ġ= Ġv ; Ġb 0 7 Ġ= Ġv ; Ġb 0 8 Ġ= Ġv ; Ġb 0 9 Ġ= Ġv ; Ġb 1 0 Ġ= Ġv ; Ġb 1 1 Ġ= Ġv ; Ġb 1 2 Ġ= Ġv ; Ġb 1 3 Ġ= Ġv ; Ġb 1 4 Ġ= Ġv ; Ġb 1 5 Ġ= Ġv ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġset All ( 1 L ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġrun () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġwriter Ġ= Ġbuffer . writer ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġx Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( x Ġ< Ġcount ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpre X Ġ= Ġx ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġsent Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( writer . offer Ġ{ Ġit . long A Ġ= Ġx ; Ġit . long B Ġ= Ġx ; Ġ}) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġx Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsent Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( x Ġ>= Ġcount Ġ|| Ġsent Ġ== Ġ 1 0 2 4 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( x Ġ> Ġpre X ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwriter . publish () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" Write Ġfinished Ġ${ System . currentTimeMillis () }") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġclass ĠRead Thread ( val Ġbuffer : ĠS pmc Pre allocated Queue < Data Holder >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcount : ĠLong , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġextra Work : ĠLong , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreader Num : ĠInt ) Ġ: ĠThread () Ġ{ ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġa 0 0 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 1 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 2 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 3 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġa 0 4 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 5 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 6 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 7 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġa 0 8 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 0 9 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 1 0 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 1 1 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġa 1 2 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 1 3 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 1 4 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġa 1 5 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġb 0 0 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 1 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 2 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 3 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġb 0 4 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 5 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 6 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 7 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġb 0 8 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 0 9 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 1 0 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 1 1 Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġinternal Ġvar Ġb 1 2 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 1 3 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 1 4 Ġ= Ġ 0 L ; Ġinternal Ġvar Ġb 1 5 Ġ= Ġ 0 L ĊĊĠĠĠĠĠĠĠ Ġpublic Ġfun ĠgetAll (): ĠLong Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġa 0 0 Ġ+ Ġa 0 1 Ġ+ Ġa 0 2 Ġ+ Ġa 0 3 Ġ+ Ġa 0 4 Ġ+ Ġa 0 5 Ġ+ Ġa 0 6 Ġ+ Ġa 0 7 Ġ+ Ġa 0 8 Ġ+ Ġa 0 9 Ġ+ Ġa 1 0 Ġ+ Ġa 1 1 Ġ+ Ġa 1 2 Ġ+ Ġa 1 3 Ġ+ Ġa 1 4 Ġ+ Ġa 1 5 Ġ+ Ġb 0 0 Ġ+ Ġb 0 1 Ġ+ Ġb 0 2 Ġ+ Ġb 0 3 Ġ+ Ġb 0 4 Ġ+ Ġb 0 5 Ġ+ Ġb 0 6 Ġ+ Ġb 0 7 Ġ+ Ġb 0 8 Ġ+ Ġb 0 9 Ġ+ Ġb 1 0 Ġ+ Ġb 1 1 Ġ+ Ġb 1 2 Ġ+ Ġb 1 3 Ġ+ Ġb 1 4 Ġ+ Ġb 1 5 ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġpublic Ġfun Ġset All ( v : ĠLong ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġa 0 0 Ġ= Ġv ; Ġa 0 1 Ġ= Ġv ; Ġa 0 2 Ġ= Ġv ; Ġa 0 3 Ġ= Ġv ; Ġa 0 4 Ġ= Ġv ; Ġa 0 5 Ġ= Ġv ; Ġa 0 6 Ġ= Ġv ; Ġa 0 7 Ġ= Ġv ; Ġa 0 8 Ġ= Ġv ; Ġa 0 9 Ġ= Ġv ; Ġa 1 0 Ġ= Ġv ; Ġa 1 1 Ġ= Ġv ; Ġa 1 2 Ġ= Ġv ; Ġa 1 3 Ġ= Ġv ; Ġa 1 4 Ġ= Ġv ; Ġa 1 5 Ġ= Ġv ; Ġb 0 0 Ġ= Ġv ; Ġb 0 1 Ġ= Ġv ; Ġb 0 2 Ġ= Ġv ; Ġb 0 3 Ġ= Ġv ; Ġb 0 4 Ġ= Ġv ; Ġb 0 5 Ġ= Ġv ; Ġb 0 6 Ġ= Ġv ; Ġb 0 7 Ġ= Ġv ; Ġb 0 8 Ġ= Ġv ; Ġb 0 9 Ġ= Ġv ; Ġb 1 0 Ġ= Ġv ; Ġb 1 1 Ġ= Ġv ; Ġb 1 2 Ġ= Ġv ; Ġb 1 3 Ġ= Ġv ; Ġb 1 4 Ġ= Ġv ; Ġb 1 5 Ġ= Ġv ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġset All ( 1 L ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġvar Ġtotal Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġvar Ġextra Ġ= Ġ 0 L ĊĊĠĠĠĠĠĠĠ Ġfun Ġdo Extra () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġy Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile ( y Ġ< Ġextra Work ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġy Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġextra Ġ+= Ġy ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġrun () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreader Ġ= Ġbuffer . get Reader ( reader Num ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġx Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( x Ġ< Ġcount ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpre X Ġ= Ġx ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġresult Ġ= Ġreader . poll () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġread Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( result Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġx Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtotal Ġ+= Ġresult . long A Ġ+ Ġresult . long B ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdo Extra () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif ( result . long A Ġ>= Ġcount Ġ- Ġbuffer . reader Count ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġend Ġcondition ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġx Ġ= Ġcount ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġread Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( read Ġ== Ġ 1 0 2 4 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġreader . poll () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( x Ġ> Ġpre X ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreader . commit () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" Read Ġfinished Ġ${ System . currentTimeMillis ()}, Ġ$ extra ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Repeated Test ( light Repeat Count ) ĊĠĠĠ Ġfun Ġmulti Thread St ress Test () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġread Thread Count Ġ= Ġ 4 ĊĠĠĠĠĠĠĠ Ġval Ġbuffer Ġ= ĠS pmc Pre allocated Queue ( 4 0 9 6 , Ġread Thread Count , Ġ{ ĠData Holder () Ġ}) ĊĠĠĠĠĠĠĠ Ġval Ġcount Ġ= Ġ 1 0 0 0 0 0 0 0 L ĊĊĠĠĠĠĠĠĠ Ġval Ġextra Work Ġ= Ġ 1 0 0 0 L ĊĠĠĠĠĠĠĠ Ġval Ġwrite Thread Ġ= ĠWrite Thread ( buffer , Ġcount ) ĊĠĠĠĠĠĠĠ Ġval Ġread Threads Ġ= Ġ( 0 Ġuntil Ġread Thread Count ). map Ġ{ ĠRead Thread ( buffer , Ġcount , Ġextra Work , Ġit ) Ġ} ĊĊĠĠĠĠĠĠĠ Ġread Threads . forEach ( Thread :: start ) ĊĠĠĠĠĠĠĠ Ġval Ġpre Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠĠĠĠĠ Ġwrite Thread . start () ĊĊĠĠĠĠĠĠĠ Ġeventually ( Duration . of Seconds ( 1 0 )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertFalse ( write Thread . is Alive ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġread Threads . forEach Ġ{ ĠassertFalse ( it . is Alive ) Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġwrite Thread . join () ĊĠĠĠĠĠĠĠ Ġread Threads . forEach ( Thread :: join ) ĊĠĠĠĠĠĠĠ Ġval Ġpost Ġ= ĠSystem . currentTimeMillis () ĊĊĠĠĠĠĠĠĠ Ġvar Ġz Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġvar Ġexpected Total Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġwhile Ġ( z Ġ< Ġcount ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpected Total Ġ+= Ġz Ġ* Ġ 2 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġz Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġlogger . info Ġ{ Ġ" T ook Ġ${ post Ġ- Ġpre } ms Ġfor Ġ$ count , Ġ${ ( count Ġ/ ĠMath . max ( 1 , Ġ( post Ġ- Ġpre ))) Ġ/ Ġ 1 0 0 0 . 0 } Ġmillion Ġper Ġsecond " Ġ} ĊĠĠĠĠĠĠĠ Ġvar Ġread Total Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġread Threads . forEach Ġ{ Ġread Total Ġ+= Ġit . total Ġ} ĊĠĠĠĠĠĠĠ ĠassertEquals ( expected Total , Ġread Total ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġclass ĠWrite Service ( override Ġval Ġworker : ĠCor outine Worker , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbuffer : ĠS pmc Pre allocated Queue < Data Holder >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcount : ĠLong , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfinished : ĠAtomicInteger ) Ġ: ĠService () Ġ{ ĊĊĠĠĠĠĠĠĠ Ġvar Ġtotal Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġoverride Ġsuspend Ġfun Ġrun () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġwriter Ġ= Ġbuffer . writer ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġx Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( x Ġ< Ġcount ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġsent Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwriter . offer Ġ{ Ġit . long A Ġ= Ġx ; Ġit . long B Ġ= Ġx Ġ} Ġ|| Ġwriter . add Async Ġ{ Ġit . long A Ġ= Ġx ; Ġit . long B Ġ= Ġx Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtotal Ġ+= Ġx Ġ* Ġ 2 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġx Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( x Ġ< Ġcount Ġ&& Ġwriter . offer Ġ{ Ġit . long A Ġ= Ġx ; Ġit . long B Ġ= Ġx Ġ}) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtotal Ġ+= Ġx Ġ* Ġ 2 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġx Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsent Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( sent Ġ== Ġ 1 0 2 4 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwriter . publish () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" write Ġdone ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfinished . increment AndGet () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfuture Ġ= ĠCor outine Future < Unit >() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfuture . await Async () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġclass ĠRead Service ( override Ġval Ġworker : ĠCor outine Worker , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbuffer : ĠS pmc Pre allocated Queue < Data Holder >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcount : ĠLong , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfinished : ĠAtomicInteger , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġextra Work : ĠLong , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreader Num : ĠInt ) Ġ: ĠService () Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġtotal Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġvar Ġextra Ġ= Ġ 0 L ĊĊĠĠĠĠĠĠĠ Ġfun Ġdo Extra () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġy Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile ( y Ġ< Ġextra Work ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġy Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġextra Ġ+= Ġy ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġsuspend Ġfun Ġrun () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreader Ġ= Ġbuffer . get Reader ( reader Num ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġx Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( x Ġ< Ġcount ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġresult : ĠData Holder ? Ġ= Ġreader . poll () Ġ?: Ġreader . await Async () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġread Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( result Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġx Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtotal Ġ+= Ġresult . long A Ġ+ Ġresult . long B ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdo Extra () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif ( result . long A Ġ>= Ġcount Ġ- Ġbuffer . reader Count ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġend Ġcondition ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġx Ġ= Ġcount ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġread Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( read Ġ== Ġ 1 0 2 4 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult Ġ= Ġreader . poll () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreader . commit () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" read Ġdone Ġ$ extra ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfinished . increment AndGet () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfuture Ġ= ĠCor outine Future < Unit >() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfuture . await Async () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġclass ĠWrite Worker ( buffer : ĠS pmc Pre allocated Queue < Data Holder >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcount : ĠLong , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfinished : ĠAtomicInteger ) Ġ: ĠCor outine Worker () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġwrite Service Ġ= ĠWrite Service ( this , Ġbuffer , Ġcount , Ġfinished ) ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġinitial Services Ġ= ĠlistOf ( write Service ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġclass ĠRead Worker ( buffer : ĠS pmc Pre allocated Queue < Data Holder >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcount : ĠLong , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfinished : ĠAtomicInteger , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġextra Work : ĠLong , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreader Num : ĠInt ) Ġ: ĠCor outine Worker () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġread Service Ġ= ĠRead Service ( this , Ġbuffer , Ġcount , Ġfinished , Ġextra Work , Ġreader Num ) ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġinitial Services Ġ= ĠlistOf ( read Service ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Repeated Test ( light Repeat Count ) ĊĠĠĠ Ġfun Ġmulti Thread Worker St ress Test () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġreader Count Ġ= Ġ 7 ĊĠĠĠĠĠĠĠ Ġval Ġbuffer Ġ= ĠS pmc Pre allocated Queue ( 4 0 9 6 , Ġreader Count , Ġ{ ĠData Holder () Ġ}) ĊĠĠĠĠĠĠĠ Ġval Ġcount Ġ= Ġ 1 0 0 0 0 0 0 0 L ĊĠĠĠĠĠĠĠ Ġval Ġextra Work Ġ= Ġ 1 0 0 0 L ĊĠĠĠĠĠĠĠ Ġval Ġfinished Ġ= ĠAtomicInteger ( 0 ) ĊĠĠĠĠĠĠĠ Ġval Ġwrite Worker Ġ= ĠWrite Worker ( buffer , Ġcount , Ġfinished ) ĊĠĠĠĠĠĠĠ Ġval Ġread Workers Ġ= Ġ( 0 Ġuntil Ġreader Count ). map Ġ{ ĠRead Worker ( buffer , Ġcount , Ġfinished , Ġextra Work , Ġit ) Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġapp Ġ= Ġobject Ġ: ĠCor outine Application < Coroutine Worker >() Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġworkers : ĠSet < Coroutine Worker > Ġ= Ġread Workers . toSet () Ġ+ Ġwrite Worker ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġpre Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠĠĠĠĠ Ġapp . start () ĊĠĠĠĠĠĠĠ Ġeventually ( Duration . of Seconds ( 3 0 0 0 ), ĠDuration . of Millis ( 1 )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 1 Ġ+ Ġread Workers . size , Ġfinished . get ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġpost Ġ= ĠSystem . currentTimeMillis () ĊĊĠĠĠĠĠĠĠ Ġapp . shutdown () ĊĠĠĠĠĠĠĠ Ġlogger . info Ġ{ Ġ" T ook Ġ${ post Ġ- Ġpre } ms Ġfor Ġ$ count , Ġ${ ( count Ġ/ ĠMath . max ( 1 , Ġ( post Ġ- Ġpre ))) Ġ/ Ġ 1 0 0 0 . 0 } Ġmillion Ġper Ġsecond " Ġ} ĊĠĠĠĠĠĠĠ Ġvar Ġread Total Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġread Workers . forEach Ġ{ Ġread Total Ġ+= Ġit . read Service . total Ġ} ĊĠĠĠĠĠĠĠ ĠassertEquals ( write Worker . write Service . total , Ġread Total ) ĊĠĠĠ Ġ} Ċ } Ċ */ Ċ