om-legion 6.9.0.5 → 6.9.0.6
raw patch · 3 files changed
+114/−64 lines, 3 filesdep ~crdt-event-foldPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: crdt-event-fold
API changes (from Hackage documentation)
Files
- om-legion.cabal +3/−2
- src/OM/Legion/RChan.hs +88/−0
- src/OM/Legion/Runtime.hs +23/−62
om-legion.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: om-legion-version: 6.9.0.5+version: 6.9.0.6 synopsis: Legion Framework. description: Framework for managing shared, replicated state across a number of homogeneous nodes.@@ -25,7 +25,7 @@ , bytestring >= 0.12.0.2 && < 0.13 , clock >= 0.8.4 && < 0.9 , containers >= 0.6.8 && < 0.7- , crdt-event-fold >= 1.8.0.2 && < 1.9+ , crdt-event-fold >= 1.8.1.0 && < 1.9 , data-default-class >= 0.1.2.0 && < 0.2 , http-api-data >= 0.6 && < 0.7 , monad-logger >= 0.3.40 && < 0.4@@ -63,6 +63,7 @@ other-modules: OM.Legion.Connection OM.Legion.MsgChan+ OM.Legion.RChan OM.Legion.Runtime -- other-extensions: hs-source-dirs: src
+ src/OM/Legion/RChan.hs view
@@ -0,0 +1,88 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++{-| Description: A specialized channel for incoming runtime messages. -}+module OM.Legion.RChan (+ RChan,+ RuntimeMessage(..),+ JoinRequest(..),+ readRChan,+ newRChan,+) where++import Control.Concurrent (MVar, newEmptyMVar, putMVar, takeMVar)+import Control.Monad.IO.Class (MonadIO(liftIO))+import Data.Binary (Binary)+import Data.ByteString.Lazy (ByteString)+import Data.CRDT.EventFold (Event(Output, State), Diff, EventFold,+ EventId)+import Data.Map (Map)+import Data.Time (DiffTime)+import GHC.Generics (Generic)+import OM.Fork (Actor(Msg, actorChan), Responder)+import OM.Legion.Connection (JoinResponse)+import OM.Legion.MsgChan (ClusterName, MessageId, Peer)+import Prelude ((.), (<$>), Maybe, Show)++{- | The type of the runtime message channel. -}+newtype RChan e = RChan {+ unRChan :: MVar (RuntimeMessage e)+ }+instance Actor (RChan e) where+ type Msg (RChan e) = RuntimeMessage e+ actorChan = putMVar . unRChan+++readRChan+ :: (MonadIO m)+ => RChan e+ -> m (RuntimeMessage e)+readRChan =+ liftIO . takeMVar . unRChan+++newRChan :: (MonadIO m) => m (RChan e)+newRChan =+ RChan <$> liftIO newEmptyMVar+++{- | The types of messages that can be sent to the runtime. -}+data RuntimeMessage e+ = ApplyFast e (Responder (Output e))+ | ApplyConsistent e (Responder (Output e))+ | Eject Peer (Responder ())+ | Merge (Diff ClusterName Peer e)+ | FullMerge (EventFold ClusterName Peer e)+ | Outputs (Map (EventId Peer) (Output e))+ | Join JoinRequest (Responder (JoinResponse e))+ | ReadState (Responder (EventFold ClusterName Peer e))+ | Call Peer ByteString (Responder ByteString)+ | Cast Peer ByteString+ | Broadcall+ DiffTime+ ByteString+ (Responder (Map Peer (Maybe ByteString)))+ | Broadcast ByteString+ | SendCallResponse Peer MessageId ByteString+ | HandleCallResponse Peer MessageId ByteString+ | Resend (Responder ())+ | GetStats (Responder (EventFold ClusterName Peer e))+deriving stock instance+ ( Show e+ , Show (Output e)+ , Show (State e)+ )+ =>+ Show (RuntimeMessage e)+++{- | This is the type of a join request message. -}+newtype JoinRequest = JoinRequest Peer+ deriving stock (Generic, Show)+instance Binary JoinRequest++
src/OM/Legion/Runtime.hs view
@@ -39,8 +39,7 @@ ) where import Control.Arrow (Arrow((&&&)))-import Control.Concurrent (Chan, newChan, readChan, threadDelay,- writeChan)+import Control.Concurrent (threadDelay) import Control.Exception.Safe (MonadCatch, tryAny) import Control.Monad (unless, void, when) import Control.Monad.IO.Class (MonadIO(liftIO))@@ -55,8 +54,8 @@ import Data.Binary (Binary) import Data.ByteString.Lazy (ByteString) import Data.CRDT.EventFold (Event(Output, State),- UpdateResult(urEventFold, urOutputs), Diff, EventFold, EventId,- infimumId, projParticipants)+ UpdateResult(urEventFold, urOutputs), EventFold, EventId, diffSize,+ events, infimumId, projParticipants) import Data.CRDT.EventFold.Monad (MonadUpdateEF(diffMerge, disassociate, event, fullMerge, participate), EventFoldT, runEventFoldT) import Data.Default.Class (Default)@@ -76,6 +75,11 @@ EventConstraints, disconnect, peerMessagePort, sendPeer) import OM.Legion.MsgChan (MessageId(M), Peer(unPeer), PeerMessage(PMCall, PMCallResponse, PMCast, PMFullMerge, PMMerge, PMOutputs), ClusterName)+import OM.Legion.RChan (JoinRequest(JoinRequest),+ RuntimeMessage(ApplyConsistent, ApplyFast, Broadcall, Broadcast,+ Call, Cast, Eject, FullMerge, GetStats, HandleCallResponse, Join,+ Merge, Outputs, ReadState, Resend, SendCallResponse), RChan, newRChan,+ readRChan) import OM.Logging (withPrefix) import OM.Show (showj, showt) import OM.Socket (AddressDescription(AddressDescription), connectServer,@@ -85,7 +89,7 @@ Eq((/=)), Functor(fmap), Maybe(Just, Nothing), Monad((>>), (>>=), return), Monoid(mempty), Ord((<=), (>=)), Semigroup((<>)), ($), (.), (<$>), (=<<), IO, Int, MonadFail, Show, String, fst, mapM_, maybe,- seq, sequence_)+ otherwise, seq, sequence_) import System.Clock (TimeSpec) import System.Random.Shuffle (shuffleM) import qualified Data.Binary as Binary@@ -141,7 +145,7 @@ = do logInfo $ "Starting up with the following Mode: " <> showt startupMode rts <- makeRuntimeState notify startupMode- runtimeChan <- RChan <$> liftIO newChan+ runtimeChan <- newRChan logging <- withPrefix (logPrefix (rsSelf rts)) <$> askLoggerIO rStats <- liftIO $ newIORef mempty (`runLoggingT` logging) $@@ -204,15 +208,6 @@ Binary.put (diffTimeToPicoseconds <$> timeWithoutProgress) -{- | The type of the runtime message channel. -}-newtype RChan e = RChan {- unRChan :: Chan (RuntimeMessage e)- }-instance Actor (RChan e) where- type Msg (RChan e) = RuntimeMessage e- actorChan = writeChan . unRChan-- {- | Update the distributed cluster state by applying an event. The event output will be returned immediately and may not reflect a totally@@ -298,36 +293,6 @@ getSelf = rSelf -{- | The types of messages that can be sent to the runtime. -}-data RuntimeMessage e- = ApplyFast e (Responder (Output e))- | ApplyConsistent e (Responder (Output e))- | Eject Peer (Responder ())- | Merge (Diff ClusterName Peer e)- | FullMerge (EventFold ClusterName Peer e)- | Outputs (Map (EventId Peer) (Output e))- | Join JoinRequest (Responder (JoinResponse e))- | ReadState (Responder (EventFold ClusterName Peer e))- | Call Peer ByteString (Responder ByteString)- | Cast Peer ByteString- | Broadcall- DiffTime- ByteString- (Responder (Map Peer (Maybe ByteString)))- | Broadcast ByteString- | SendCallResponse Peer MessageId ByteString- | HandleCallResponse Peer MessageId ByteString- | Resend (Responder ())- | GetStats (Responder (EventFold ClusterName Peer e))-deriving stock instance- ( Show e- , Show (Output e)- , Show (State e)- )- =>- Show (RuntimeMessage e)-- {- | Execute the Legion runtime, with the given user definitions, and framework settings. This function never returns (except maybe with an@@ -389,7 +354,7 @@ let -- handleMessages :: StateT (RuntimeState e3) m Void handleMessages = do- msg <- liftIO $ readChan (unRChan runtimeChan)+ msg <- readRChan runtimeChan RuntimeState {rsClusterState = cluster1} <- get logDebug $ "Handling: " <> showt msg handleRuntimeMessage msg@@ -439,7 +404,7 @@ PMCallResponse source mid responseMsg -> Stream.yield (HandleCallResponse source mid responseMsg) )- & Stream.mapM_ (liftIO . writeChan (unRChan runtimeChan))+ & Stream.mapM_ (Fork.cast runtimeChan) runJoinListener :: ( MonadCatch m@@ -821,15 +786,17 @@ liftIO (shuffleM (Set.toList targets)) >>= \case [] -> return () target:_ ->- -- case events target cluster of- -- Nothing -> do- -- logInfo- -- $ "Sending full merge because the target's join event "- -- <> "has not reaached the infimum"- -- sendPeer (PMFullMerge cluster) target- -- Just diff ->- -- sendPeer (PMMerge diff) target- sendPeer (PMFullMerge cluster) target+ case events target cluster of+ Nothing -> do+ logInfo+ $ "Sending full merge because the target's join event "+ <> "has not reaached the infimum"+ sendPeer (PMFullMerge cluster) target+ Just diff+ | diffSize diff <= 0 ->+ sendPeer (PMMerge diff) target+ | otherwise ->+ sendPeer (PMFullMerge cluster) target disconnectObsolete where {- |@@ -966,12 +933,6 @@ , rsNotify = notify self , rsJoins = mempty }---{- | This is the type of a join request message. -}-newtype JoinRequest = JoinRequest Peer- deriving stock (Generic, Show)-instance Binary JoinRequest {- |