libraft 0.2.1.0 → 0.3.0.0
raw patch · 22 files changed
+850/−511 lines, 22 filesdep +asyncdep ~dejafu
Dependencies added: async
Dependency ranges changed: dejafu
Files
- ChangeLog.md +14/−0
- README.md +108/−3
- app/Main.hs +66/−123
- libraft.cabal +12/−4
- src/Control/Concurrent/STM/Timer.hs +17/−20
- src/Examples/Raft/FileStore/Log.hs +50/−7
- src/Examples/Raft/FileStore/Persistent.hs +63/−6
- src/Examples/Raft/Socket/Client.hs +1/−2
- src/Examples/Raft/Socket/Node.hs +62/−19
- src/Raft.hs +72/−130
- src/Raft/Candidate.hs +1/−1
- src/Raft/Client.hs +15/−11
- src/Raft/Config.hs +1/−1
- src/Raft/Follower.hs +1/−1
- src/Raft/Handle.hs +1/−1
- src/Raft/Leader.hs +1/−1
- src/Raft/Log/PostgreSQL.hs +36/−12
- src/Raft/Monad.hs +106/−155
- src/Raft/StateMachine.hs +6/−3
- src/Raft/Transition.hs +192/−0
- test/TestDejaFu.hs +20/−6
- test/TestUtils.hs +5/−5
ChangeLog.md view
@@ -1,5 +1,19 @@ # Changelog for raft +## 0.3.0.0++- API change: `runRaftNode` now requires the monad it runs in to provide an+ instance of the `MonadRaftAsync` and `MonadRaftChan` in the `Raft.Monad`+ module in lieu of the previously necessary `MonadConc` instance.+- API change: Removed the `MonadConc` constraint on all example monad+ transformers and implemented inherited `MonadRaftChan` and `MonadRaftAsync` + instances +- API change: Removed the `MonadConc` const+- API change: Renamed the old `Raft.Monad` module to `Raft.Transtion` and moved the + `RaftT` monad transformer from `Raft.hs` to the `Raft.Monad` module+- Improvement: Rework example monad `RaftExampleT` for simplicity++ ## 0.2.0.0 - Feature: Client requests are now cached by the current leader such that duplicate
README.md view
@@ -190,12 +190,27 @@ ### Effectful Layers -In the protocol, there are two main components that need access to global+In this Raft implementation, there are four components that need access to global state and system resources. Firstly, raft nodes must maintain some persistent state for efficient and correct recovery from network outages or partitions. Secondly, raft nodes need to send messages to other raft nodes for the network-(the replicated state machine) to be operational.+(the replicated state machine) to be operational. Next, library users must+specify what event channel datastructure to use. Finally, the programmer must+provide a way to fork threads and run a list of effectful actions concurrently. +The reasons for the latter two design decisions-- requiring the programmer to+provide an event channel type and new/read/writeChannel primitives, a way to+fork an effectful action to run concurrently, and a way to run a list of actions+concurrently-- is a result of property based concurrency testing that we do,+found in `test/TestDejaFu.hs`. In order to test the system as a whole, to run+several nodes concurrently and test invariants about the system such as the+absence of deadlocks, we must be able to swap out the base monad for the+`ConcIO` monad, which has implementations of concurrency primitives that act+deterministically. This allows us to test that the raft nodes run correctly in +a wide space of thread interleavings giving us more confidence that our code is+correct, assuming "correct" implementations of the `MonadRaftAsync` and+`MonadRaftChan` typeclasses.+ #### Persistent State Each node persists data to disk, including the replicated log@@ -249,6 +264,8 @@ => m (Either (RaftReadLogError m) (Maybe (Entry v))) ``` +---+ To read and write the `PersistentData` type (the remaining persistent data that is not log entries), we ask the user to use the following `RaftPersist` typeclass.@@ -312,7 +329,95 @@ We have written a default implementation for network sockets over TCP in [src/Examples/Raft/Socket](https://github.com/adjoint-io/raft/blob/master/src/Examples/Raft/Socket) -# Run example+### Event Channel++The core of the effectful layers of this Raft implmentation is the _event+channel_. Since different data channels have different performance, we ask the+programmer to supply an implementation of such a channel via yet another type+class and type family:++```haskell+class Monad m => MonadRaftChan v m where+ type RaftEventChan v m+ readRaftChan :: RaftEventChan v m -> m (Event v)+ writeRaftChan :: RaftEventChan v m -> Event v -> m ()+ newRaftChan :: m (RaftEventChan v m)+```++On spawning a raft node, the program will create a new event channel using+`newRaftChan`. Then, the event producers will be forked; These event producers+will use the aforementioned typeclasses like `RaftRecvRPC` and `RaftRecvClient`+to wait for messages from other raft nodes or clients wishing to contact the+node. Once a message is received, a message event is constructed from the+message contents and written to the main event channel via `writeRaftChan`. In+the main thread, the core event handler will be repeatedly reading events from+the event channel using `readRaftChan` and performing the correct action in+response to each event.++### Concurrency++The last of the type class instances the programmer must provide for the monad+they are running the raft node in is a `MonadRaftAsync`, which provides the main+raft loop with a small set of concurrency primitives; The raft node needs to+know how to fork actions in the monad, and how to run a list of actions+concurrently, waiting for all to finish. The former is necessary for the raft+node to be able to fork its event producers, and the latter is required in order+to respond to clients and other raft nodes in a timely manner. The typeclass is+defined as follows:++```haskell+class Monad m => MonadRaftAsync m where+ type RaftAsync m :: * -> *+ raftAsync :: m a -> m (RaftAsync m a)+ raftMapConcurrently :: Traversable t => (a -> m b) -> t a -> m (t b)+ raftMapConcurrently_ :: Foldable t => (a -> m b) -> t a -> m ()+```++The implementation of this typeclass is a bit subtle, and it is advised that+programmers do not implement it from scratch themselves. A default +implementation is provided for the `IO` type using standard concurrency +primitives making it easy for the programmer to simply rely on that+implementation. An example instance of this type class for a custom monad+transformer stack with `IO` the bottom:++```haskell+instance MonadRaftAsync MyMonad where+ type RaftAsync MyMonad = RaftAsync IO+ raftAsync myMonad = lift $ raftAsync (runMyMonad myMonad) + raftMapConcurrently f as = lift (raftMapConcurrently (runMyMonad . f) as)+ raftMapConcurrently_ f as = lift (raftMapConcurrently_ (runMyMonad . f) as)+```++# The Raft Example (`raft-example`)++In this library we provide a full fledged, non-production ready, example+implementation/s of monad transformers and type class instances for _all_ type+classes necessary to run a raft node. They can be found in+`src/Examples/Raft/...` or in `app/Main.hs`:++`RaftExampleT` (found in `app/Main.hs`):+- `MonadRaftAsync`+- `MonadRaftChan`+- `RaftStateMachine`++`RaftSocketT` (found in `src/Examples/Raft/Socket/Node.hs`):+- `RaftSendRPC`+- `RaftRecvRPC`+- `RaftSendClient`+- `RaftRecvClient`++`RaftLogFileStoreT` (found in `src/Examples/Raft/FileStore/Log.hs`):+- `RaftReadLog`+- `RaftWriteLog`+- `RaftDeleteLog`+- `RaftInitLog`++`RaftPersistFileStoreT` (found in `src/Examples/Raft/FileStore/Persistent.hs`):+- `RaftPersistent`++Programmers can use these files and implementations for references when implementing +the necessary type class instances for their bespoke monads, or even use some of+the monad transformers in their own stack! We provide a complete example of the library where nodes communicate via network sockets, and they write their logs on text files. See
app/Main.hs view
@@ -2,25 +2,23 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE TypeApplications #-} {-# LANGUAGE ScopedTypeVariables #-} module Main where -import Protolude hiding- ( MVar, putMVar, takeMVar, newMVar, newEmptyMVar, readMVar- , atomically, STM(..), Chan, newTVar, readTVar, writeTVar- , newChan, writeChan, readChan- , threadDelay, killThread, TVar(..)- , catch, handle, takeWhile, takeWhile1, (<|>)- , lift- )+import Protolude -import Control.Concurrent.Classy hiding (catch)+import Control.Concurrent.Lifted (fork)+import Control.Concurrent.STM.TChan+ import Control.Monad.Fail import Control.Monad.Catch import Control.Monad.Trans.Class@@ -28,7 +26,7 @@ import qualified Data.Map as Map import qualified Data.List as L import qualified Data.Set as Set-import qualified Data.Serialize as S+import Data.Serialize (Serialize) import Numeric.Natural @@ -41,6 +39,7 @@ import Raft.Config import Raft.Log import Raft.Log.PostgreSQL+import Raft.Monad import Raft.Client import Database.PostgreSQL.Simple@@ -63,9 +62,8 @@ data StoreCmd = Set Var Natural | Incr Var- deriving (Show, Generic)--instance S.Serialize StoreCmd+ deriving stock (Show, Generic)+ deriving anyclass Serialize type Store = Map Var Natural @@ -78,85 +76,32 @@ Set x n -> Map.insert x n store Incr x -> Map.adjust succ x store -instance (Monad m, sm ~ Store, v ~ StoreCmd, RaftStateMachinePure sm v) => RaftStateMachine (RaftExampleM m sm v) sm v where- validateCmd _ = pure (Right ())- askRaftStateMachinePureCtx = pure ()------------------------- Raft instances -------------------------data NodeEnv sm = NodeEnv- { nEnvStore :: TVar (STM IO) sm- , nEnvNodeId :: NodeId- }--newtype RaftExampleM m sm v a = RaftExampleM {- unRaftExampleM :: ReaderT (NodeEnv sm) (RaftSocketT v (RaftPersistFileStoreT m)) a- }--deriving instance Functor m => Functor (RaftExampleM m sm v)-deriving instance Applicative m => Applicative (RaftExampleM m sm v)-deriving instance Monad m => Monad (RaftExampleM m sm v)-deriving instance MonadIO m => MonadIO (RaftExampleM m sm v)-deriving instance MonadFail m => MonadFail (RaftExampleM m sm v)-deriving instance Monad m => MonadReader (NodeEnv sm) (RaftExampleM m sm v)-deriving instance Alternative m => Alternative (RaftExampleM m sm v)-deriving instance MonadPlus m => MonadPlus (RaftExampleM m sm v)--deriving instance MonadThrow m => MonadThrow (RaftExampleM m sm v)-deriving instance MonadCatch m => MonadCatch (RaftExampleM m sm v)-deriving instance MonadMask m => MonadMask (RaftExampleM m sm v)-deriving instance MonadConc m => MonadConc (RaftExampleM m sm v)--runRaftExampleM- :: (MonadIO m, MonadConc m)- => NodeEnv sm- -> NodeSocketEnv v- -> RaftPersistFile- -> RaftExampleM m sm v a- -> m a-runRaftExampleM nodeEnv nodeSocketEnv raftPersistFile raftExampleM =- flip runReaderT raftPersistFile . unRaftPersistFileStoreT $- flip runReaderT nodeSocketEnv . unRaftSocketT $- flip runReaderT nodeEnv $ unRaftExampleM raftExampleM--instance (MonadIO m, MonadConc m) => RaftSendClient (RaftExampleM m Store StoreCmd) Store StoreCmd where- sendClient cid msg = (RaftExampleM . lift) $ sendClient cid msg--instance (MonadIO m, MonadConc m) => RaftRecvClient (RaftExampleM m Store StoreCmd) StoreCmd where- type RaftRecvClientError (RaftExampleM m Store StoreCmd) StoreCmd = Text- receiveClient = RaftExampleM $ lift receiveClient--instance (MonadIO m, MonadConc m) => RaftSendRPC (RaftExampleM m Store StoreCmd) StoreCmd where- sendRPC nid msg = (RaftExampleM . lift) $ sendRPC nid msg+--------------------------------------------------------------------------------+-- Raft Example Monad Transformer+-------------------------------------------------------------------------------- -instance (MonadIO m, MonadConc m) => RaftRecvRPC (RaftExampleM m Store StoreCmd) StoreCmd where- type RaftRecvRPCError (RaftExampleM m Store StoreCmd) StoreCmd = Text- receiveRPC = RaftExampleM $ lift receiveRPC+newtype RaftExampleT m a = RaftExampleT { unRaftExampleT :: m a }+ deriving newtype (Functor, Applicative, Monad, MonadIO, MonadFail, MonadMask, MonadCatch, MonadThrow) -instance (MonadIO m, MonadConc m) => RaftPersist (RaftExampleM m Store StoreCmd) where- type RaftPersistError (RaftExampleM m Store StoreCmd) = RaftPersistFileStoreError- initializePersistentState = RaftExampleM $ lift $ lift initializePersistentState- writePersistentState ps = RaftExampleM $ lift $ lift $ writePersistentState ps- readPersistentState = RaftExampleM $ lift $ lift $ readPersistentState+instance MonadTrans RaftExampleT where+ lift = RaftExampleT -instance (MonadConc m, RaftInitLog m StoreCmd) => RaftInitLog (RaftExampleM m Store StoreCmd) StoreCmd where- type RaftInitLogError (RaftExampleM m Store StoreCmd) = RaftInitLogError m- initializeLog p = RaftExampleM $ lift $ lift $ lift $ initializeLog p+instance (Monad m, RaftStateMachinePure Store StoreCmd) => RaftStateMachine (RaftExampleT m) Store StoreCmd where+ validateCmd _ = pure (Right ())+ askRaftStateMachinePureCtx = pure () -instance RaftWriteLog m StoreCmd => RaftWriteLog (RaftExampleM m Store StoreCmd) StoreCmd where- type RaftWriteLogError (RaftExampleM m Store StoreCmd) = RaftWriteLogError m- writeLogEntries entries = RaftExampleM $ lift $ lift $ lift $ writeLogEntries entries+instance MonadRaftChan v m => MonadRaftChan v (RaftExampleT m) where+ type RaftEventChan v (RaftExampleT m) = RaftEventChan v m+ readRaftChan = lift . readRaftChan+ writeRaftChan chan = lift . writeRaftChan chan+ newRaftChan = lift (newRaftChan @v @m) -instance RaftReadLog m StoreCmd => RaftReadLog (RaftExampleM m Store StoreCmd) StoreCmd where- type RaftReadLogError (RaftExampleM m Store StoreCmd) = RaftReadLogError m- readLogEntry idx = RaftExampleM $ lift $ lift $ lift $ readLogEntry idx- readLastLogEntry = RaftExampleM $ lift $ lift $ lift $ readLastLogEntry+instance (MonadIO m, MonadRaftFork m) => MonadRaftFork (RaftExampleT m) where+ type RaftThreadId (RaftExampleT m) = RaftThreadId m+ raftFork m = lift $ raftFork (runRaftExampleT m) -instance RaftDeleteLog m StoreCmd => RaftDeleteLog (RaftExampleM m Store StoreCmd) StoreCmd where- type RaftDeleteLogError (RaftExampleM m Store StoreCmd) = RaftDeleteLogError m- deleteLogEntriesFrom idx = RaftExampleM $ lift $ lift $ lift $ deleteLogEntriesFrom idx+runRaftExampleT :: RaftExampleT m a -> m a+runRaftExampleT = unRaftExampleT -------------------- -- Client console --@@ -177,7 +122,7 @@ newtype ConsoleM a = ConsoleM { unConsoleM :: HaskelineT (RS.RaftSocketClientM Store StoreCmd) a- } deriving (Functor, Applicative, Monad, MonadIO)+ } deriving newtype (Functor, Applicative, Monad, MonadIO) liftRSCM = ConsoleM . lift @@ -222,7 +167,7 @@ main :: IO () main = do- args <- (toS <$>) <$> getArgs+ args :: [ByteString] <- fmap toS <$> getArgs case args of ["client"] -> clientMain ("node":"fresh":"file":nid:nids) -> initNode New FileStore (nid:nids)@@ -236,44 +181,51 @@ New -> cleanStorage nodeDir storageType Existing -> pure () nSocketEnv <- initSocketEnv nid- nEnv <- initNodeEnv nid nPersistFile <- RaftPersistFile <$> persistentFilePath nid++ let (host, port) = RS.nidToHostPort (toS nid)+ -- Launch the server receiving connections from raft other nodes+ fork $ flip runReaderT nSocketEnv . unRaftSocketT $+ acceptConnections host port+ case storageType of FileStore -> do nLogsFile <- RaftLogFile <$> logsFilePath nid- runRaftLogFileStoreT nLogsFile $- runRaftNode' nSocketEnv nEnv nPersistFile+ runRaftExampleT $+ runRaftLogFileStoreT nLogsFile $+ runRaftNode' nSocketEnv nPersistFile PostgreSQL dbName -> do let pgConnInfo = raftDatabaseConnInfo "libraft_test" "libraft_test" dbName- runRaftPostgresM pgConnInfo $- runRaftNode' nSocketEnv nEnv nPersistFile+ runRaftExampleT $+ runRaftPostgresT pgConnInfo $+ runRaftNode' nSocketEnv nPersistFile where runRaftNode'- :: ( MonadIO m, MonadConc m, MonadFail m- , RaftInitLog m StoreCmd, RaftReadLog m StoreCmd, RaftWriteLog m StoreCmd- , RaftDeleteLog m StoreCmd, Exception (RaftInitLogError m), Exception (RaftReadLogError m)- , Exception (RaftWriteLogError m), Exception (RaftDeleteLogError m), Typeable m+ :: ( MonadIO m, MonadRaft StoreCmd m, MonadFail m, MonadMask m+ , MonadRaft StoreCmd m, RaftStateMachine m Store StoreCmd+ , RaftInitLog m StoreCmd, RaftReadLog m StoreCmd, RaftWriteLog m StoreCmd, RaftDeleteLog m StoreCmd+ , Exception (RaftInitLogError m), Exception (RaftReadLogError m)+ , Exception (RaftWriteLogError m), Exception (RaftDeleteLogError m)+ , Typeable m ) => NodeSocketEnv StoreCmd- -> NodeEnv Store -> RaftPersistFile -> m ()- runRaftNode' nSocketEnv nEnv nPersistFile =- runRaftExampleM nEnv nSocketEnv nPersistFile $ do- let allNodeIds = Set.fromList (nid : nids)- let (host, port) = RS.nidToHostPort (toS nid)- let nodeConfig = NodeConfig- { configNodeId = toS nid- , configNodeIds = allNodeIds- -- These are recommended timeouts from the original- -- raft paper and the ARC report.- , configElectionTimeout = (150000, 300000)- , configHeartbeatTimeout = 50000- , configStorageState = storageState- }- fork $ RaftExampleM $ lift (acceptConnections host port)- electionTimerSeed <- liftIO randomIO- runRaftNode nodeConfig (LogCtx LogStdout Debug) electionTimerSeed (mempty :: Store)+ runRaftNode' nSocketEnv nPersistFile =+ runRaftSocketT nSocketEnv $+ runRaftPersistFileStoreT nPersistFile $ do+ let allNodeIds = Set.fromList (nid : nids)+ let nodeConfig = RaftNodeConfig+ { configNodeId = toS nid+ , configNodeIds = allNodeIds+ -- These are recommended timeouts from the original+ -- raft paper and the ARC report.+ , configElectionTimeout = (150000, 300000)+ , configHeartbeatTimeout = 50000+ , configStorageState = storageState+ }+ electionTimerSeed <- liftIO randomIO+ runRaftNode nodeConfig (LogCtx LogStdout Debug) electionTimerSeed (mempty :: Store) cleanStorage :: FilePath -> LogStorage -> IO () cleanStorage nodeDir ls = do@@ -294,15 +246,6 @@ logsFilePath nid = do tmpDir <- Directory.getTemporaryDirectory pure (tmpDir ++ "/" ++ toS nid ++ "/" ++ "logs")-- initNodeEnv :: NodeId -> IO (NodeEnv Store)- initNodeEnv nid = do- let (host, port) = RS.nidToHostPort (toS nid)- storeTVar <- atomically (newTVar mempty)- pure NodeEnv- { nEnvStore = storeTVar- , nEnvNodeId = toS host <> ":" <> toS port- } initSocketEnv :: NodeId -> IO (NodeSocketEnv v) initSocketEnv nid = do
libraft.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 81b50bfa684188cd50f291971fbe81bd55d41e2066f5c5b43dd9b98f5b21e2da+-- hash: 2eb0f5597b5304ede9da9c1a97cc9bc16e5df0eb384c9f1ce56e8b1dc122f5b9 name: libraft-version: 0.2.1.0+version: 0.3.0.0 synopsis: Raft consensus algorithm description: Please see the README on GitHub at <https://github.com/adjoint-io/raft#readme> category: Distributed Systems@@ -52,6 +52,7 @@ Raft.Persistent Raft.RPC Raft.StateMachine+ Raft.Transition Raft.Types other-modules: Paths_libraft@@ -60,7 +61,8 @@ default-extensions: NoImplicitPrelude OverloadedStrings LambdaCase ghc-options: -fwarn-unused-binds -fwarn-unused-imports build-depends:- atomic-write+ async+ , atomic-write , attoparsec , base >=4.7 && <5 , base16-bytestring@@ -69,6 +71,7 @@ , concurrency >=1.3.0.0 && <1.7.0.0 , containers , cryptohash-sha256+ , dejafu , directory , exceptions , file-embed@@ -83,6 +86,7 @@ , protolude , random , repline+ , stm , text , time , transformers@@ -99,7 +103,8 @@ default-extensions: NoImplicitPrelude OverloadedStrings LambdaCase ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends:- atomic-write+ async+ , atomic-write , attoparsec , base >=4.7 && <5 , base16-bytestring@@ -108,6 +113,7 @@ , concurrency >=1.3.0.0 && <1.7.0.0 , containers , cryptohash-sha256+ , dejafu , directory , exceptions , file-embed@@ -145,6 +151,7 @@ ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: QuickCheck+ , async , atomic-write , attoparsec , base >=4.7 && <5@@ -173,6 +180,7 @@ , quickcheck-state-machine , random , repline+ , stm , tasty , tasty-dejafu , tasty-discover
src/Control/Concurrent/STM/Timer.hs view
@@ -8,51 +8,49 @@ waitTimer, ) where -import Protolude hiding (wait, async, withAsync, cancel, Async, STM, killThread, ThreadId, threadDelay, myThreadId, atomically)+import Protolude -import Control.Monad.Conc.Class-import Control.Concurrent.Classy.STM-import Control.Concurrent.Classy.Async+import Control.Concurrent.Async+import Control.Concurrent.STM import System.Random (StdGen, randomR, mkStdGen) import Numeric.Natural -data Timer m = Timer- { timerAsync :: TMVar (STM m) (Async m ())+data Timer = Timer+ { timerAsync :: TMVar (Async ()) -- ^ The async computation of the timer- , timerLock :: TMVar (STM m) ()+ , timerLock :: TMVar () -- ^ When the TMVar is empty, the timer is being used- , timerGen :: TVar (STM m) StdGen+ , timerGen :: TVar StdGen , timerRange :: (Natural, Natural)- , timerAction :: m () } -- | Create a new timer with the supplied timer action and timer length,-newTimer :: MonadConc m => m () -> Natural -> m (Timer m)-newTimer action timeout = newTimerRange action 0 (timeout, timeout)+newTimer :: Natural -> IO Timer+newTimer timeout = newTimerRange 0 (timeout, timeout) -- | Create a new timer with the supplied timer action, random seed, and range -- from which the the timer will choose a random timer length at each -- start or reset.-newTimerRange :: MonadConc m => m () -> Int -> (Natural, Natural) -> m (Timer m)-newTimerRange action seed timeoutRange = do+newTimerRange :: Int -> (Natural, Natural) -> IO Timer+newTimerRange seed timeoutRange = do (timerAsync, timerLock, timerGen) <- atomically $ (,,) <$> newEmptyTMVar <*> newTMVar () <*> newTVar (mkStdGen seed)- pure $ Timer timerAsync timerLock timerGen timeoutRange action+ pure $ Timer timerAsync timerLock timerGen timeoutRange -------------------------------------------------------------------------------- -- | Start the timer. If the timer is already running, the timer is not started. -- Returns True if the timer was succesfully started.-startTimer :: MonadConc m => Timer m -> m Bool+startTimer :: Timer -> IO Bool startTimer timer = do- mlock <- atomically $ tryTakeTMVar (timerLock timer)+ mlock <- liftIO . atomically $ tryTakeTMVar (timerLock timer) case mlock of Nothing -> pure False Just () -> resetTimer timer >> pure True -- | Resets the timer with a new random timeout.-resetTimer :: MonadConc m => Timer m -> m ()+resetTimer :: Timer -> IO () resetTimer timer = do -- Check if a timer is already running. If it is, asynchronously kill the@@ -67,7 +65,6 @@ -- timer finishing. ta <- async $ do threadDelay =<< randomDelay timer- timerAction timer success <- atomically $ tryPutTMVar (timerLock timer) () when (not success) $ panic "[Failed Invariant]: Putting the timer lock back should succeed"@@ -80,12 +77,12 @@ void $ async (uninterruptibleCancel ta) -- | Wait for a timer to complete-waitTimer :: MonadConc m => Timer m -> m ()+waitTimer :: Timer -> IO () waitTimer timer = atomically $ readTMVar (timerLock timer) -------------------------------------------------------------------------------- -randomDelay :: MonadConc m => Timer m -> m Int+randomDelay :: Timer -> IO Int randomDelay timer = atomically $ do g <- readTVar (timerGen timer) let (tmin, tmax) = timerRange timer
src/Examples/Raft/FileStore/Log.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE TypeApplications #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}@@ -7,14 +8,12 @@ {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE StandaloneDeriving #-}-{-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE ScopedTypeVariables #-} module Examples.Raft.FileStore.Log where import Protolude hiding (try) -import Control.Concurrent.Classy hiding (catch, ThreadId) import Control.Monad.Fail import Control.Monad.Catch import Control.Monad.Trans.Class@@ -27,6 +26,11 @@ import qualified System.AtomicWrite.Writer.ByteString as AW import Raft.Log+import Raft.Monad+import Raft.StateMachine+import Raft.RPC+import Raft.Client+import Raft.Persistent import Raft.Types newtype RaftLogFileStoreError = RaftLogFileStoreError Text@@ -42,12 +46,11 @@ deriving instance MonadThrow m => MonadThrow (RaftLogFileStoreT m) deriving instance MonadCatch m => MonadCatch (RaftLogFileStoreT m) deriving instance MonadMask m => MonadMask (RaftLogFileStoreT m)-deriving instance MonadConc m => MonadConc (RaftLogFileStoreT m) runRaftLogFileStoreT :: RaftLogFile -> RaftLogFileStoreT m a -> m a runRaftLogFileStoreT rlogFile = flip runReaderT rlogFile . unRaftLogFileStoreT -instance (S.Serialize v, MonadIO m, MonadConc m) => RaftInitLog (RaftLogFileStoreT m) v where+instance (S.Serialize v, MonadIO m) => RaftInitLog (RaftLogFileStoreT m) v where type RaftInitLogError (RaftLogFileStoreT m) = RaftLogFileStoreError initializeLog _ = do RaftLogFile logFile <- ask@@ -56,7 +59,7 @@ Left (e :: SomeException) -> pure $ Left (RaftLogFileStoreError (show e)) Right _ -> pure $ Right () -instance (MonadIO m, MonadConc m, S.Serialize v) => RaftWriteLog (RaftLogFileStoreT m) v where+instance (MonadIO m, S.Serialize v) => RaftWriteLog (RaftLogFileStoreT m) v where type RaftWriteLogError (RaftLogFileStoreT m) = RaftLogFileStoreError writeLogEntries newEntries = do RaftLogFile logFile <- ask@@ -65,7 +68,7 @@ Left err -> panic ("writeLogEntries: " <> err) Right currEntries -> liftIO $ Right <$> AW.atomicWriteFile logFile (S.encode (currEntries >< newEntries)) -instance (MonadIO m, MonadConc m, S.Serialize v) => RaftReadLog (RaftLogFileStoreT m) v where+instance (MonadIO m, S.Serialize v) => RaftReadLog (RaftLogFileStoreT m) v where type RaftReadLogError (RaftLogFileStoreT m) = RaftLogFileStoreError readLogEntry (Index idx) = do eLogEntries <- readLogEntries@@ -84,7 +87,7 @@ Seq.Empty -> pure (Right Nothing) (_ Seq.:|> e) -> pure (Right (Just e)) -instance (MonadIO m, MonadConc m, S.Serialize v) => RaftDeleteLog (RaftLogFileStoreT m) v where+instance (MonadIO m, S.Serialize v) => RaftDeleteLog (RaftLogFileStoreT m) v where type RaftDeleteLogError (RaftLogFileStoreT m) = RaftLogFileStoreError deleteLogEntriesFrom idx = do eLogEntries <- readLogEntries@@ -98,3 +101,43 @@ readLogEntries :: (MonadIO m, S.Serialize v) => RaftLogFileStoreT m (Either Text (Entries v)) readLogEntries = liftIO . fmap (first toS . S.decode) . BS.readFile . unRaftLogFile =<< ask++--------------------------------------------------------------------------------+-- Inherited Raft instances+--------------------------------------------------------------------------------++instance RaftPersist m => RaftPersist (RaftLogFileStoreT m) where+ type RaftPersistError (RaftLogFileStoreT m) = RaftPersistError m+ initializePersistentState = lift initializePersistentState+ readPersistentState = lift readPersistentState+ writePersistentState = lift . writePersistentState++instance (Monad m, RaftSendRPC m v) => RaftSendRPC (RaftLogFileStoreT m) v where+ sendRPC nid msg = lift (sendRPC nid msg)++instance (Monad m, RaftRecvRPC m v) => RaftRecvRPC (RaftLogFileStoreT m) v where+ type RaftRecvRPCError (RaftLogFileStoreT m) v = RaftRecvRPCError m v+ receiveRPC = lift receiveRPC++instance (Monad m, RaftSendClient m sm v) => RaftSendClient (RaftLogFileStoreT m) sm v where+ sendClient clientId = lift . sendClient clientId++instance (Monad m, RaftRecvClient m v) => RaftRecvClient (RaftLogFileStoreT m) v where+ type RaftRecvClientError (RaftLogFileStoreT m) v = RaftRecvClientError m v+ receiveClient = lift receiveClient++instance RaftStateMachine m sm v => RaftStateMachine (RaftLogFileStoreT m) sm v where+ validateCmd = lift . validateCmd+ askRaftStateMachinePureCtx = lift askRaftStateMachinePureCtx++instance MonadRaftChan v m => MonadRaftChan v (RaftLogFileStoreT m) where+ type RaftEventChan v (RaftLogFileStoreT m) = RaftEventChan v m+ readRaftChan = lift . readRaftChan+ writeRaftChan chan = lift . writeRaftChan chan+ newRaftChan = lift (newRaftChan @v @m)++instance (MonadIO m, MonadRaftFork m) => MonadRaftFork (RaftLogFileStoreT m) where+ type RaftThreadId (RaftLogFileStoreT m) = RaftThreadId m+ raftFork m = do+ raftLogFile <- ask+ lift $ raftFork (runRaftLogFileStoreT raftLogFile m)
src/Examples/Raft/FileStore/Persistent.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE TypeApplications #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}@@ -14,7 +15,6 @@ import Protolude hiding (try) -import Control.Concurrent.Classy hiding (catch, ThreadId) import Control.Monad.Fail import Control.Monad.Catch import Control.Monad.Trans.Class@@ -25,7 +25,12 @@ import System.Directory (doesFileExist) import qualified System.AtomicWrite.Writer.ByteString as AW +import Raft.Client+import Raft.Log+import Raft.Monad import Raft.Persistent+import Raft.RPC+import Raft.StateMachine newtype RaftPersistFileStoreError = RaftPersistFileStoreError Text deriving (Show)@@ -40,8 +45,10 @@ deriving instance MonadThrow m => MonadThrow (RaftPersistFileStoreT m) deriving instance MonadCatch m => MonadCatch (RaftPersistFileStoreT m) deriving instance MonadMask m => MonadMask (RaftPersistFileStoreT m)-deriving instance MonadConc m => MonadConc (RaftPersistFileStoreT m) +runRaftPersistFileStoreT :: RaftPersistFile -> RaftPersistFileStoreT m a -> m a+runRaftPersistFileStoreT persistFile = flip runReaderT persistFile . unRaftPersistFileStoreT+ -------------------------------------------------------------------------------- -- Raft Instances --------------------------------------------------------------------------------@@ -51,7 +58,7 @@ -- Warning: `atomicWriteFile` causes large pauses on the order of hundreds of -- milliseconds. This can cause leadership to change since the node become -- unresponsive for much longer than the recommended election timeouts.-instance (MonadIO m, MonadConc m) => RaftPersist (RaftPersistFileStoreT m) where+instance MonadIO m => RaftPersist (RaftPersistFileStoreT m) where type RaftPersistError (RaftPersistFileStoreT m) = RaftPersistFileStoreError initializePersistentState = do@@ -60,9 +67,8 @@ if fileExists then pure $ Left (RaftPersistFileStoreError ("Persistent file " <> toS psFile <> " already exists!")) else do- eRes <-- liftIO . try $- AW.atomicWriteFile psFile (toS (S.encode initPersistentState))+ eRes <- liftIO . try $+ AW.atomicWriteFile psFile (toS (S.encode initPersistentState)) case eRes of Left (err :: SomeException) -> do let errPrefix = "Failed initializing persistent storage: "@@ -79,3 +85,54 @@ case S.decode fileContent of Left err -> panic (toS $ "readPersistentState: " ++ err) Right ps -> pure $ Right ps++--------------------------------------------------------------------------------+-- Inherited Instances+--------------------------------------------------------------------------------++instance RaftStateMachine m sm v => RaftStateMachine (RaftPersistFileStoreT m) sm v where+ validateCmd = lift . validateCmd+ askRaftStateMachinePureCtx = lift askRaftStateMachinePureCtx++instance (MonadIO m, MonadMask m, RaftSendClient m sm v) => RaftSendClient (RaftPersistFileStoreT m) sm v where+ sendClient cid msg = lift $ sendClient cid msg++instance (MonadIO m, RaftRecvClient m v) => RaftRecvClient (RaftPersistFileStoreT m) v where+ type RaftRecvClientError (RaftPersistFileStoreT m) v = RaftRecvClientError m v+ receiveClient = lift receiveClient++instance (MonadIO m, MonadMask m, RaftSendRPC m v) => RaftSendRPC (RaftPersistFileStoreT m) v where+ sendRPC nid msg = lift $ sendRPC nid msg++instance (MonadIO m, RaftRecvRPC m v) => RaftRecvRPC (RaftPersistFileStoreT m) v where+ type RaftRecvRPCError (RaftPersistFileStoreT m) v = RaftRecvRPCError m v+ receiveRPC = lift receiveRPC++instance (MonadIO m, RaftInitLog m v) => RaftInitLog (RaftPersistFileStoreT m) v where+ type RaftInitLogError (RaftPersistFileStoreT m) = RaftInitLogError m+ initializeLog p = lift $ initializeLog p++instance RaftWriteLog m v => RaftWriteLog (RaftPersistFileStoreT m) v where+ type RaftWriteLogError (RaftPersistFileStoreT m) = RaftWriteLogError m+ writeLogEntries entries = lift $ writeLogEntries entries++instance RaftReadLog m v => RaftReadLog (RaftPersistFileStoreT m) v where+ type RaftReadLogError (RaftPersistFileStoreT m) = RaftReadLogError m+ readLogEntry idx = lift $ readLogEntry idx+ readLastLogEntry = lift readLastLogEntry++instance RaftDeleteLog m v => RaftDeleteLog (RaftPersistFileStoreT m) v where+ type RaftDeleteLogError (RaftPersistFileStoreT m) = RaftDeleteLogError m+ deleteLogEntriesFrom idx = lift $ deleteLogEntriesFrom idx++instance MonadRaftChan v m => MonadRaftChan v (RaftPersistFileStoreT m) where+ type RaftEventChan v (RaftPersistFileStoreT m) = RaftEventChan v m+ readRaftChan = lift . readRaftChan+ writeRaftChan chan = lift . writeRaftChan chan+ newRaftChan = lift (newRaftChan @v @m)++instance (MonadIO m, MonadRaftFork m) => MonadRaftFork (RaftPersistFileStoreT m) where+ type RaftThreadId (RaftPersistFileStoreT m) = RaftThreadId m+ raftFork m = do+ persistFile <- ask+ lift $ raftFork (runRaftPersistFileStoreT persistFile m)
src/Examples/Raft/Socket/Client.hs view
@@ -61,7 +61,6 @@ deriving instance MonadThrow m => MonadThrow (RaftClientRespChanT s v m) deriving instance MonadCatch m => MonadCatch (RaftClientRespChanT s v m) deriving instance MonadMask m => MonadMask (RaftClientRespChanT s v m)-deriving instance MonadConc m => MonadConc (RaftClientRespChanT s v m) -- This annoying instance is because of the Haskeline library, letting us use a -- custom monad transformer stack as the base monad of 'InputT'. IMO it should@@ -115,7 +114,7 @@ $ rscm clientResponseServer- :: forall s v m. (S.Serialize s, S.Serialize v, MonadIO m, MonadConc m)+ :: forall s v m. (S.Serialize s, S.Serialize v, MonadIO m) => N.HostName -> N.ServiceName -> RaftClientRespChanT s v m ()
src/Examples/Raft/Socket/Node.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE TypeApplications #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}@@ -7,23 +8,16 @@ {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE StandaloneDeriving #-}-{-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE ScopedTypeVariables #-} module Examples.Raft.Socket.Node where -import Protolude hiding- ( MVar, putMVar, takeMVar, newMVar, newEmptyMVar, readMVar- , atomically, STM(..), Chan, newTVar, readTVar, writeTVar- , newChan, writeChan, readChan- , threadDelay, killThread, TVar(..)- , catch, handle, takeWhile, takeWhile1, (<|>)- )+import Protolude -import Control.Concurrent.Classy hiding (catch, ThreadId) import Control.Monad.Fail import Control.Monad.Catch import Control.Monad.Trans.Class+import Control.Concurrent.STM.TChan import qualified Data.Serialize as S import qualified Network.Simple.TCP as NS@@ -31,15 +25,22 @@ import Examples.Raft.Socket.Common -import Raft+import Raft.Client+import Raft.Event+import Raft.Log+import Raft.Monad+import Raft.Persistent+import Raft.RPC+import Raft.StateMachine+import Raft.Types -------------------------------------------------------------------------------- -- Network -------------------------------------------------------------------------------- data NodeSocketEnv v = NodeSocketEnv- { nsMsgQueue :: TChan (STM IO) (RPCMessage v)- , nsClientReqQueue :: TChan (STM IO) (ClientRequest v)+ { nsMsgQueue :: TChan (RPCMessage v)+ , nsClientReqQueue :: TChan (ClientRequest v) } newtype RaftSocketT v m a = RaftSocketT { unRaftSocketT :: ReaderT (NodeSocketEnv v) m a }@@ -48,13 +49,12 @@ deriving instance MonadThrow m => MonadThrow (RaftSocketT v m) deriving instance MonadCatch m => MonadCatch (RaftSocketT v m) deriving instance MonadMask m => MonadMask (RaftSocketT v m)-deriving instance MonadConc m => MonadConc (RaftSocketT v m) -------------------- -- Raft Instances -- -------------------- -instance (MonadIO m, MonadConc m, S.Serialize sm, S.Serialize v) => RaftSendClient (RaftSocketT v m) sm v where+instance (MonadMask m, MonadCatch m, MonadIO m, S.Serialize sm, S.Serialize v) => RaftSendClient (RaftSocketT v m) sm v where sendClient clientId@(ClientId nid) msg = do let (cHost, cPort) = nidToHostPort (toS nid) eRes <- Control.Monad.Catch.try $@@ -64,13 +64,13 @@ Left (err :: SomeException) -> putText ("Failed to send Client: " <> show err) Right _ -> pure () -instance (MonadIO m, MonadConc m, S.Serialize v) => RaftRecvClient (RaftSocketT v m) v where+instance (MonadIO m, S.Serialize v) => RaftRecvClient (RaftSocketT v m) v where type RaftRecvClientError (RaftSocketT v m) v = Text receiveClient = do cReq <- asks nsClientReqQueue fmap Right . liftIO . atomically $ readTChan cReq -instance (MonadIO m, MonadConc m, S.Serialize v, Show v) => RaftSendRPC (RaftSocketT v m) v where+instance (MonadCatch m, MonadMask m, MonadIO m, S.Serialize v, Show v) => RaftSendRPC (RaftSocketT v m) v where sendRPC nid msg = do eRes <- Control.Monad.Catch.try $ connect host port $ \(sock,_) -> do@@ -81,17 +81,17 @@ where (host, port) = nidToHostPort nid -instance (MonadIO m, MonadConc m, Show v) => RaftRecvRPC (RaftSocketT v m) v where+instance (MonadIO m, Show v) => RaftRecvRPC (RaftSocketT v m) v where type RaftRecvRPCError (RaftSocketT v m) v = Text receiveRPC = do msgQueue <- asks nsMsgQueue fmap Right . liftIO . atomically $ readTChan msgQueue -runRaftSocketT :: (MonadIO m, MonadConc m) => NodeSocketEnv v -> RaftSocketT v m a -> m a+runRaftSocketT :: MonadIO m => NodeSocketEnv v -> RaftSocketT v m a -> m a runRaftSocketT nodeSocketEnv = flip runReaderT nodeSocketEnv . unRaftSocketT acceptConnections- :: forall v m. (S.Serialize v, MonadIO m, MonadConc m)+ :: forall v m. (S.Serialize v, MonadIO m) => HostName -> ServiceName -> RaftSocketT v m ()@@ -111,3 +111,46 @@ newSock :: HostName -> ServiceName -> IO Socket newSock host port = listen (Host host) port (pure . fst)++--------------------------------------------------------------------------------+-- Inherited Instances+--------------------------------------------------------------------------------++instance (MonadIO m, RaftPersist m) => RaftPersist (RaftSocketT v m) where+ type RaftPersistError (RaftSocketT v m) = RaftPersistError m+ initializePersistentState = lift initializePersistentState+ writePersistentState ps = lift $ writePersistentState ps+ readPersistentState = lift readPersistentState++instance (MonadIO m, RaftInitLog m v) => RaftInitLog (RaftSocketT v m) v where+ type RaftInitLogError (RaftSocketT v m) = RaftInitLogError m+ initializeLog p = lift $ initializeLog p++instance RaftWriteLog m v => RaftWriteLog (RaftSocketT v m) v where+ type RaftWriteLogError (RaftSocketT v m) = RaftWriteLogError m+ writeLogEntries entries = lift $ writeLogEntries entries++instance RaftReadLog m v => RaftReadLog (RaftSocketT v m) v where+ type RaftReadLogError (RaftSocketT v m) = RaftReadLogError m+ readLogEntry idx = lift $ readLogEntry idx+ readLastLogEntry = lift readLastLogEntry++instance RaftDeleteLog m v => RaftDeleteLog (RaftSocketT v m) v where+ type RaftDeleteLogError (RaftSocketT v m) = RaftDeleteLogError m+ deleteLogEntriesFrom idx = lift $ deleteLogEntriesFrom idx++instance RaftStateMachine m sm v => RaftStateMachine (RaftSocketT v m) sm v where+ validateCmd = lift . validateCmd+ askRaftStateMachinePureCtx = lift askRaftStateMachinePureCtx++instance MonadRaftChan v m => MonadRaftChan v (RaftSocketT v m) where+ type RaftEventChan v (RaftSocketT v m) = RaftEventChan v m+ readRaftChan = lift . readRaftChan+ writeRaftChan chan = lift . writeRaftChan chan+ newRaftChan = lift (newRaftChan @v @m)++instance (MonadIO m, MonadRaftFork m) => MonadRaftFork (RaftSocketT v m) where+ type RaftThreadId (RaftSocketT v m) = RaftThreadId m+ raftFork m = do+ persistFile <- ask+ lift $ raftFork (runRaftSocketT persistFile m)
src/Raft.hs view
@@ -8,6 +8,7 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE AllowAmbiguousTypes #-} {-# LANGUAGE TypeApplications #-} module Raft@@ -23,8 +24,6 @@ , RaftRecvClient(..) , RaftPersist(..) - , EventChan- , RaftEnv(..) , runRaftNode , runRaftT@@ -40,7 +39,7 @@ , ClientRedirResp(..) -- * Configuration- , NodeConfig(..)+ , RaftNodeConfig(..) -- * Events , Event(..)@@ -104,16 +103,12 @@ , AppendEntriesData(..) ) where -import Protolude hiding (STM, TChan, newTChan, readBoundedChan, writeBoundedChan, atomically)+import Protolude hiding (STM, TChan, newRaftChan, readBoundedChan, writeBoundedChan, atomically) -import Control.Monad.Conc.Class import Control.Concurrent.STM.Timer-import Control.Concurrent.Classy.STM.TChan-import Control.Concurrent.Classy.Async import Control.Monad.Fail import Control.Monad.Catch-import Control.Monad.Trans.Class import qualified Data.Map as Map import Data.Serialize (Serialize)@@ -125,9 +120,10 @@ import Raft.Config import Raft.Event import Raft.Handle+import Raft.Monad import Raft.Log import Raft.Logging hiding (logInfo, logDebug, logCritical, logAndPanic)-import Raft.Monad hiding (logInfo, logDebug)+import Raft.Transition hiding (logInfo, logDebug) import Raft.NodeState import Raft.Persistent import Raft.RPC@@ -135,62 +131,12 @@ import Raft.Types -type EventChan m v = TChan (STM m) (Event v)---- | The raft server environment composed of the concurrent variables used in--- the effectful raft layer.-data RaftEnv v m = RaftEnv- { eventChan :: EventChan m v- , resetElectionTimer :: m ()- , resetHeartbeatTimer :: m ()- , raftNodeConfig :: NodeConfig- , raftNodeLogCtx :: LogCtx- }--newtype RaftT v m a = RaftT- { unRaftT :: ReaderT (RaftEnv v m) (StateT (RaftNodeState v) m) a- } deriving (Functor, Applicative, Monad, MonadReader (RaftEnv v m), MonadState (RaftNodeState v), MonadFail, Alternative, MonadPlus)--instance MonadTrans (RaftT v) where- lift = RaftT . lift . lift--deriving instance MonadIO m => MonadIO (RaftT v m)-deriving instance MonadThrow m => MonadThrow (RaftT v m)-deriving instance MonadCatch m => MonadCatch (RaftT v m)-deriving instance MonadMask m => MonadMask (RaftT v m)-deriving instance MonadConc m => MonadConc (RaftT v m)--instance Monad m => RaftLogger v (RaftT v m) where- loggerCtx = (,) <$> asks (configNodeId . raftNodeConfig) <*> get--runRaftT- :: MonadConc m- => RaftNodeState v- -> RaftEnv v m- -> RaftT v m ()- -> m ()-runRaftT raftNodeState raftEnv =- flip evalStateT raftNodeState . flip runReaderT raftEnv . unRaftT----------------------------------------------------------------------------------logDebug :: MonadIO m => Text -> RaftT v m ()-logDebug msg = flip logDebugIO msg =<< asks raftNodeLogCtx--logCritical :: MonadIO m => Text -> RaftT v m ()-logCritical msg = flip logCriticalIO msg =<< asks raftNodeLogCtx--logAndPanic :: MonadIO m => Text -> RaftT v m a-logAndPanic msg = flip logAndPanicIO msg =<< asks raftNodeLogCtx--------------------------------------------------------------------------------- -- | Run timers, RPC and client request handlers and start event loop. -- It should run forever runRaftNode :: forall m sm v.- ( Show v, Show sm, Serialize v, Show (Action sm v), Show (RaftLogError m), Show (RaftStateMachinePureError sm v)- , Typeable m, MonadIO m, MonadConc m, MonadFail m+ ( Typeable m, Show v, Show sm, Serialize v, Show (Action sm v), Show (RaftLogError m), Show (RaftStateMachinePureError sm v)+ , MonadIO m, MonadCatch m, MonadFail m, MonadRaft v m , RaftStateMachine m sm v , RaftSendRPC m v , RaftRecvRPC m v@@ -201,34 +147,31 @@ , RaftPersist m , Exception (RaftPersistError m) )- => NodeConfig -- ^ Node configuration+ => RaftNodeConfig -- ^ Node configuration -> LogCtx -- ^ Logs destination -> Int -- ^ Timer seed -> sm -- ^ Initial state machine state -> m ()-runRaftNode nodeConfig@NodeConfig{..} logCtx timerSeed initRaftStateMachine = do+runRaftNode nodeConfig@RaftNodeConfig{..} logCtx timerSeed initRaftStateMachine = do -- Initialize the persistent state and logs storage if specified initializeStorage - eventChan <- atomically newTChan-- electionTimer <-- newTimerRange (writeTimeoutEvent eventChan ElectionTimeout) timerSeed configElectionTimeout-- heartbeatTimer <-- newTimer (writeTimeoutEvent eventChan HeartbeatTimeout) configHeartbeatTimeout+ eventChan <- newRaftChan @v - let resetElectionTimer = void $ resetTimer electionTimer- resetHeartbeatTimer = void $ resetTimer heartbeatTimer- raftEnv = RaftEnv eventChan resetElectionTimer resetHeartbeatTimer nodeConfig logCtx+ -- Create timers and reset timer actions+ electionTimer <- liftIO $ newTimerRange timerSeed configElectionTimeout+ heartbeatTimer <- liftIO $ newTimer configHeartbeatTimeout+ let resetElectionTimer = liftIO $ void $ resetTimer electionTimer+ resetHeartbeatTimer = liftIO $ void $ resetTimer heartbeatTimer + let raftEnv = RaftEnv eventChan resetElectionTimer resetHeartbeatTimer nodeConfig logCtx runRaftT initRaftNodeState raftEnv $ do - -- Fork all event producers to run concurrently- lift $ fork (electionTimeoutTimer electionTimer)- lift $ fork (heartbeatTimeoutTimer heartbeatTimer)- fork (rpcHandler eventChan)- fork (clientReqHandler eventChan)+ -- These event producers need access to logging, thus they live in RaftT+ raftFork . lift $ electionTimeoutTimer @m @v eventChan electionTimer+ raftFork . lift $ heartbeatTimeoutTimer @m @v eventChan heartbeatTimer+ raftFork (rpcHandler @m @v eventChan)+ raftFork (clientReqHandler @m @v eventChan) -- Start the main event handling loop handleEventLoop initRaftStateMachine@@ -239,18 +182,18 @@ New -> do ipsRes <- initializePersistentState case ipsRes of- Left err -> throw err+ Left err -> throwM err Right _ -> do ilRes <- initializeLog (Proxy :: Proxy v) case ilRes of- Left err -> throw err+ Left err -> throwM err Right _ -> pure () Existing -> pure () handleEventLoop :: forall sm v m. ( Show v, Serialize v, Show sm, Show (Action sm v), Show (RaftLogError m), Typeable m- , MonadIO m, MonadConc m, MonadFail m+ , MonadIO m, MonadRaft v m, MonadFail m, MonadThrow m , RaftStateMachine m sm v , Show (RaftStateMachinePureError sm v) , RaftPersist m@@ -267,12 +210,12 @@ setInitLastLogEntry ePersistentState <- lift readPersistentState case ePersistentState of- Left err -> throw err+ Left err -> throwM err Right pstate -> handleEventLoop' initRaftStateMachine pstate where handleEventLoop' :: sm -> PersistentState -> RaftT v m () handleEventLoop' stateMachine persistentState = do- event <- atomically . readTChan =<< asks eventChan+ event <- lift . readRaftChan =<< asks eventChan loadLogEntryTermAtAePrevLogIndex event raftNodeState <- get logDebug $ "[Event]: " <> show event@@ -293,7 +236,7 @@ when (resPersistentState /= persistentState) $ do eRes <- lift $ writePersistentState resPersistentState case eRes of- Left err -> throw err+ Left err -> throwM err Right _ -> pure () -- Update raft node state with the resulting node state@@ -301,7 +244,7 @@ -- Handle logs produced by core state machine handleLogs logMsgs -- Handle actions produced by core state machine- handleActions nodeConfig actions+ handleActions actions -- Apply new log entries to the state machine resRaftStateMachine <- applyLogEntries stateMachine handleEventLoop' resRaftStateMachine resPersistentState@@ -317,7 +260,7 @@ NodeFollowerState fs -> do eEntry <- lift $ readLogEntry (aePrevLogIndex ae) case eEntry of- Left err -> throw err+ Left err -> throwM err Right (mEntry :: Maybe (Entry v)) -> put $ RaftNodeState $ NodeFollowerState fs { fsTermAtAEPrevIndex = entryTerm <$> mEntry }@@ -330,54 +273,54 @@ RaftNodeState rns <- get eLogEntry <- lift readLastLogEntry case eLogEntry of- Left err -> throw err+ Left err -> throwM err Right Nothing -> pure () Right (Just e) -> put (RaftNodeState (setLastLogEntry rns (singleton e))) handleActions :: ( Show v, Show sm, Show (Action sm v), Show (RaftLogError m), Typeable m- , MonadIO m, MonadConc m+ , MonadIO m, MonadRaft v m, MonadThrow m , RaftStateMachine m sm v , RaftSendRPC m v , RaftSendClient m sm v , RaftLog m v , RaftLogExceptions m )- => NodeConfig- -> [Action sm v]+ => [Action sm v] -> RaftT v m ()-handleActions = mapM_ . handleAction+handleActions = mapM_ handleAction handleAction :: forall sm v m. ( Show v, Show sm, Show (Action sm v), Show (RaftLogError m), Typeable m- , MonadIO m, MonadConc m+ , MonadIO m, MonadRaft v m, MonadThrow m , RaftStateMachine m sm v , RaftSendRPC m v , RaftSendClient m sm v , RaftLog m v , RaftLogExceptions m )- => NodeConfig- -> Action sm v+ => Action sm v -> RaftT v m ()-handleAction nodeConfig action = do+handleAction action = do logDebug $ "[Action]: " <> show action case action of SendRPC nid sendRpcAction -> do rpcMsg <- mkRPCfromSendRPCAction sendRpcAction lift (sendRPC nid rpcMsg) SendRPCs rpcMap ->- forConcurrently_ (Map.toList rpcMap) $ \(nid, sendRpcAction) -> do- rpcMsg <- mkRPCfromSendRPCAction sendRpcAction- lift (sendRPC nid rpcMsg)+ flip mapM_ (Map.toList rpcMap) $ \(nid, sendRpcAction) ->+ raftFork $ do+ rpcMsg <- mkRPCfromSendRPCAction sendRpcAction+ lift (sendRPC nid rpcMsg) BroadcastRPC nids sendRpcAction -> do rpcMsg <- mkRPCfromSendRPCAction sendRpcAction- mapConcurrently_ (lift . flip sendRPC rpcMsg) nids+ mapM_ (raftFork . lift . flip sendRPC rpcMsg) nids RespondToClient cid cr -> do clientResp <- mkClientResp cr- void . fork . lift $ sendClient cid clientResp+ -- TODO log failure if sendClient fails+ void $ raftFork $ lift $ sendClient cid clientResp ResetTimeoutTimer tout -> do case tout of ElectionTimeout -> lift . resetElectionTimer =<< ask@@ -396,7 +339,7 @@ NodeLeaderState ls@LeaderState{..} -> do eentries <- lift (readLogEntriesFrom idx) case eentries of- Left err -> throw err+ Left err -> throwM err Right (entries :: Entries v) -> do let committedClientReqs = clientReqData entries when (Map.size committedClientReqs > 0) $ do@@ -408,8 +351,7 @@ where respondClientWrite :: (ClientId, (SerialNum, Index)) -> RaftT v m () respondClientWrite (cid, (sn,idx)) =- handleAction nodeConfig $- RespondToClient cid (ClientWriteRespSpec idx sn :: ClientRespSpec sm)+ handleAction $ RespondToClient cid (ClientWriteRespSpec idx sn :: ClientRespSpec sm) mkClientResp :: ClientRespSpec sm -> RaftT v m (ClientResponse sm v) mkClientResp crs =@@ -420,7 +362,7 @@ ClientReadRespSpecEntries res -> do eRes <- lift (readEntries res) case eRes of- Left err -> throw err+ Left err -> throwM err Right res -> case res of OneEntry e -> pure (ClientReadRespEntry e)@@ -435,6 +377,7 @@ mkRPCfromSendRPCAction :: SendRPCAction v -> RaftT v m (RPCMessage v) mkRPCfromSendRPCAction sendRPCAction = do RaftNodeState ns <- get+ nodeConfig <- asks raftNodeConfig RPCMessage (configNodeId nodeConfig) <$> case sendRPCAction of SendAppendEntriesRPC aeData -> do@@ -443,7 +386,7 @@ FromIndex idx -> do eLogEntries <- lift (readLogEntriesFrom (decrIndexWithDefault0 idx)) case eLogEntries of- Left err -> throw err+ Left err -> throwM err Right log -> case log of pe :<| entries@(e :<| _)@@ -484,7 +427,7 @@ let prevLogEntryIdx = decrIndexWithDefault0 (entryIndex e) eLogEntry <- lift $ readLogEntry prevLogEntryIdx case eLogEntry of- Left err -> throw err+ Left err -> throwM err Right Nothing -> pure (singleton e, index0, term0) Right (Just (prevEntry :: Entry v)) -> pure (singleton e, entryIndex prevEntry, entryTerm prevEntry)@@ -494,13 +437,10 @@ -- is up to date with all the committed log entries applyLogEntries :: forall sm m v.- ( Show sm- , MonadIO m- , MonadConc m- , RaftReadLog m v- , Exception (RaftReadLogError m)+ ( Show sm, Show (RaftStateMachinePureError sm v)+ , MonadIO m, MonadThrow m, MonadRaft v m+ , RaftReadLog m v, Exception (RaftReadLogError m) , RaftStateMachine m sm v- , Show (RaftStateMachinePureError sm v) ) => sm -> RaftT v m sm@@ -513,7 +453,7 @@ let newLastAppliedIndex = lastApplied resNodeState eLogEntry <- lift $ readLogEntry newLastAppliedIndex case eLogEntry of- Left err -> throw err+ Left err -> throwM err Right Nothing -> logAndPanic $ "No log entry at 'newLastAppliedIndex := " <> show newLastAppliedIndex <> "'" Right (Just logEntry) -> do -- The command should be verified by the leader, thus all node@@ -545,7 +485,7 @@ commitIndex = snd . getLastAppliedAndCommitIndex handleLogs- :: (MonadIO m, MonadConc m)+ :: (MonadIO m, MonadRaft v m) => [LogMsg] -> RaftT v m () handleLogs logs = do@@ -558,8 +498,8 @@ -- | Producer for rpc message events rpcHandler- :: (MonadIO m, MonadConc m, Show v, RaftRecvRPC m v)- => EventChan m v+ :: forall m v. (MonadIO m, MonadRaft v m, MonadCatch m, Show v, RaftRecvRPC m v)+ => RaftEventChan v m -> RaftT v m () rpcHandler eventChan = forever $ do@@ -569,12 +509,12 @@ Right (Left err) -> logCritical (show err) Right (Right rpcMsg) -> do let rpcMsgEvent = MessageEvent (RPCMessageEvent rpcMsg)- atomically $ writeTChan eventChan rpcMsgEvent+ lift $ writeRaftChan @v @m eventChan rpcMsgEvent -- | Producer for rpc message events clientReqHandler- :: (MonadIO m, MonadConc m, RaftRecvClient m v)- => EventChan m v+ :: forall m v. (MonadIO m, MonadRaft v m, MonadCatch m, RaftRecvClient m v)+ => RaftEventChan v m -> RaftT v m () clientReqHandler eventChan = forever $ do@@ -584,27 +524,29 @@ Right (Left err) -> logCritical (show err) Right (Right clientReq) -> do let clientReqEvent = MessageEvent (ClientRequestEvent clientReq)- atomically $ writeTChan eventChan clientReqEvent+ lift $ writeRaftChan @v @m eventChan clientReqEvent -- | Producer for the election timeout event-electionTimeoutTimer :: (MonadIO m, MonadConc m) => Timer m -> m ()-electionTimeoutTimer timer =+electionTimeoutTimer :: forall m v. (MonadIO m, MonadRaft v m) => RaftEventChan v m -> Timer -> m ()+electionTimeoutTimer eventChan timer = forever $ do- success <- startTimer timer+ success <- liftIO $ startTimer timer when (not success) $ panic "[Failed invariant]: Election timeout timer failed to start."- waitTimer timer+ liftIO $ waitTimer timer+ writeTimeoutEvent @m @v eventChan ElectionTimeout -- | Producer for the heartbeat timeout event-heartbeatTimeoutTimer :: (MonadIO m, MonadConc m) => Timer m -> m ()-heartbeatTimeoutTimer timer =+heartbeatTimeoutTimer :: forall m v. (MonadIO m, MonadRaft v m) => RaftEventChan v m -> Timer -> m ()+heartbeatTimeoutTimer eventChan timer = forever $ do- success <- startTimer timer+ success <- liftIO $ startTimer timer when (not success) $ panic "[Failed invariant]: Heartbeat timeout timer failed to start."- waitTimer timer+ liftIO $ waitTimer timer+ writeTimeoutEvent @m @v eventChan HeartbeatTimeout -writeTimeoutEvent :: (MonadIO m , MonadConc m) => EventChan m v -> Timeout -> m ()+writeTimeoutEvent :: forall m v. (MonadIO m , MonadRaft v m) => RaftEventChan v m -> Timeout -> m () writeTimeoutEvent eventChan timeout = do now <- liftIO getSystemTime- atomically $ writeTChan eventChan (TimeoutEvent now timeout)+ writeRaftChan @v @m eventChan (TimeoutEvent now timeout)
src/Raft/Candidate.hs view
@@ -33,7 +33,7 @@ import Raft.Persistent import Raft.Log import Raft.Config-import Raft.Monad+import Raft.Transition import Raft.Types --------------------------------------------------------------------------------
src/Raft/Client.hs view
@@ -69,13 +69,15 @@ import Protolude hiding (threadDelay, STM, ) -import Control.Concurrent.Classy+import Control.Concurrent.Lifted (threadDelay)+ import Control.Monad.Base import Control.Monad.Catch import Control.Monad.Fail import Control.Monad.Trans.Class import Control.Monad.Trans.Control + import qualified Data.Set as Set import qualified Data.Serialize as S @@ -219,8 +221,6 @@ deriving newtype instance MonadThrow m => MonadThrow (RaftClientT s v m) deriving newtype instance MonadCatch m => MonadCatch (RaftClientT s v m) deriving newtype instance MonadMask m => MonadMask (RaftClientT s v m)-deriving newtype instance MonadSTM m => MonadSTM (RaftClientT s v m)-deriving newtype instance MonadConc m => MonadConc (RaftClientT s v m) instance MonadTrans (RaftClientT s v) where lift = RaftClientT . lift . lift@@ -232,6 +232,10 @@ liftWith = defaultLiftWith2 RaftClientT unRaftClientT restoreT = defaultRestoreT2 RaftClientT +-- These instances are for use of the 'timeout' function that requires a+-- MonadBaseControl IO constraint.+--+-- TODO is this still necessary after the removal of MonadConc? instance (MonadBaseControl IO m) => MonadBaseControl IO (RaftClientT s v m) where type StM (RaftClientT s v m) a = ComposeSt (RaftClientT s v) m a liftBaseWith = defaultLiftBaseWith@@ -274,7 +278,7 @@ -- | Send a read request to the curent leader and wait for a response clientRead- :: (Show (RaftClientSendError m v), RaftClientSend m v, RaftClientRecv m s v)+ :: (RaftClientSend m v, RaftClientRecv m s v) => ClientReadReq -> RaftClientT s v m (Either (RaftClientError s v m) (ClientReadResp s v)) clientRead crr = do@@ -298,7 +302,7 @@ -- | 'clientRead' but with a timeout clientReadTimeout- :: (MonadBaseControl IO m, Show (RaftClientSendError m v), RaftClientSend m v, RaftClientRecv m s v)+ :: (MonadBaseControl IO m, RaftClientSend m v, RaftClientRecv m s v) => Int -> ClientReadReq -> RaftClientT s v m (Either (RaftClientError s v m) (ClientReadResp s v))@@ -306,7 +310,7 @@ -- | Send a write request to the current leader and wait for a response clientWrite- :: (Show (RaftClientSendError m v), RaftClientSend m v, RaftClientRecv m s v)+ :: (RaftClientSend m v, RaftClientRecv m s v) => v -> RaftClientT s v m (Either (RaftClientError s v m) ClientWriteResp) clientWrite cmd = do@@ -329,7 +333,7 @@ Right _ -> clientRecvWrite clientWriteTimeout- :: (MonadBaseControl IO m, Show (RaftClientSendError m v), RaftClientSend m v, RaftClientRecv m s v)+ :: (MonadBaseControl IO m, RaftClientSend m v, RaftClientRecv m s v) => Int -> v -> RaftClientT s v m (Either (RaftClientError s v m) ClientWriteResp)@@ -351,7 +355,7 @@ -- | Given a blocking client send/receive, retry if the received value is not -- expected retryOnRedirect- :: MonadConc m+ :: MonadBaseControl IO m => RaftClientT s v m (Either (RaftClientError s v m) r) -> RaftClientT s v m (Either (RaftClientError s v m) r) retryOnRedirect action = do@@ -369,7 +373,7 @@ -- | Send a read request to the current leader. Nonblocking. clientSendRead- :: (Show (RaftClientSendError m v), RaftClientSend m v)+ :: RaftClientSend m v => ClientReadReq -> RaftClientT s v m (Either (RaftClientSendError m v) ()) clientSendRead crr =@@ -387,7 +391,7 @@ -- | Send a write request to the current leader. Nonblocking. clientSendWrite- :: (Show (RaftClientSendError m v), RaftClientSend m v)+ :: RaftClientSend m v => v -> RaftClientT s v m (Either (RaftClientSendError m v) ()) clientSendWrite v = do@@ -407,7 +411,7 @@ -- | Send a request to the current leader. Nonblocking. clientSend- :: (Show (RaftClientSendError m v), RaftClientSend m v)+ :: (RaftClientSend m v) => ClientRequest v -> RaftClientT s v m (Either (RaftClientSendError m v) ()) clientSend creq = do
src/Raft/Config.hs view
@@ -9,7 +9,7 @@ import Raft.Types -- | Configuration of a node in the cluster-data NodeConfig = NodeConfig+data RaftNodeConfig = RaftNodeConfig { configNodeId :: NodeId -- ^ Node id of the running node , configNodeIds :: NodeIds -- ^ Set of all other node ids in the cluster , configElectionTimeout :: (Natural, Natural) -- ^ Range of times an election timeout can take
src/Raft/Follower.hs view
@@ -27,7 +27,7 @@ import Raft.Event import Raft.Persistent import Raft.Log (entryIndex)-import Raft.Monad+import Raft.Transition import Raft.Types --------------------------------------------------------------------------------
src/Raft/Handle.hs view
@@ -19,7 +19,7 @@ import Raft.Action import Raft.Event-import Raft.Monad+import Raft.Transition import Raft.NodeState import Raft.Persistent import Raft.RPC
src/Raft/Leader.hs view
@@ -32,7 +32,7 @@ import Raft.Event import Raft.Persistent import Raft.Log (Entry(..), EntryIssuer(..), EntryValue(..))-import Raft.Monad+import Raft.Transition import Raft.Types --------------------------------------------------------------------------------
src/Raft/Log/PostgreSQL.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeApplications #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE DeriveAnyClass #-}@@ -11,7 +12,6 @@ {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE DeriveFunctor #-}-{-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE TemplateHaskell #-} module Raft.Log.PostgreSQL (@@ -28,10 +28,10 @@ ) where -import Protolude hiding (atomically, try, bracket, catches, tryReadTMVar, newEmptyTMVar, STM, Handler)+import Protolude hiding (Handler, catches, bracket) -import Control.Concurrent.Classy (STM, MonadConc, atomically)-import Control.Concurrent.Classy.STM.TMVar+import Control.Concurrent.STM.TMVar+ import Control.Monad.Catch import Control.Monad.Fail import Control.Monad.Trans.Class@@ -45,6 +45,7 @@ import Raft.Client import Raft.RPC+import Raft.Monad import Raft.Log import Raft.StateMachine import Raft.Persistent@@ -52,24 +53,32 @@ data RaftPostgresEnv = RaftPostgresEnv { raftPostgresConnInfo :: ConnectInfo- , raftPostgresConn :: TMVar (STM IO) Connection+ , raftPostgresConn :: TMVar Connection } -- | A single threaded PostgreSQL storage monad transformer newtype RaftPostgresT m a = RaftPostgresT { unRaftPostgresT :: ReaderT RaftPostgresEnv m a } deriving newtype (Functor, Applicative, Monad, MonadIO, MonadFail, MonadReader RaftPostgresEnv, Alternative, MonadPlus, MonadTrans) -deriving newtype instance MonadThrow m => MonadThrow (RaftPostgresT m) deriving newtype instance MonadCatch m => MonadCatch (RaftPostgresT m)+deriving newtype instance MonadThrow m => MonadThrow (RaftPostgresT m) deriving newtype instance MonadMask m => MonadMask (RaftPostgresT m)-deriving newtype instance MonadConc m => MonadConc (RaftPostgresT m) +initRaftPostgresEnv :: MonadIO m => ConnectInfo -> m RaftPostgresEnv+initRaftPostgresEnv connInfo =+ RaftPostgresEnv connInfo <$> liftIO (atomically newEmptyTMVar)++-- | Run a RaftPostgresT computation by supplying the database connection info runRaftPostgresT :: MonadIO m => ConnectInfo -> RaftPostgresT m a -> m a runRaftPostgresT connInfo m = do connTMVar <- liftIO $ atomically newEmptyTMVar let postgresEnv = RaftPostgresEnv connInfo connTMVar runReaderT (unRaftPostgresT m) postgresEnv +runRaftPostgresT' :: MonadIO m => RaftPostgresEnv -> RaftPostgresT m a -> m a+runRaftPostgresT' raftPostgresEnv =+ flip runReaderT raftPostgresEnv . unRaftPostgresT+ type RaftPostgresM = RaftPostgresT IO runRaftPostgresM :: ConnectInfo -> RaftPostgresM a -> IO a@@ -86,7 +95,7 @@ -- | Helper function for RaftXLog typeclasses such that we can easily catch all -- @PGError@s and connect to the DB if we haven't already withRaftPostgresConn- :: (MonadIO m, MonadConc m)+ :: MonadIO m => (Connection -> IO (Either RaftPostgresError a)) -> RaftPostgresT m (Either RaftPostgresError a) withRaftPostgresConn f = do@@ -113,6 +122,10 @@ Right (Left err) -> pure (Left err) Right (Right a) -> pure (Right a) +--------------------------------------------------------------------------------+-- Raft instances+--------------------------------------------------------------------------------+ instance (MonadIO m) => RaftInitLog (RaftPostgresT m) v where type RaftInitLogError (RaftPostgresT m) = RaftPostgresError initializeLog _ = do@@ -123,7 +136,7 @@ connTMVar <- asks raftPostgresConn fmap Right . liftIO . atomically $ putTMVar connTMVar conn -instance (Typeable v, Serialize v, MonadIO m, MonadConc m) => RaftReadLog (RaftPostgresT m) v where+instance (Typeable v, Serialize v, MonadIO m) => RaftReadLog (RaftPostgresT m) v where type RaftReadLogError (RaftPostgresT m) = RaftPostgresError readLogEntry idx = withRaftPostgresConn $ \conn ->@@ -140,14 +153,14 @@ Right . fmap rowTypeToEntry . listToMaybe <$> query_ conn "SELECT * FROM entries ORDER BY entryIndex DESC LIMIT 1" -instance (Serialize v, MonadIO m, MonadConc m) => RaftWriteLog (RaftPostgresT m) v where+instance (Serialize v, MonadIO m) => RaftWriteLog (RaftPostgresT m) v where type RaftWriteLogError (RaftPostgresT m) = RaftPostgresError writeLogEntries entries = withRaftPostgresConn $ \conn -> fmap Right . void $ executeMany conn "INSERT INTO entries VALUES (?,?,?,?,?,?)" (map entryToRowType (toList entries)) -instance (Serialize v, MonadIO m, MonadConc m) => RaftDeleteLog (RaftPostgresT m) v where+instance (Serialize v, MonadIO m) => RaftDeleteLog (RaftPostgresT m) v where type RaftDeleteLogError (RaftPostgresT m) = RaftPostgresError deleteLogEntriesFrom idx = withRaftPostgresConn $ \conn ->@@ -182,6 +195,18 @@ validateCmd = lift . validateCmd askRaftStateMachinePureCtx = lift askRaftStateMachinePureCtx +instance MonadRaftChan v m => MonadRaftChan v (RaftPostgresT m) where+ type RaftEventChan v (RaftPostgresT m) = RaftEventChan v m+ readRaftChan = lift . readRaftChan+ writeRaftChan chan = lift . writeRaftChan chan+ newRaftChan = lift (newRaftChan @v @m)++instance (MonadIO m, MonadRaftFork m) => MonadRaftFork (RaftPostgresT m) where+ type RaftThreadId (RaftPostgresT m) = RaftThreadId m+ raftFork m = do+ raftPostgresEnv <- ask+ lift $ raftFork (runRaftPostgresT' raftPostgresEnv m)+ -------------------------------------------------------------------------------- data EntryRow v = EntryRow@@ -288,5 +313,4 @@ createEntriesTable conn pure conn where- dbName = connectDatabase connInfo
src/Raft/Monad.hs view
@@ -1,192 +1,143 @@-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE FunctionalDependencies #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE TypeFamilyDependencies #-}-{-# LANGUAGE DefaultSignatures #-}-{-# LANGUAGE GADTs #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# Language ConstraintKinds #-}+{-# LANGUAGE AllowAmbiguousTypes #-} module Raft.Monad where -import Protolude hiding (pass)+import Protolude hiding (STM, TChan, readTChan, writeTChan, newTChan, atomically) -import Control.Arrow ((&&&))-import Control.Monad.RWS+import Control.Monad.Catch+import Control.Monad.Fail+import Control.Monad.Trans.Class+import qualified Control.Monad.Conc.Class as Conc -import qualified Data.Set as Set+import Control.Concurrent.Classy.STM.TChan -import Raft.Action-import Raft.Client import Raft.Config import Raft.Event-import Raft.Log-import Raft.Persistent+import Raft.Logging import Raft.NodeState-import Raft.RPC-import Raft.Types-import Raft.Logging (RaftLogger, runRaftLoggerT, RaftLoggerT(..), LogMsg)-import qualified Raft.Logging as Logging +import Test.DejaFu.Conc (ConcIO)+import qualified Test.DejaFu.Types as TDT ----------------------------------------------------------------------------------- Raft Transition Monad+-- Raft Monad Class -------------------------------------------------------------------------------- -tellAction :: Action sm v -> TransitionM sm v ()-tellAction a = tell [a]--tellActions :: [Action sm v] -> TransitionM sm v ()-tellActions as = tell as--data TransitionEnv sm v = TransitionEnv- { nodeConfig :: NodeConfig- , stateMachine :: sm- , nodeState :: RaftNodeState v- }--newtype TransitionM sm v a = TransitionM- { unTransitionM :: RaftLoggerT v (RWS (TransitionEnv sm v) [Action sm v] PersistentState) a- } deriving (Functor, Applicative, Monad)--instance MonadWriter [Action sm v] (TransitionM sm v) where- tell = TransitionM . RaftLoggerT . tell- listen = TransitionM . RaftLoggerT . listen . unRaftLoggerT . unTransitionM- pass = TransitionM . RaftLoggerT . pass . unRaftLoggerT . unTransitionM--instance MonadReader (TransitionEnv sm v) (TransitionM sm v) where- ask = TransitionM . RaftLoggerT $ ask- local f = TransitionM . RaftLoggerT . local f . unRaftLoggerT . unTransitionM+type MonadRaft v m = (MonadRaftChan v m, MonadRaftFork m) -instance MonadState PersistentState (TransitionM sm v) where- get = TransitionM . RaftLoggerT $ lift get- put = TransitionM . RaftLoggerT . lift . put+-- | The typeclass specifying the datatype used as the core event channel in the+-- main raft event loop, as well as functions for creating, reading, and writing+-- to the channel, and how to fork a computation that performs some action with+-- the channel.+--+-- Note: This module uses AllowAmbiguousTypes which removes the necessity for+-- Proxy value arguments in lieu of TypeApplication. For example:+--+-- @+-- newRaftChan @v+-- @+--+-- instead of+--+-- @+-- newRaftChan (Proxy :: Proxy v)+-- @+class Monad m => MonadRaftChan v m where+ type RaftEventChan v m+ readRaftChan :: RaftEventChan v m -> m (Event v)+ writeRaftChan :: RaftEventChan v m -> Event v -> m ()+ newRaftChan :: m (RaftEventChan v m) -instance RaftLogger v (RWS (TransitionEnv sm v) [Action sm v] PersistentState) where- loggerCtx = asks ((configNodeId . nodeConfig) &&& nodeState)+instance MonadRaftChan v IO where+ type RaftEventChan v IO = TChan (Conc.STM IO) (Event v)+ readRaftChan = Conc.atomically . readTChan+ writeRaftChan chan = Conc.atomically . writeTChan chan+ newRaftChan = Conc.atomically newTChan -runTransitionM- :: TransitionEnv sm v- -> PersistentState- -> TransitionM sm v a- -> ((a, [LogMsg]), PersistentState, [Action sm v])-runTransitionM transEnv persistentState transitionM =- runRWS (runRaftLoggerT (unTransitionM transitionM)) transEnv persistentState+instance MonadRaftChan v ConcIO where+ type RaftEventChan v ConcIO = TChan (Conc.STM ConcIO) (Event v)+ readRaftChan = Conc.atomically . readTChan+ writeRaftChan chan = Conc.atomically . writeTChan chan+ newRaftChan = Conc.atomically newTChan -askNodeId :: TransitionM sm v NodeId-askNodeId = asks (configNodeId . nodeConfig)+-- | The typeclass encapsulating the concurrency operations necessary for the+-- implementation of the main event handling loop.+--+-- This typeclass should not be manually defined.+class Monad m => MonadRaftFork m where+ type RaftThreadId m+ raftFork :: m () -> m (RaftThreadId m) ------------------------------------------------------------------------------------ Handlers---------------------------------------------------------------------------------+instance MonadRaftFork IO where+ type RaftThreadId IO = Protolude.ThreadId+ raftFork = forkIO -type RPCHandler ns sm r v = (RPCType r v, Show v) => NodeState ns v -> NodeId -> r -> TransitionM sm v (ResultState ns v)-type TimeoutHandler ns sm v = Show v => NodeState ns v -> Timeout -> TransitionM sm v (ResultState ns v)-type ClientReqHandler ns sm v = Show v => NodeState ns v -> ClientRequest v -> TransitionM sm v (ResultState ns v)+instance MonadRaftFork ConcIO where+ type RaftThreadId ConcIO = TDT.ThreadId+ raftFork = Conc.fork ----------------------------------------------------------------------------------- RWS Helpers+-- Raft Monad -------------------------------------------------------------------------------- -broadcast :: SendRPCAction v -> TransitionM sm v ()-broadcast sendRPC = do- selfNodeId <- askNodeId- tellAction =<<- flip BroadcastRPC sendRPC- <$> asks (Set.filter (selfNodeId /=) . configNodeIds . nodeConfig)--send :: NodeId -> SendRPCAction v -> TransitionM sm v ()-send nodeId sendRPC = tellAction (SendRPC nodeId sendRPC)---- | Resets the election timeout.-resetElectionTimeout :: TransitionM sm v ()-resetElectionTimeout = tellAction (ResetTimeoutTimer ElectionTimeout)--resetHeartbeatTimeout :: TransitionM sm v ()-resetHeartbeatTimeout = tellAction (ResetTimeoutTimer HeartbeatTimeout)--redirectClientToLeader :: ClientId -> CurrentLeader -> TransitionM sm v ()-redirectClientToLeader clientId currentLeader = do- let clientRedirRespSpec = ClientRedirRespSpec currentLeader- tellAction (RespondToClient clientId clientRedirRespSpec)--respondClientRead :: ClientId -> ClientReadReq -> TransitionM sm v ()-respondClientRead clientId readReq = do- readReqData <-- case readReq of- ClientReadEntries res -> pure (ClientReadRespSpecEntries res)- ClientReadStateMachine -> do- sm <- asks stateMachine- pure (ClientReadRespSpecStateMachine sm)- tellAction . RespondToClient clientId . ClientReadRespSpec $ readReqData-+-- | The raft server environment composed of the concurrent variables used in+-- the effectful raft layer.+data RaftEnv v m = RaftEnv+ { eventChan :: RaftEventChan v m+ , resetElectionTimer :: m ()+ , resetHeartbeatTimer :: m ()+ , raftNodeConfig :: RaftNodeConfig+ , raftNodeLogCtx :: LogCtx+ } -respondClientWrite :: ClientId -> Index -> SerialNum -> TransitionM sm v ()-respondClientWrite cid entryIdx sn =- tellAction (RespondToClient cid (ClientWriteRespSpec entryIdx sn))+newtype RaftT v m a = RaftT+ { unRaftT :: ReaderT (RaftEnv v m) (StateT (RaftNodeState v) m) a+ } deriving (Functor, Applicative, Monad, MonadReader (RaftEnv v m), MonadState (RaftNodeState v), MonadFail, Alternative, MonadPlus) -respondClientRedir :: ClientId -> CurrentLeader -> TransitionM sm v ()-respondClientRedir cid cl =- tellAction (RespondToClient cid (ClientRedirRespSpec cl))+instance MonadTrans (RaftT v) where+ lift = RaftT . lift . lift -appendLogEntries :: Show v => Seq (Entry v) -> TransitionM sm v ()-appendLogEntries = tellAction . AppendLogEntries+deriving instance MonadIO m => MonadIO (RaftT v m)+deriving instance MonadThrow m => MonadThrow (RaftT v m)+deriving instance MonadCatch m => MonadCatch (RaftT v m) -updateClientReqCacheFromIdx :: Index -> TransitionM sm v ()-updateClientReqCacheFromIdx = tellAction . UpdateClientReqCacheFrom+instance MonadRaftFork m => MonadRaftFork (RaftT v m) where+ type RaftThreadId (RaftT v m) = RaftThreadId m+ raftFork m = do+ raftEnv <- ask+ raftState <- get+ lift $ raftFork (runRaftT raftState raftEnv m) ---------------------------------------------------------------------------------+instance Monad m => RaftLogger v (RaftT v m) where+ loggerCtx = (,) <$> asks (configNodeId . raftNodeConfig) <*> get -startElection- :: Index- -> Index- -> LastLogEntry v- -> ClientWriteReqCache- -> TransitionM sm v (CandidateState v)-startElection commitIndex lastApplied lastLogEntry clientReqCache = do- incrementTerm- voteForSelf- resetElectionTimeout- broadcast =<< requestVoteMessage- selfNodeId <- askNodeId- -- Return new candidate state- pure CandidateState- { csCommitIndex = commitIndex- , csLastApplied = lastApplied- , csVotes = Set.singleton selfNodeId- , csLastLogEntry = lastLogEntry- , csClientReqCache = clientReqCache- }- where- requestVoteMessage = do- term <- currentTerm <$> get- selfNodeId <- askNodeId- pure $ SendRequestVoteRPC- RequestVote- { rvTerm = term- , rvCandidateId = selfNodeId- , rvLastLogIndex = lastLogEntryIndex lastLogEntry- , rvLastLogTerm = lastLogEntryTerm lastLogEntry- }+runRaftT+ :: Monad m+ => RaftNodeState v+ -> RaftEnv v m+ -> RaftT v m a+ -> m a+runRaftT raftNodeState raftEnv =+ flip evalStateT raftNodeState . flip runReaderT raftEnv . unRaftT - incrementTerm = do- psNextTerm <- incrTerm . currentTerm <$> get- modify $ \pstate ->- pstate { currentTerm = psNextTerm- , votedFor = Nothing- }+------------------------------------------------------------------------------+-- Logging+------------------------------------------------------------------------------ - voteForSelf = do- selfNodeId <- askNodeId- modify $ \pstate ->- pstate { votedFor = Just selfNodeId }+logDebug :: MonadIO m => Text -> RaftT v m ()+logDebug msg = flip logDebugIO msg =<< asks raftNodeLogCtx ------------------------------------------------------------------------------------ Logging---------------------------------------------------------------------------------+logCritical :: MonadIO m => Text -> RaftT v m ()+logCritical msg = flip logCriticalIO msg =<< asks raftNodeLogCtx -logInfo = TransitionM . Logging.logInfo-logDebug = TransitionM . Logging.logDebug+logAndPanic :: MonadIO m => Text -> RaftT v m a+logAndPanic msg = flip logAndPanicIO msg =<< asks raftNodeLogCtx
src/Raft/StateMachine.hs view
@@ -20,10 +20,13 @@ -------------------------------------------------------------------------------- -- | Interface to handle commands in the underlying state machine. Functional---dependency permitting only a single state machine command to be defined to---update the state machine.+-- dependency permitting only a single state machine command to be defined to+-- update the state machine. class RaftStateMachinePure sm v | sm -> v where data RaftStateMachinePureError sm v+ -- This type family dependency is to make the 'RaftStateMachinePureCtx` type+ -- family injective; i.e. to allow GHC to infer the state machine and command+ -- types from values of type 'RaftStateMachinePureCtx sm v'. type RaftStateMachinePureCtx sm v = ctx | ctx -> sm v rsmTransition :: RaftStateMachinePureCtx sm v@@ -31,7 +34,7 @@ -> v -> Either (RaftStateMachinePureError sm v) sm -class (Monad m, RaftStateMachinePure sm v) => RaftStateMachine m sm v | m sm -> v where+class (Monad m, RaftStateMachinePure sm v) => RaftStateMachine m sm v where validateCmd :: v -> m (Either (RaftStateMachinePureError sm v) ()) askRaftStateMachinePureCtx :: m (RaftStateMachinePureCtx sm v)
+ src/Raft/Transition.hs view
@@ -0,0 +1,192 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeFamilyDependencies #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE GADTs #-}++module Raft.Transition where++import Protolude hiding (pass)++import Control.Arrow ((&&&))+import Control.Monad.RWS++import qualified Data.Set as Set++import Raft.Action+import Raft.Client+import Raft.Config+import Raft.Event+import Raft.Log+import Raft.Persistent+import Raft.NodeState+import Raft.RPC+import Raft.Types+import Raft.Logging (RaftLogger, runRaftLoggerT, RaftLoggerT(..), LogMsg)+import qualified Raft.Logging as Logging+++--------------------------------------------------------------------------------+-- Raft Transition Monad+--------------------------------------------------------------------------------++tellAction :: Action sm v -> TransitionM sm v ()+tellAction a = tell [a]++tellActions :: [Action sm v] -> TransitionM sm v ()+tellActions as = tell as++data TransitionEnv sm v = TransitionEnv+ { nodeConfig :: RaftNodeConfig+ , stateMachine :: sm+ , nodeState :: RaftNodeState v+ }++newtype TransitionM sm v a = TransitionM+ { unTransitionM :: RaftLoggerT v (RWS (TransitionEnv sm v) [Action sm v] PersistentState) a+ } deriving (Functor, Applicative, Monad)++instance MonadWriter [Action sm v] (TransitionM sm v) where+ tell = TransitionM . RaftLoggerT . tell+ listen = TransitionM . RaftLoggerT . listen . unRaftLoggerT . unTransitionM+ pass = TransitionM . RaftLoggerT . pass . unRaftLoggerT . unTransitionM++instance MonadReader (TransitionEnv sm v) (TransitionM sm v) where+ ask = TransitionM . RaftLoggerT $ ask+ local f = TransitionM . RaftLoggerT . local f . unRaftLoggerT . unTransitionM++instance MonadState PersistentState (TransitionM sm v) where+ get = TransitionM . RaftLoggerT $ lift get+ put = TransitionM . RaftLoggerT . lift . put++instance RaftLogger v (RWS (TransitionEnv sm v) [Action sm v] PersistentState) where+ loggerCtx = asks ((configNodeId . nodeConfig) &&& nodeState)++runTransitionM+ :: TransitionEnv sm v+ -> PersistentState+ -> TransitionM sm v a+ -> ((a, [LogMsg]), PersistentState, [Action sm v])+runTransitionM transEnv persistentState transitionM =+ runRWS (runRaftLoggerT (unTransitionM transitionM)) transEnv persistentState++askNodeId :: TransitionM sm v NodeId+askNodeId = asks (configNodeId . nodeConfig)++--------------------------------------------------------------------------------+-- Handlers+--------------------------------------------------------------------------------++type RPCHandler ns sm r v = (RPCType r v, Show v) => NodeState ns v -> NodeId -> r -> TransitionM sm v (ResultState ns v)+type TimeoutHandler ns sm v = Show v => NodeState ns v -> Timeout -> TransitionM sm v (ResultState ns v)+type ClientReqHandler ns sm v = Show v => NodeState ns v -> ClientRequest v -> TransitionM sm v (ResultState ns v)++--------------------------------------------------------------------------------+-- RWS Helpers+--------------------------------------------------------------------------------++broadcast :: SendRPCAction v -> TransitionM sm v ()+broadcast sendRPC = do+ selfNodeId <- askNodeId+ tellAction =<<+ flip BroadcastRPC sendRPC+ <$> asks (Set.filter (selfNodeId /=) . configNodeIds . nodeConfig)++send :: NodeId -> SendRPCAction v -> TransitionM sm v ()+send nodeId sendRPC = tellAction (SendRPC nodeId sendRPC)++-- | Resets the election timeout.+resetElectionTimeout :: TransitionM sm v ()+resetElectionTimeout = tellAction (ResetTimeoutTimer ElectionTimeout)++resetHeartbeatTimeout :: TransitionM sm v ()+resetHeartbeatTimeout = tellAction (ResetTimeoutTimer HeartbeatTimeout)++redirectClientToLeader :: ClientId -> CurrentLeader -> TransitionM sm v ()+redirectClientToLeader clientId currentLeader = do+ let clientRedirRespSpec = ClientRedirRespSpec currentLeader+ tellAction (RespondToClient clientId clientRedirRespSpec)++respondClientRead :: ClientId -> ClientReadReq -> TransitionM sm v ()+respondClientRead clientId readReq = do+ readReqData <-+ case readReq of+ ClientReadEntries res -> pure (ClientReadRespSpecEntries res)+ ClientReadStateMachine -> do+ sm <- asks stateMachine+ pure (ClientReadRespSpecStateMachine sm)+ tellAction . RespondToClient clientId . ClientReadRespSpec $ readReqData+++respondClientWrite :: ClientId -> Index -> SerialNum -> TransitionM sm v ()+respondClientWrite cid entryIdx sn =+ tellAction (RespondToClient cid (ClientWriteRespSpec entryIdx sn))++respondClientRedir :: ClientId -> CurrentLeader -> TransitionM sm v ()+respondClientRedir cid cl =+ tellAction (RespondToClient cid (ClientRedirRespSpec cl))++appendLogEntries :: Show v => Seq (Entry v) -> TransitionM sm v ()+appendLogEntries = tellAction . AppendLogEntries++updateClientReqCacheFromIdx :: Index -> TransitionM sm v ()+updateClientReqCacheFromIdx = tellAction . UpdateClientReqCacheFrom++--------------------------------------------------------------------------------++startElection+ :: Index+ -> Index+ -> LastLogEntry v+ -> ClientWriteReqCache+ -> TransitionM sm v (CandidateState v)+startElection commitIndex lastApplied lastLogEntry clientReqCache = do+ incrementTerm+ voteForSelf+ resetElectionTimeout+ broadcast =<< requestVoteMessage+ selfNodeId <- askNodeId+ -- Return new candidate state+ pure CandidateState+ { csCommitIndex = commitIndex+ , csLastApplied = lastApplied+ , csVotes = Set.singleton selfNodeId+ , csLastLogEntry = lastLogEntry+ , csClientReqCache = clientReqCache+ }+ where+ requestVoteMessage = do+ term <- currentTerm <$> get+ selfNodeId <- askNodeId+ pure $ SendRequestVoteRPC+ RequestVote+ { rvTerm = term+ , rvCandidateId = selfNodeId+ , rvLastLogIndex = lastLogEntryIndex lastLogEntry+ , rvLastLogTerm = lastLogEntryTerm lastLogEntry+ }++ incrementTerm = do+ psNextTerm <- incrTerm . currentTerm <$> get+ modify $ \pstate ->+ pstate { currentTerm = psNextTerm+ , votedFor = Nothing+ }++ voteForSelf = do+ selfNodeId <- askNodeId+ modify $ \pstate ->+ pstate { votedFor = Just selfNodeId }++--------------------------------------------------------------------------------+-- Logging+--------------------------------------------------------------------------------++logInfo = TransitionM . Logging.logInfo+logDebug = TransitionM . Logging.logDebug
test/TestDejaFu.hs view
@@ -37,8 +37,9 @@ import TestUtils import Raft-import Raft.Log import Raft.Client+import Raft.Log+import Raft.Monad import Data.Time.Clock.System (getSystemTime) @@ -71,7 +72,7 @@ validateCmd _ = pure (Right ()) askRaftStateMachinePureCtx = pure StoreCtx -type TestEventChan = EventChan ConcIO StoreCmd+type TestEventChan = RaftEventChan StoreCmd RaftTestM type TestEventChans = Map NodeId TestEventChan type TestClientRespChan = TChan (STM ConcIO) (ClientResponse Store StoreCmd)@@ -81,7 +82,7 @@ data TestNodeEnv = TestNodeEnv { testNodeEventChans :: TestEventChans , testClientRespChans :: TestClientRespChans- , testNodeConfig :: NodeConfig+ , testRaftNodeConfig :: RaftNodeConfig } -- | Node specific state@@ -113,7 +114,7 @@ throwTestErr = throw . RaftTestError askSelfNodeId :: RaftTestM NodeId-askSelfNodeId = asks (configNodeId . testNodeConfig)+askSelfNodeId = asks (configNodeId . testRaftNodeConfig) lookupNodeEventChan :: NodeId -> RaftTestM TestEventChan lookupNodeEventChan nid = do@@ -207,6 +208,19 @@ Empty -> pure (Right Nothing) _ :|> lastEntry -> pure (Right (Just lastEntry)) +instance MonadRaftChan StoreCmd RaftTestM where+ type RaftEventChan StoreCmd RaftTestM = TChan (STM ConcIO) (Event StoreCmd)+ readRaftChan = RaftTestM . lift . lift . readRaftChan+ writeRaftChan chan = RaftTestM . lift . lift . writeRaftChan chan+ newRaftChan = RaftTestM . lift . lift $ newRaftChan++instance MonadRaftFork RaftTestM where+ type RaftThreadId RaftTestM = RaftThreadId ConcIO+ raftFork m = do+ testNodeEnv <- ask+ testNodeStates <- get+ RaftTestM . lift . lift $ raftFork (runRaftTestM testNodeEnv testNodeStates m)+ -------------------------------------------------------------------------------- data TestClientEnv = TestClientEnv@@ -272,9 +286,9 @@ runRaftT initRaftNodeState raftEnv $ handleEventLoop (mempty :: Store) where- nid = configNodeId (testNodeConfig testEnv)+ nid = configNodeId (testRaftNodeConfig testEnv) Just eventChan = Map.lookup nid (testNodeEventChans testEnv)- raftEnv = RaftEnv eventChan dummyTimer dummyTimer (testNodeConfig testEnv) NoLogs+ raftEnv = RaftEnv eventChan dummyTimer dummyTimer (testRaftNodeConfig testEnv) NoLogs dummyTimer = pure () forkTestNodes :: [TestNodeEnv] -> TestNodeStates -> ConcIO [ThreadId ConcIO]
test/TestUtils.hs view
@@ -45,7 +45,7 @@ nodeIds :: NodeIds nodeIds = Set.fromList [node0, node1, node2] -testConfigs :: [NodeConfig]+testConfigs :: [RaftNodeConfig] testConfigs = [testConfig0, testConfig1, testConfig2] msToMicroS :: Num n => n -> n@@ -54,22 +54,22 @@ pairMsToMicroS :: Num n => (n, n) -> (n, n) pairMsToMicroS = bimap msToMicroS msToMicroS -testConfig0, testConfig1, testConfig2 :: NodeConfig-testConfig0 = NodeConfig+testConfig0, testConfig1, testConfig2 :: RaftNodeConfig+testConfig0 = RaftNodeConfig { configNodeId = node0 , configNodeIds = nodeIds , configElectionTimeout = pairMsToMicroS (150, 300) , configHeartbeatTimeout = msToMicroS 50 , configStorageState = New }-testConfig1 = NodeConfig+testConfig1 = RaftNodeConfig { configNodeId = node1 , configNodeIds = nodeIds , configElectionTimeout = pairMsToMicroS (150, 300) , configHeartbeatTimeout = msToMicroS 50 , configStorageState = New }-testConfig2 = NodeConfig+testConfig2 = RaftNodeConfig { configNodeId = node2 , configNodeIds = nodeIds , configElectionTimeout = pairMsToMicroS (150, 300)