libraft 0.4.0.0 → 0.4.1.0
raw patch · 9 files changed
+155/−94 lines, 9 files
Files
- ChangeLog.md +13/−0
- app/Main.hs +5/−3
- libraft.cabal +2/−2
- src/Examples/Raft/Socket/Client.hs +13/−18
- src/Examples/Raft/Socket/Node.hs +92/−51
- src/Raft.hs +1/−1
- src/Raft/Logging.hs +18/−13
- src/Raft/Monad.hs +1/−1
- test/QuickCheckStateMachine.hs +10/−5
ChangeLog.md view
@@ -1,5 +1,18 @@ # Changelog for raft +## 0.4.1.0++- Improvement: Users can now supply an existing logging function to log internal+ raft node logs, useful for integration into existing applications.+- Improvement: The example instances for `RaftClientSend` and+ `RaftClientRecv` using unix sockets no longer need to fork a server to handle+ responses from the raft node; the response is sent on the same socket the+ client opens while sending the request.++## 0.4.0.0++- API change: `MonadRaftAsync` is now `MonadRaftFork`, with a simpler API+ ## 0.3.0.0 - API change: `runRaftNode` now requires the monad it runs in to provide an
app/Main.hs view
@@ -18,6 +18,7 @@ import Control.Concurrent.Lifted (fork) import Control.Concurrent.STM.TChan+import Control.Concurrent.STM.TVar import Control.Monad.Fail import Control.Monad.Catch@@ -208,7 +209,7 @@ , Exception (RaftWriteLogError m), Exception (RaftDeleteLogError m) , Typeable m )- => NodeSocketEnv StoreCmd+ => NodeSocketEnv Store StoreCmd -> RaftPersistFile -> m () runRaftNode' nSocketEnv nPersistFile =@@ -247,13 +248,15 @@ tmpDir <- Directory.getTemporaryDirectory pure (tmpDir ++ "/" ++ toS nid ++ "/" ++ "logs") - initSocketEnv :: NodeId -> IO (NodeSocketEnv v)+ initSocketEnv :: NodeId -> IO (NodeSocketEnv sm v) initSocketEnv nid = do msgQueue <- atomically newTChan clientReqQueue <- atomically newTChan+ clientReqResps <- atomically $ newTVar Map.empty pure NodeSocketEnv { nsMsgQueue = msgQueue , nsClientReqQueue = clientReqQueue+ , nsClientReqResps = clientReqResps } mkExampleDir :: NodeId -> IO FilePath@@ -269,7 +272,6 @@ let clientId = ClientId $ RS.hostPortToNid (clientHost, clientPort) clientRespChan <- RS.newClientRespChan RS.runRaftSocketClientM clientId mempty clientRespChan $ do- fork (lift (RS.clientResponseServer clientHost clientPort)) evalRepl (pure ">>> ") (unConsoleM . handleConsoleCmd) [] Nothing (Word completer) (pure ()) -- Tab Completion: return a completion for partial words entered
libraft.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: ff992f3d3136b159258f789e3155e8842a492ac71a9863ae998fa57526e5b9ec+-- hash: 1ae69f1e10ee0c8f4df745d5d10ec04cf4768915f3c9e5ed05af531fe5d717b0 name: libraft-version: 0.4.0.0+version: 0.4.1.0 synopsis: Raft consensus algorithm description: Please see the README on GitHub at <https://github.com/adjoint-io/raft#readme> category: Distributed Systems
src/Examples/Raft/Socket/Client.hs view
@@ -73,17 +73,27 @@ let run' = RunIO (fmap (RaftClientRespChanT . ReaderT . const) . run . flip runReaderT r . unRaftClientRespChanT) in fmap (flip runReaderT r . unRaftClientRespChanT) $ f run' -instance (S.Serialize v, MonadIO m) => RaftClientSend (RaftClientRespChanT s v m) v where+instance (S.Serialize s, S.Serialize v, MonadIO m) => RaftClientSend (RaftClientRespChanT s v m) v where type RaftClientSendError (RaftClientRespChanT s v m) v = Text raftClientSend nid creq = do+ respChan <- asks clientRespChan let (host,port) = nidToHostPort nid+ errPrefix = "Failed to send client request: " mRes <- liftIO $ timeout 100000 $ try $ do -- Warning: blocks if socket is allocated by OS, even though the socket -- may have been closed by the running process- N.connect host port $ \(sock, sockAddr) ->+ N.connect host port $ \(sock, _sockAddr) -> do N.send sock (S.encode (ClientRequestEvent creq))- let errPrefix = "Failed to send client request: "+ mResp <- recvSerialized sock+ case mResp of+ Nothing+ -> pure . Left+ $ errPrefix <> "Socket was closed or client server failed to decode message"+ Just resp+ -> do+ atomically $ writeTChan respChan resp+ pure . pure $ () case mRes of Nothing -> pure (Left (errPrefix <> "'connect' timed out")) Just (Left (err :: SomeException)) -> pure $ Left (errPrefix <> show err)@@ -112,21 +122,6 @@ . unRaftClientRespChanT . runRaftClientT raftClientEnv raftClientState $ rscm--clientResponseServer- :: forall s v m. (S.Serialize s, S.Serialize v, MonadIO m)- => N.HostName- -> N.ServiceName- -> RaftClientRespChanT s v m ()-clientResponseServer host port = do- respChan <- asks clientRespChan- N.serve (N.Host host) port $ \(sock, _) -> do- mBytes <- N.recv sock (4 * 4096)- case mBytes of- Nothing -> putText "Socket was closed on the other end"- Just bytes -> case S.decode bytes of- Left err -> putText $ "Failed to decode message: " <> toS err- Right cresp -> atomically $ writeTChan respChan cresp -- | Send a client read request using the example socket interface of RaftSocketClientM socketClientRead
src/Examples/Raft/Socket/Node.hs view
@@ -18,10 +18,13 @@ import Control.Monad.Catch import Control.Monad.Trans.Class import Control.Concurrent.STM.TChan+import Control.Concurrent.STM.TVar+import Control.Concurrent.STM.TMVar +import qualified Data.Map as Map import qualified Data.Serialize as S import qualified Network.Simple.TCP as NS-import Network.Simple.TCP+import Network.Simple.TCP (HostName, ServiceName) import Examples.Raft.Socket.Common @@ -38,42 +41,58 @@ -- Network -------------------------------------------------------------------------------- -data NodeSocketEnv v = NodeSocketEnv+data ResponseSignal sm v+ = OkResponse (ClientResponse sm v)+ -- ^ we managed to write a valid response to the @TMVar@+ | DeadResponse+ -- ^ if we get overlapping requests coming in with the same client+ -- id, we "kill" one of them++data NodeSocketEnv sm v = NodeSocketEnv { nsMsgQueue :: TChan (RPCMessage v)+ -- ^ Queue of RPC messages to be processed by event handlers , nsClientReqQueue :: TChan (ClientRequest v)+ -- ^ Queue of client request messages to be processed by event handlers+ , nsClientReqResps :: TVar (Map ClientId (TMVar (ResponseSignal sm v)))+ -- ^ Map of variables to which responses to a request are written. N.B.:+ -- this assumes a client id uniquely identifies a request; A client+ -- will never send a request without having either 1) given up on the+ -- a previous request because of a timeout, or 2) received a response to+ -- each previous request issued. } -newtype RaftSocketT v m a = RaftSocketT { unRaftSocketT :: ReaderT (NodeSocketEnv v) m a }- deriving (Functor, Applicative, Monad, MonadIO, MonadFail, MonadReader (NodeSocketEnv v), Alternative, MonadPlus, MonadTrans)+newtype RaftSocketT sm v m a = RaftSocketT { unRaftSocketT :: ReaderT (NodeSocketEnv sm v) m a }+ deriving (Functor, Applicative, Monad, MonadIO, MonadFail, MonadReader (NodeSocketEnv sm v), Alternative, MonadPlus, MonadTrans) -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 MonadThrow m => MonadThrow (RaftSocketT sm v m)+deriving instance MonadCatch m => MonadCatch (RaftSocketT sm v m)+deriving instance MonadMask m => MonadMask (RaftSocketT sm v m) -------------------- -- Raft Instances -- -------------------- -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 $- connect cHost cPort $ \(cSock, _cSockAddr) ->- send cSock (S.encode msg)- case eRes of- Left (err :: SomeException) -> putText ("Failed to send Client: " <> show err)- Right _ -> pure ()+instance (MonadMask m, MonadCatch m, MonadIO m, S.Serialize sm, S.Serialize v) => RaftSendClient (RaftSocketT sm v m) sm v where+ sendClient clientId msg = do+ NodeSocketEnv{..} <- ask+ mRespVar <- liftIO . atomically . fmap (Map.lookup clientId) . readTVar $ nsClientReqResps+ -- We write the response to the TMVar corresponding to the client+ -- id, such that @acceptConnections@ can send it back to the+ -- client.+ case mRespVar of+ Nothing -> liftIO $ putText "sendClient: response lookup failed"+ Just respVar -> liftIO . atomically . putTMVar respVar . OkResponse $ msg -instance (MonadIO m, S.Serialize v) => RaftRecvClient (RaftSocketT v m) v where- type RaftRecvClientError (RaftSocketT v m) v = Text+instance (MonadIO m, S.Serialize v) => RaftRecvClient (RaftSocketT sm v m) v where+ type RaftRecvClientError (RaftSocketT sm v m) v = Text receiveClient = do cReq <- asks nsClientReqQueue fmap Right . liftIO . atomically $ readTChan cReq -instance (MonadCatch m, MonadMask m, MonadIO 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 sm v m) v where sendRPC nid msg = do eRes <- Control.Monad.Catch.try $- connect host port $ \(sock,_) -> do+ NS.connect host port $ \(sock,_) -> do NS.send sock (S.encode $ RPCMessageEvent msg) case eRes of Left (err :: SomeException) -> putText ("Failed to send RPC: " <> show err)@@ -81,76 +100,98 @@ where (host, port) = nidToHostPort nid -instance (MonadIO m, Show v) => RaftRecvRPC (RaftSocketT v m) v where- type RaftRecvRPCError (RaftSocketT v m) v = Text+instance (MonadIO m, Show v) => RaftRecvRPC (RaftSocketT sm v m) v where+ type RaftRecvRPCError (RaftSocketT sm v m) v = Text receiveRPC = do msgQueue <- asks nsMsgQueue fmap Right . liftIO . atomically $ readTChan msgQueue -runRaftSocketT :: MonadIO m => NodeSocketEnv v -> RaftSocketT v m a -> m a+runRaftSocketT :: MonadIO m => NodeSocketEnv sm v -> RaftSocketT sm v m a -> m a runRaftSocketT nodeSocketEnv = flip runReaderT nodeSocketEnv . unRaftSocketT acceptConnections- :: forall v m. (S.Serialize v, MonadIO m)+ :: forall sm v m. (S.Serialize sm, S.Serialize v, MonadIO m) => HostName -> ServiceName- -> RaftSocketT v m ()+ -> RaftSocketT sm v m () acceptConnections host port = do- socketEnv@NodeSocketEnv{..} <- ask- serve (Host host) port $ \(sock, _) -> do+ NodeSocketEnv{..} <- ask+ NS.serve (NS.Host host) port $ \(sock, _) -> do mVal <- recvSerialized sock case mVal of Nothing -> putText "Socket was closed on the other end"- Just val ->- case val of- ClientRequestEvent req ->- atomically $ writeTChan nsClientReqQueue req- RPCMessageEvent msg ->- atomically $ writeTChan nsMsgQueue msg+ Just (ClientRequestEvent req@(ClientRequest clientId _)) -> do+ -- Create and register TMVar where the response should be+ -- written to for this client id in the 'sendClient' impl+ respVar <- atomically $ do+ newRespVar <- newEmptyTMVar+ clientReqResps <- readTVar nsClientReqResps+ -- If there's an outstanding request for this+ -- client id, we send a signal that the request+ -- is "dead". Given sufficiently unique client+ -- ids, this case should never occur.+ when (Map.member clientId clientReqResps) $+ case Map.lookup clientId clientReqResps of+ Nothing -> pure ()+ Just reqVar -> putTMVar reqVar DeadResponse -newSock :: HostName -> ServiceName -> IO Socket-newSock host port =- listen (Host host) port (pure . fst)+ writeTVar nsClientReqResps . Map.insert clientId newRespVar $ clientReqResps+ pure newRespVar+ -- Register request to be handled by event handler+ atomically $ writeTChan nsClientReqQueue req+ -- Wait until response has been written to the TMVar by a 'sendClient'+ -- call and send the response to the client.+ respMsg <- atomically $ takeTMVar respVar+ case respMsg of+ OkResponse okResp -> NS.send sock (S.encode okResp)+ DeadResponse -> pure () -- ignored for now + -- Remove response variable for the client id+ atomically $ do+ modifyTVar nsClientReqResps $ \_nsClientReqResps ->+ Map.delete clientId _nsClientReqResps+ Just (RPCMessageEvent msg) ->+ atomically $ writeTChan nsMsgQueue msg+ -------------------------------------------------------------------------------- -- Inherited Instances -------------------------------------------------------------------------------- -instance (MonadIO m, RaftPersist m) => RaftPersist (RaftSocketT v m) where- type RaftPersistError (RaftSocketT v m) = RaftPersistError m+instance (MonadIO m, RaftPersist m) => RaftPersist (RaftSocketT sm v m) where+ type RaftPersistError (RaftSocketT sm 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+instance (MonadIO m, RaftInitLog m v) => RaftInitLog (RaftSocketT sm v m) v where+ type RaftInitLogError (RaftSocketT sm 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+instance RaftWriteLog m v => RaftWriteLog (RaftSocketT sm v m) v where+ type RaftWriteLogError (RaftSocketT sm 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+instance RaftReadLog m v => RaftReadLog (RaftSocketT sm v m) v where+ type RaftReadLogError (RaftSocketT sm 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+instance RaftDeleteLog m v => RaftDeleteLog (RaftSocketT sm v m) v where+ type RaftDeleteLogError (RaftSocketT sm v m) = RaftDeleteLogError m deleteLogEntriesFrom idx = lift $ deleteLogEntriesFrom idx -instance RaftStateMachine m sm v => RaftStateMachine (RaftSocketT v m) sm v where+instance RaftStateMachine m sm v => RaftStateMachine (RaftSocketT sm 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+instance MonadRaftChan v m => MonadRaftChan v (RaftSocketT sm v m) where+ type RaftEventChan v (RaftSocketT sm 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+instance (MonadIO m, MonadRaftFork m) => MonadRaftFork (RaftSocketT sm v m) where+ type RaftThreadId (RaftSocketT sm v m) = RaftThreadId m raftFork r m = do persistFile <- ask lift $ raftFork r (runRaftSocketT persistFile m)
src/Raft.hs view
@@ -148,7 +148,7 @@ , Exception (RaftPersistError m) ) => RaftNodeConfig -- ^ Node configuration- -> LogCtx -- ^ Logs destination+ -> LogCtx (RaftT v m) -- ^ Logs destination -> Int -- ^ Timer seed -> sm -- ^ Initial state machine state -> m ()
src/Raft/Logging.hs view
@@ -21,16 +21,17 @@ import Raft.Types -- | Representation of the logs' context-data LogCtx+data LogCtx m = LogCtx- { logCtxDest :: LogDest+ { logCtxDest :: LogDest m , logCtxSeverity :: Severity } | NoLogs -- | Representation of the logs' destination-data LogDest- = LogFile FilePath+data LogDest m+ = LogWith (MonadIO m => Severity -> Text -> m ())+ | LogFile FilePath | LogStdout -- | Representation of the severity of the logs@@ -83,13 +84,17 @@ -- Logging with IO -------------------------------------------------------------------------------- -logToDest :: MonadIO m => LogCtx -> LogMsg -> m ()-logToDest LogCtx{..} logMsg =+logToDest :: MonadIO m => LogCtx m -> LogMsg -> m ()+logToDest LogCtx{..} logMsg = do+ let msgSeverity = severity logMsg case logCtxDest of- LogStdout -> if severity logMsg >= logCtxSeverity+ LogWith f -> if msgSeverity >= logCtxSeverity+ then f msgSeverity (logMsgToText logMsg)+ else pure ()+ LogStdout -> if msgSeverity >= logCtxSeverity then liftIO $ putText (logMsgToText logMsg) else pure ()- LogFile fp -> if severity logMsg >= logCtxSeverity+ LogFile fp -> if msgSeverity >= logCtxSeverity then liftIO $ appendFile fp (logMsgToText logMsg <> "\n") else pure () logToDest NoLogs _ = pure ()@@ -100,20 +105,20 @@ logToFile :: MonadIO m => FilePath -> Severity -> LogMsg -> m () logToFile fp s = logToDest $ LogCtx (LogFile fp) s -logWithSeverityIO :: forall m v. (RaftLogger v m, MonadIO m) => Severity -> LogCtx -> Text -> m ()+logWithSeverityIO :: forall m v. (RaftLogger v m, MonadIO m) => Severity -> LogCtx m -> Text -> m () logWithSeverityIO s logCtx msg = do logMsgData <- mkLogMsgData msg sysTime <- liftIO getSystemTime let logMsg = LogMsg (Just sysTime) s logMsgData logToDest logCtx logMsg -logInfoIO :: (RaftLogger v m, MonadIO m) => LogCtx -> Text -> m ()+logInfoIO :: (RaftLogger v m, MonadIO m) => LogCtx m -> Text -> m () logInfoIO = logWithSeverityIO Info -logDebugIO :: (RaftLogger v m, MonadIO m) => LogCtx -> Text -> m ()+logDebugIO :: (RaftLogger v m, MonadIO m) => LogCtx m -> Text -> m () logDebugIO = logWithSeverityIO Debug -logCriticalIO :: (RaftLogger v m, MonadIO m) => LogCtx -> Text -> m ()+logCriticalIO :: (RaftLogger v m, MonadIO m) => LogCtx m -> Text -> m () logCriticalIO = logWithSeverityIO Critical --------------------------------------------------------------------------------@@ -161,7 +166,7 @@ runRaftLoggerT $ logCritical msg panic ("logAndPanic: " <> msg) -logAndPanicIO :: (RaftLogger v m, MonadIO m) => LogCtx -> Text -> m a+logAndPanicIO :: (RaftLogger v m, MonadIO m) => LogCtx m -> Text -> m a logAndPanicIO logCtx msg = do logCriticalIO logCtx msg panic ("logAndPanicIO: " <> msg)
src/Raft/Monad.hs view
@@ -103,7 +103,7 @@ , resetElectionTimer :: m () , resetHeartbeatTimer :: m () , raftNodeConfig :: RaftNodeConfig- , raftNodeLogCtx :: LogCtx+ , raftNodeLogCtx :: LogCtx (RaftT v m) } newtype RaftT v m a = RaftT
test/QuickCheckStateMachine.hs view
@@ -11,7 +11,7 @@ import Control.Exception (bracket) import Control.Monad.IO.Class (liftIO) import Data.Bifunctor (bimap)-import Data.Char (isDigit)+import Data.Char (isDigit, toLower) import Data.List (isInfixOf, (\\)) import Data.Maybe (isJust, isNothing) import qualified Data.Set as Set@@ -156,6 +156,7 @@ command :: Handle -> ClientHandleRefs Concrete -> String -> IO (Int, Maybe String) command h chs@ClientHandleRefs{..} cmd = go 3 0 where+ -- unex is the number of unexpected responses go 0 unex = pure (unex, Nothing) go n unex = do eRes <- do@@ -165,17 +166,21 @@ case mresp of Nothing -> do hPutStrLn h "'getResponse' timed out"- pure (Just (unex, Nothing))+ pure (Just (unex + 1, Nothing)) Just resp ->- if "Timeout" `isInfixOf` resp+ if "timeout" `isInfixOf` (map toLower resp) then do hPutStrLn h ("Command timed out, retrying: " ++ show resp)- pure (Just (unex, Nothing))+ pure (Just (unex + 1, Nothing)) else if "Unexpected" `isInfixOf` resp then do hPutStrLn h ("Unexpected read/write response, retrying: " ++ show resp) pure (Just (unex + 1, Nothing))- else do+ else if "RaftClientSendError" `isInfixOf` resp+ then do+ hPutStrLn h ("Unexpected RaftClientSendError, retrying: " ++ show resp)+ pure (Just (unex + 1, Nothing))+ else do hPutStrLn h ("got response `" ++ resp ++ "'") pure (Just (unex, Just resp)) case eRes of