package Ġmonitor Ċ Ċ import Ġmonitor . State .* Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . locks . Re entrant Lock Ċ import Ġkotlin . concurrent . with Lock ĊĊ Ċ enum Ġclass ĠState Ġ{ ĊĠĠĠ ĠRELEASE D , ĊĠĠĠ ĠAC QUI RING _ LOCK , ĊĠĠĠ ĠIN SIDE _ CRITICAL _ SECTION , ĊĠĠĠ ĠWAIT ING Ċ } Ċ Ċ class ĠDistributed Lock ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġcluster : ĠCluster Socket , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġthis NodeId : ĠNode Id Ċ ) Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlock Ġ= ĠRe entrant Lock () ĊĊĠĠĠ Ġprivate Ġvar Ġstate Ġ= ĠRELEASE D ĊĊĠĠĠ Ġprivate Ġval Ġqueue : ĠQueue < Request > Ġ= ĠPriority Queue () ĊĊĠĠĠ Ġprivate Ġval Ġconditions Ġ= Ġmutable ListOf < Distributed Condition >() ĊĊĠĠĠ Ġprivate Ġval Ġfirst In Queue Cond Ġ= Ġlock . new Condition () ĊĊĠĠĠ Ġfun Ġacquire () Ġ{ ĊĠĠĠĠĠĠĠ Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstate Ġ= ĠAC QUI RING _ LOCK ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġcluster . broadcast ( Message ( MessageType . REQUEST )) ĊĠĠĠĠĠĠĠ Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġensure First In Queue () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstate Ġ= ĠIN SIDE _ CRITICAL _ SECTION ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" Lock Ġacquired ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġensure First In Queue () Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile Ġ( queue . first OrNull ()?. nodeId Ġ!= Ġthis NodeId ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfirst In Queue Cond . await () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġrelease () Ġ= Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( state Ġ!= ĠIN SIDE _ CRITICAL _ SECTION ) Ġthrow ĠAssertionError () ĊĠĠĠĠĠĠĠ Ġcluster . broadcast ( Message ( MessageType . RELEASE )) ĊĠĠĠĠĠĠĠ Ġprintln (" Lock Ġreleased ") ĊĠĠĠĠĠĠĠ Ġstate Ġ= ĠRELEASE D ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġnew Condition (): ĠDistributed Condition Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcondition Id Ġ= Ġconditions . size ĊĠĠĠĠĠĠĠ Ġreturn ĠDistributed Condition ( cluster , Ġthis , Ġcondition Id ). also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconditions . add ( it ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinternal Ġfun Ġawait () Ġ= Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠ Ġstate Ġ= ĠWAIT ING ĊĠĠĠĠĠĠĠ Ġensure First In Queue () ĊĠĠĠĠĠĠĠ Ġstate Ġ= ĠIN SIDE _ CRITICAL _ SECTION ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinternal Ġfun Ġhandle Message ( message : ĠCluster Message ) Ġ= Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠ Ġwhen Ġ( message . content . type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMessageType . REQUEST Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġqueue . add ( Request ( message . timestamp , Ġmessage . nodeId )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMessageType . RELEASE Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( state Ġ== ĠIN SIDE _ CRITICAL _ SECTION ) Ġthrow ĠAssertionError () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Release ( message . nodeId ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMessageType . WAIT Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(( message . nodeId Ġ!= Ġthis NodeId Ġ&& Ġstate Ġ== ĠIN SIDE _ CRITICAL _ SECTION ) Ġ|| ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ( message . nodeId Ġ== Ġthis NodeId Ġ&& Ġstate Ġ!= ĠIN SIDE _ CRITICAL _ SECTION )) Ġthrow ĠAssertionError () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcondition Ġ= Ġconditions [ message . content . condition Id ] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcondition . wait Queue . add ( Request ( message . timestamp , Ġmessage . nodeId )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Release ( message . nodeId ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMessageType . NOTIFY Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(( message . nodeId Ġ!= Ġthis NodeId Ġ&& Ġstate Ġ== ĠIN SIDE _ CRITICAL _ SECTION ) Ġ|| ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ( message . nodeId Ġ== Ġthis NodeId Ġ&& Ġstate Ġ!= ĠIN SIDE _ CRITICAL _ SECTION )) Ġthrow ĠAssertionError () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcondition Ġ= Ġconditions [ message . content . condition Id ] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcondition . wait Queue . poll ()?. let Ġ{ Ġwait Request Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġqueue . add ( Request ( message . timestamp , Ġwait Request . nodeId )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġthrow ĠAssertionError () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprintln (" Queue : Ġ$ queue ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġon Release ( nodeId : ĠNode Id ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( queue . first OrNull ()?. nodeId Ġ== ĠnodeId ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġqueue . remove () ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( queue . none Ġ{ Ġit . nodeId Ġ== ĠnodeId Ġ}) Ġthrow ĠAssertionError () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġqueue . remove If Ġ{ Ġit . nodeId Ġ== ĠnodeId Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfirst In Queue Cond . signal () ĊĠĠĠ Ġ} Ċ } Ċ