diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -208,7 +208,7 @@ `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 +correct, assuming "correct" implementations of the `MonadRaftFork` and `MonadRaftChan` typeclasses. #### Persistent State @@ -357,20 +357,22 @@ ### 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: +they are running the raft node in is a `MonadRaftFork`, which provides the main +raft loop with the ability to fork a concurrent action; The raft node needs to +know how to fork actions in the monad. This is necessary for the raft node to +be able to fork its event producers, and run other actions concurrently; e.g. +a leader responding to all followers at the same time during a heartbeat RPC +broadcast. 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 typeclass encapsulating the concurrency operations necessary for the +-- implementation of the main event handling loop. +class Monad m => MonadRaftFork m where + type RaftThreadId m + raftFork + :: RaftThreadRole -- ^ The role of the current thread being forked + -> m () -- ^ The action to fork + -> m (RaftThreadId m) ``` The implementation of this typeclass is a bit subtle, and it is advised that @@ -381,13 +383,19 @@ 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) +instance MonadRaftFork MyMonad where + type RaftThreadId MyMonad = RaftThreadId IO + raftFork threadRole myMonad = + lift $ raftFork threadRole (runMyMonad myMonad) ``` +The last thing to mention about this typeclass is the `RaftThreadRole` value +that must be passed to invocations of the `raftFork` function. In some +applications (and, noteably our concurrency testing suite) thread names can be +used for debugging and even message passing purposes. For instance of +`MonadRaftFork` that do not need to distinguish threads by name, simply ignore +the argument. + # The Raft Example (`raft-example`) In this library we provide a full fledged, non-production ready, example @@ -396,7 +404,7 @@ `src/Examples/Raft/...` or in `app/Main.hs`: `RaftExampleT` (found in `app/Main.hs`): -- `MonadRaftAsync` +- `MonadRaftFork` - `MonadRaftChan` - `RaftStateMachine` diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -98,7 +98,7 @@ instance (MonadIO m, MonadRaftFork m) => MonadRaftFork (RaftExampleT m) where type RaftThreadId (RaftExampleT m) = RaftThreadId m - raftFork m = lift $ raftFork (runRaftExampleT m) + raftFork r m = lift $ raftFork r (runRaftExampleT m) runRaftExampleT :: RaftExampleT m a -> m a runRaftExampleT = unRaftExampleT diff --git a/libraft.cabal b/libraft.cabal --- a/libraft.cabal +++ b/libraft.cabal @@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 2eb0f5597b5304ede9da9c1a97cc9bc16e5df0eb384c9f1ce56e8b1dc122f5b9 +-- hash: ff992f3d3136b159258f789e3155e8842a492ac71a9863ae998fa57526e5b9ec name: libraft -version: 0.3.0.0 +version: 0.4.0.0 synopsis: Raft consensus algorithm description: Please see the README on GitHub at category: Distributed Systems diff --git a/src/Examples/Raft/FileStore/Log.hs b/src/Examples/Raft/FileStore/Log.hs --- a/src/Examples/Raft/FileStore/Log.hs +++ b/src/Examples/Raft/FileStore/Log.hs @@ -138,6 +138,6 @@ instance (MonadIO m, MonadRaftFork m) => MonadRaftFork (RaftLogFileStoreT m) where type RaftThreadId (RaftLogFileStoreT m) = RaftThreadId m - raftFork m = do + raftFork r m = do raftLogFile <- ask - lift $ raftFork (runRaftLogFileStoreT raftLogFile m) + lift $ raftFork r (runRaftLogFileStoreT raftLogFile m) diff --git a/src/Examples/Raft/FileStore/Persistent.hs b/src/Examples/Raft/FileStore/Persistent.hs --- a/src/Examples/Raft/FileStore/Persistent.hs +++ b/src/Examples/Raft/FileStore/Persistent.hs @@ -133,6 +133,6 @@ instance (MonadIO m, MonadRaftFork m) => MonadRaftFork (RaftPersistFileStoreT m) where type RaftThreadId (RaftPersistFileStoreT m) = RaftThreadId m - raftFork m = do + raftFork r m = do persistFile <- ask - lift $ raftFork (runRaftPersistFileStoreT persistFile m) + lift $ raftFork r (runRaftPersistFileStoreT persistFile m) diff --git a/src/Examples/Raft/Socket/Node.hs b/src/Examples/Raft/Socket/Node.hs --- a/src/Examples/Raft/Socket/Node.hs +++ b/src/Examples/Raft/Socket/Node.hs @@ -151,6 +151,6 @@ instance (MonadIO m, MonadRaftFork m) => MonadRaftFork (RaftSocketT v m) where type RaftThreadId (RaftSocketT v m) = RaftThreadId m - raftFork m = do + raftFork r m = do persistFile <- ask - lift $ raftFork (runRaftSocketT persistFile m) + lift $ raftFork r (runRaftSocketT persistFile m) diff --git a/src/Raft.hs b/src/Raft.hs --- a/src/Raft.hs +++ b/src/Raft.hs @@ -168,10 +168,18 @@ runRaftT initRaftNodeState raftEnv $ do -- 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) + -- + -- Note: Changing the roles of these event producers (the strings passed as + -- arguments to 'raftFork' should incur a minor (or major?) version bump, + -- because some implementations of 'MonadRaftFork' rely on these strings to + -- reliably send messages to the event producers (e.g. cloud-haskell + -- processes). + raftFork (CustomThreadRole "Election Timeout Timer") . lift $ + electionTimeoutTimer @m @v eventChan electionTimer + raftFork (CustomThreadRole "Heartbeat Timeout Timer") . lift $ + heartbeatTimeoutTimer @m @v eventChan heartbeatTimer + raftFork RPCHandler (rpcHandler @m @v eventChan) + raftFork ClientRequestHandler (clientReqHandler @m @v eventChan) -- Start the main event handling loop handleEventLoop initRaftStateMachine @@ -311,16 +319,16 @@ lift (sendRPC nid rpcMsg) SendRPCs rpcMap -> flip mapM_ (Map.toList rpcMap) $ \(nid, sendRpcAction) -> - raftFork $ do + raftFork (CustomThreadRole "Send RPC") $ do rpcMsg <- mkRPCfromSendRPCAction sendRpcAction lift (sendRPC nid rpcMsg) BroadcastRPC nids sendRpcAction -> do rpcMsg <- mkRPCfromSendRPCAction sendRpcAction - mapM_ (raftFork . lift . flip sendRPC rpcMsg) nids + mapM_ (raftFork (CustomThreadRole "RPC Broadcast Thread") . lift . flip sendRPC rpcMsg) nids RespondToClient cid cr -> do clientResp <- mkClientResp cr -- TODO log failure if sendClient fails - void $ raftFork $ lift $ sendClient cid clientResp + void $ raftFork (CustomThreadRole "Respond to Client") $ lift $ sendClient cid clientResp ResetTimeoutTimer tout -> do case tout of ElectionTimeout -> lift . resetElectionTimer =<< ask diff --git a/src/Raft/Log/PostgreSQL.hs b/src/Raft/Log/PostgreSQL.hs --- a/src/Raft/Log/PostgreSQL.hs +++ b/src/Raft/Log/PostgreSQL.hs @@ -203,9 +203,9 @@ instance (MonadIO m, MonadRaftFork m) => MonadRaftFork (RaftPostgresT m) where type RaftThreadId (RaftPostgresT m) = RaftThreadId m - raftFork m = do + raftFork r m = do raftPostgresEnv <- ask - lift $ raftFork (runRaftPostgresT' raftPostgresEnv m) + lift $ raftFork r (runRaftPostgresT' raftPostgresEnv m) -------------------------------------------------------------------------------- diff --git a/src/Raft/Monad.hs b/src/Raft/Monad.hs --- a/src/Raft/Monad.hs +++ b/src/Raft/Monad.hs @@ -69,21 +69,28 @@ writeRaftChan chan = Conc.atomically . writeTChan chan newRaftChan = Conc.atomically newTChan +data RaftThreadRole + = RPCHandler + | ClientRequestHandler + | CustomThreadRole Text + deriving Show + -- | 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) + raftFork + :: RaftThreadRole -- ^ The role of the current thread being forked + -> m () -- ^ The action to fork + -> m (RaftThreadId m) instance MonadRaftFork IO where type RaftThreadId IO = Protolude.ThreadId - raftFork = forkIO + raftFork _ = forkIO instance MonadRaftFork ConcIO where type RaftThreadId ConcIO = TDT.ThreadId - raftFork = Conc.fork + raftFork r = Conc.forkN (show r) -------------------------------------------------------------------------------- -- Raft Monad @@ -112,10 +119,10 @@ instance MonadRaftFork m => MonadRaftFork (RaftT v m) where type RaftThreadId (RaftT v m) = RaftThreadId m - raftFork m = do + raftFork s m = do raftEnv <- ask raftState <- get - lift $ raftFork (runRaftT raftState raftEnv m) + lift $ raftFork s (runRaftT raftState raftEnv m) instance Monad m => RaftLogger v (RaftT v m) where loggerCtx = (,) <$> asks (configNodeId . raftNodeConfig) <*> get diff --git a/test/TestDejaFu.hs b/test/TestDejaFu.hs --- a/test/TestDejaFu.hs +++ b/test/TestDejaFu.hs @@ -216,10 +216,10 @@ instance MonadRaftFork RaftTestM where type RaftThreadId RaftTestM = RaftThreadId ConcIO - raftFork m = do + raftFork r m = do testNodeEnv <- ask testNodeStates <- get - RaftTestM . lift . lift $ raftFork (runRaftTestM testNodeEnv testNodeStates m) + RaftTestM . lift . lift $ raftFork r (runRaftTestM testNodeEnv testNodeStates m) --------------------------------------------------------------------------------