diff --git a/patterns.cabal b/patterns.cabal --- a/patterns.cabal +++ b/patterns.cabal @@ -1,5 +1,5 @@ Name: patterns -Version: 0.1.0 +Version: 0.1.1 Cabal-Version: >= 1.14.0 Copyright: Copyright (c) Tobias Schoofs, 2011 - 2013 License: LGPL @@ -29,6 +29,13 @@ . Release History: + + . + + [0.1.1] Client (Basic and MDP): + checkResult added; + MDP Broker: + Many brokers in one process. . diff --git a/src/Network/Mom/Patterns.hs b/src/Network/Mom/Patterns.hs --- a/src/Network/Mom/Patterns.hs +++ b/src/Network/Mom/Patterns.hs @@ -207,7 +207,7 @@ Advanced communication patterns use message exchange protocols and other means, - even brokers, to make communication reliably and scalable. + even brokers, to make communication reliable and scalable. The 'patterns' library aims to provide advanced topics as libraries that can be used diff --git a/src/Network/Mom/Patterns/Basic/Client.hs b/src/Network/Mom/Patterns/Basic/Client.hs --- a/src/Network/Mom/Patterns/Basic/Client.hs +++ b/src/Network/Mom/Patterns/Basic/Client.hs @@ -9,7 +9,7 @@ -- Client side of Client/Server ------------------------------------------------------------------------------- module Network.Mom.Patterns.Basic.Client ( - Client, clService, withClient, request) + Client, clService, withClient, request, checkResult) where import qualified System.ZMQ as Z @@ -85,8 +85,21 @@ -- > Nothing -> return Nothing -- > Just x -> return $ Just $ B.unpack x ------------------------------------------------------------------------ - request :: Client -> Timeout -> Source -> SinkR (Maybe a) -> IO (Maybe a) + request :: Client -> Timeout -> Source -> SinkR (Maybe r) -> IO (Maybe r) request c tmo src snk = do runSender (clSock c) src if tmo /= 0 then runReceiver (clSock c) tmo snk else return Nothing + + ------------------------------------------------------------------------ + -- | Check for a of a previously requested result; + -- use case: request with timout 0, do some work + -- and check for a result later. + -- Do not use this function without having requested the service + -- previously. + -- The parameters equal those of 'request', + -- but do not include a 'Source'. + ------------------------------------------------------------------------ + checkResult :: Client -> Timeout -> SinkR (Maybe r) -> IO (Maybe r) + checkResult c tmo snk | tmo == 0 = return Nothing + | otherwise = runReceiver (clSock c) tmo snk diff --git a/src/Network/Mom/Patterns/Broker.hs b/src/Network/Mom/Patterns/Broker.hs --- a/src/Network/Mom/Patterns/Broker.hs +++ b/src/Network/Mom/Patterns/Broker.hs @@ -87,12 +87,6 @@ The functionality provided by the broker module can be used to implement a standalone broker or an in-process broker linked with its clients and servers. - Note, however, - that brokers must not share the same process - with other brokers, since, internally, - the broker module uses a global registry, - which would be shared among all brokers - in the same process as well. -} {- $common diff --git a/src/Network/Mom/Patterns/Broker/Broker.hs b/src/Network/Mom/Patterns/Broker/Broker.hs --- a/src/Network/Mom/Patterns/Broker/Broker.hs +++ b/src/Network/Mom/Patterns/Broker/Broker.hs @@ -28,11 +28,9 @@ import Control.Monad.Trans (liftIO) import Control.Monad (when) import Prelude hiding (catch) - import Control.Exception (throwIO, finally) + import Control.Exception (throwIO) import Control.Concurrent.MVar - import System.IO.Unsafe - ------------------------------------------------------------------------ -- | Start a broker as a background process -- @@ -59,22 +57,24 @@ OnError_ -> (Controller -> IO r) -> IO r withBroker ctx srv tmo aClients aServers onerr ctrl | tmo <= 0 = throwIO $ MDPExc "Heartbeat is mandatory" - | otherwise = - lockBroker $ \_ -> do - R.clean - R.setHbPeriod tmo - t <- getCurrentTime - m <- newMVar t - withStreams ctx srv (1000 * fromIntegral tmo) + | otherwise = do + r <- R.newReg tmo + t <- getCurrentTime + m <- newMVar t + withStreams ctx srv (1000 * fromIntegral tmo) [Poll "servers" aServers RouterT Bind [] [], - Poll "clients" aClients RouterT Bind [] []] - (handleTmo m tmo) onerr - (handleStream m tmo) ctrl + Poll "clients" aClients RouterT Bind [] []] + (handleTmo r m tmo) onerr + (handleStream r m tmo) ctrl - -- handle heartbeat ------------------------------------------------------- - handleTmo :: MVar UTCTime -> Msec -> Streamer -> IO () - handleTmo m tmo s = hbPeriodReached m tmo >>= \x -> - when x $ R.checkWorker >>= mapM_ sndHb + ------------------------------------------------------------------------ + -- handle heartbeat: + -- - remove unresponsive workers + -- - heartbeat those that have been inactive for at least one hb period + ------------------------------------------------------------------------ + handleTmo :: R.Registry -> MVar UTCTime -> Msec -> Streamer -> IO () + handleTmo r m tmo s = hbPeriodReached m tmo >>= \x -> + when x $ R.checkWorker r >>= mapM_ sndHb where hbM i = [i, B.empty, mdpW01, xHeartBeat] sndHb i = C.runResourceT $ streamList (hbM i) $$ passAll s ["servers"] @@ -82,23 +82,23 @@ ------------------------------------------------------------------------ -- Handle incoming Streams ------------------------------------------------------------------------ - handleStream :: MVar UTCTime -> Msec -> StreamSink - handleStream m x s = - let action | getSource s == "clients" = recvClient s - | getSource s == "servers" = recvWorker s + handleStream :: R.Registry -> MVar UTCTime -> Msec -> StreamSink + handleStream r m x s = + let action | getSource s == "clients" = recvClient r s + | getSource s == "servers" = recvWorker r s | otherwise = return () -- and handle heartbeat afterwards-------------------------------- - in action >> liftIO (handleTmo m x s) + in action >> liftIO (handleTmo r m x s) ------------------------------------------------------------------------ -- Receive stream from client ------------------------------------------------------------------------ - recvClient :: StreamSink - recvClient s = mdpCRcvReq >>= uncurry go + recvClient :: R.Registry -> StreamSink + recvClient r s = mdpCRcvReq >>= uncurry go where go i sn | hdr sn == mmiHdr = handleMMI i sn | otherwise = handleReq i sn handleReq i sn = do - mbW <- liftIO $ R.getWorker sn + mbW <- liftIO $ R.getWorker r sn case mbW of Nothing -> noWorker sn Just w -> sendRequest w [i] s @@ -110,7 +110,7 @@ Nothing -> liftIO (throwIO $ MMIExc "No ServiceName in mmi.service request") Just x -> do - m <- bool2MMI <$> liftIO (R.lookupService x) + m <- bool2MMI <$> liftIO (R.lookupService r x) C.yield m =$ sendReply sn [i] s bool2MMI True = mmiFound bool2MMI False = mmiNotFound @@ -122,25 +122,25 @@ ------------------------------------------------------------------------ -- Receive stream from worker ------------------------------------------------------------------------ - recvWorker :: StreamSink - recvWorker s = do + recvWorker :: R.Registry -> StreamSink + recvWorker r s = do f <- mdpWRcvRep case f of - WBeat w -> liftIO (R.updWorkerHb w) -- update his heartbeat - WReady w sn -> liftIO $ R.insert w sn -- insert new worker - WReply w is -> handleReply w is s -- handle reply from worker - WDisc w -> liftIO $ R.remove w -- disconnect from worker + WBeat w -> liftIO (R.updWorkerHb r w) -- update his heartbeat + WReady w sn -> liftIO $ R.insert r w sn -- insert new worker + WReply w is -> handleReply r w is s -- handle reply from worker + WDisc w -> liftIO $ R.remove r w -- disconnect from worker _ -> liftIO (throwIO $ Ouch "Unexpected Frame from Worker!") ------------------------------------------------------------------------ -- Handle reply ------------------------------------------------------------------------ - handleReply :: Identity -> [Identity] -> StreamSink - handleReply w is s = do - mbS <- liftIO $ R.getServiceName w + handleReply :: R.Registry -> Identity -> [Identity] -> StreamSink + handleReply r w is s = do + mbS <- liftIO $ R.getServiceName r w case mbS of Nothing -> liftIO (throwIO $ ServerExc "Unknown Worker") - Just sn -> sendReply sn is s >> liftIO (R.freeWorker w) + Just sn -> sendReply sn is s >> liftIO (R.freeWorker r w) ------------------------------------------------------------------------ -- Send request to worker @@ -153,20 +153,4 @@ ------------------------------------------------------------------------ sendReply :: B.ByteString -> [Identity] -> StreamSink sendReply sn is s = mdpCSndRep sn is =$ passAll s ["clients"] - - ------------------------------------------------------------------------ - -- Mechanism to exclude two brokers - -- from running in the same process - ------------------------------------------------------------------------ - {-# NOINLINE _brk #-} - _brk :: MVar () - _brk = unsafePerformIO $ newMVar () - - lockBroker :: (() -> IO b) -> IO b - lockBroker act = do - mb_ <- tryTakeMVar _brk -- withMVar _brk - case mb_ of - Nothing -> throwIO $ SingleBrokerExc "Another broker is running!" - Just _ -> finally (act ()) (putMVar _brk ()) - diff --git a/src/Network/Mom/Patterns/Broker/Client.hs b/src/Network/Mom/Patterns/Broker/Client.hs --- a/src/Network/Mom/Patterns/Broker/Client.hs +++ b/src/Network/Mom/Patterns/Broker/Client.hs @@ -10,7 +10,7 @@ ------------------------------------------------------------------------------- module Network.Mom.Patterns.Broker.Client ( Client, withClient, checkService, - request) + request, checkResult) where import qualified Data.ByteString.Char8 as B @@ -117,13 +117,22 @@ request :: Client -> Timeout -> Source -> SinkR (Maybe a) -> IO (Maybe a) request c tmo src snk = do runSender (clSock c) $ mdpSrc (clService c) src - runReceiver (clSock c) tmo $ mdpSnk (clService c) snk + if tmo /= 0 then runReceiver (clSock c) tmo $ mdpSnk (clService c) snk + else return Nothing - {- currently not provided - checkReceive :: Client -> Timeout -> SinkR (Maybe a) -> IO (Maybe a) - checkReceive c tmo snk = + ------------------------------------------------------------------------ + -- | Check for a of a previously requested result; + -- use case: request with timout 0, do some work + -- and check for a result later. + -- Do not use this function without having requested the service + -- previously. + -- The parameters equal those of 'request', + -- but do not include a 'Source'. + ------------------------------------------------------------------------ + checkResult :: Client -> Timeout -> SinkR (Maybe a) -> IO (Maybe a) + checkResult c tmo snk | tmo == 0 = return Nothing + | otherwise = runReceiver (clSock c) tmo $ mdpSnk (clService c) snk - -} ------------------------------------------------------------------------ -- Add MDP headers to a stream created by source diff --git a/src/Network/Mom/Patterns/Broker/Common.hs b/src/Network/Mom/Patterns/Broker/Common.hs --- a/src/Network/Mom/Patterns/Broker/Common.hs +++ b/src/Network/Mom/Patterns/Broker/Common.hs @@ -289,9 +289,6 @@ | MDPExc String -- | MMI Protocol | MMIExc String - -- | SingleBroker error - -- (another broker is already running in the same process) - | SingleBrokerExc String -- move to Broker.Common deriving (Show, Read, Typeable, Eq) instance Exception MDPException diff --git a/src/broker/Registry.hs b/src/broker/Registry.hs --- a/src/broker/Registry.hs +++ b/src/broker/Registry.hs @@ -11,7 +11,8 @@ ------------------------------------------------------------------------------- module Registry #ifndef TEST - (insert, remove, + (Registry, newReg, + insert, remove, checkWorker, getWorker, freeWorker, getServiceName, updWorkerHb, lookupService, @@ -25,11 +26,8 @@ import Network.Mom.Patterns.Types (Msec, Identity) import Control.Concurrent - import Control.Monad (when) import Control.Applicative ((<$>)) - import System.IO.Unsafe - import qualified Data.ByteString.Char8 as B import Data.Map (Map) import qualified Data.Map as M -- filter, keys @@ -37,7 +35,34 @@ import Data.Sequence (Seq, (|>), (<|), (><), ViewL(..), ViewR(..)) import Data.Foldable (toList) import Data.Time.Clock + + ------------------------------------------------------------------------ + -- Registry + ------------------------------------------------------------------------ + type Registry = MVar Reg + data Reg = NewReg { + rgHb :: Msec, + rgSrv :: SrvTree, + rgWrk :: WrkTree, + rgS :: Seq B.ByteString + } + newReg :: Msec -> IO Registry + newReg ms = newMVar NewReg { + rgHb = ms, + rgSrv = M.empty, + rgWrk = M.empty, + rgS = S.empty} + + modifyReg :: Registry -> (Reg -> IO (Reg, r)) -> IO r + modifyReg = modifyMVar + + modifyReg_ :: Registry -> (Reg -> IO Reg) -> IO () + modifyReg_ = modifyMVar_ + + withReg :: Registry -> (Reg -> IO r) -> IO r + withReg = withMVar + ------------------------------------------------------------------------ -- Worker may be free or busy ------------------------------------------------------------------------ @@ -49,25 +74,16 @@ ------------------------------------------------------------------------ data Worker = Worker { wrkId :: Identity, - wrkState :: State, wrkHB :: Heartbeat, wrkQ :: MVar Queue } - ------------------------------------------------------------------------ - -- Base for Heartbeat calculation: - -- expected heartbeat is _hb * tolerance, - -- heartbeat to be sent is _hb - ------------------------------------------------------------------------ - {-# NOINLINE _hb #-} - _hb :: MVar Msec - _hb = unsafePerformIO $ newMVar 2000 -- default: 2 seconds - ------------------------------------------------------------------------- -- Set common heartbeat period ------------------------------------------------------------------------- - setHbPeriod :: Msec -> IO () - setHbPeriod hb = modifyMVar_ _hb $ \_ -> return hb + setHbPeriod :: Registry -> Msec -> IO () + setHbPeriod r hb = modifyReg_ r $ \reg -> + return reg{rgHb = hb} ------------------------------------------------------------------------- -- Update worker's heartbeat descriptor according to function @@ -79,7 +95,7 @@ ------------------------------------------------------------------------ -- A Worker Node is an Identity paired with a worker (lookup) -- A Worker Tree is a map of identities and services, - -- where service is a queue of free and busy workers; + -- where a service is basically a queue of free and busy workers; -- the worker tree, hence, does not point directly to the worker, -- but to the queue where the worker can be found: -- @@ -111,119 +127,95 @@ type SrvTree = Map B.ByteString Service ------------------------------------------------------------------------- - -- service tree - ------------------------------------------------------------------------- - {-# NOINLINE _srv #-} - _srv :: MVar SrvTree - _srv = unsafePerformIO $ newMVar M.empty - - ------------------------------------------------------------------------- - -- A sequence of services, used for removing unresponsive workers: - -- We clean up one service at a time, so there are not too long - -- work intervals due to clean-up. - ------------------------------------------------------------------------- - {-# NOINLINE _s #-} - _s :: MVar (Seq B.ByteString) - _s = unsafePerformIO $ newMVar S.empty - - ------------------------------------------------------------------------- - -- worker tree - ------------------------------------------------------------------------- - {-# NOINLINE _wrk #-} - _wrk :: MVar WrkTree - _wrk = unsafePerformIO $ newMVar M.empty - - ------------------------------------------------------------------------- -- Register new worker for service ------------------------------------------------------------------------- - insert :: Identity -> B.ByteString -> IO () - insert i s = do - mbW <- lookupW i - case mbW of - Just _ -> return () -- identity already known + insert :: Registry -> Identity -> B.ByteString -> IO () + insert r i s = modifyReg_ r $ \reg -> + case M.lookup i (rgWrk reg) of + Just _ -> return reg -- identity already known Nothing -> do - sn <- lookupS s >>= getSn - initW i sn - insertW i sn - where getSn mbS = - case mbS of - Just sn -> return sn - Nothing -> do - q <- newMVar $ Q S.empty S.empty - let sn = Service { - srvName = s, - srvQ = q} - insertS s sn - return sn + (reg2, sn) <- case M.lookup s (rgSrv reg) of + Just x -> return (reg, x) -- service exists + Nothing -> do -- new service + q <- newMVar $ Q S.empty S.empty + let x = Service { + srvName = s, + srvQ = q} + return (reg {rgSrv = M.insert s x (rgSrv reg), + rgS = rgS reg |> s}, x) + hb <- newHeartbeat (rgHb reg2) + initW i hb sn -- add worker to service queue + return reg2 {rgWrk = -- insert worker into wrk tree + M.insert i sn $ rgWrk reg2} ------------------------------------------------------------------------- -- Remove worker ------------------------------------------------------------------------- - remove :: Identity -> IO () - remove i = do - mbW <- lookupW i - case mbW of - Nothing -> return () + remove :: Registry -> Identity -> IO () + remove r i = modifyReg_ r $ \reg -> + case M.lookup i (rgWrk reg) of + Nothing -> return reg -- worker does not exist Just sn -> do - deleteW i - e <- modifyMVar (srvQ sn) $ \q -> + let reg2 = reg {rgWrk = M.delete i (rgWrk reg)} -- delete from tree + e <- modifyMVar (srvQ sn) $ \q -> -- remove from queue let q' = removeQ i q e = emptyQ q' in return (q', e) - when e $ deleteS (srvName sn) + if e then return reg2{rgSrv = -- remove empty service + M.delete (srvName sn) (rgSrv reg2)} + else return reg2 ------------------------------------------------------------------------- -- Free a worker that has been busy -- updating his heatbeat ------------------------------------------------------------------------- - freeWorker :: Identity -> IO () - freeWorker i = getCurrentTime >>= \now -> - freeWorkerWithUpd i (updHb updHim now) + freeWorker :: Registry -> Identity -> IO () + freeWorker r i = getCurrentTime >>= \now -> + freeWorkerWithUpd r i (updHb updHim now) ------------------------------------------------------------------------- -- Free a worker that has been busy -- updating heartbeat (because we apparently have feedback ) ------------------------------------------------------------------------- - freeWorkerWithUpd :: Identity -> (WrkNode -> WrkNode) -> IO () - freeWorkerWithUpd i upd = do - mbW <- lookupW i - case mbW of - Nothing -> return () -- silent error - Just sn -> modifyMVar_ (srvQ sn) $ \q -> return $ setStateQ i Free upd q + freeWorkerWithUpd :: Registry -> Identity -> (WrkNode -> WrkNode) -> IO () + freeWorkerWithUpd r i upd = withReg r $ \reg -> + case M.lookup i (rgWrk reg) of + Nothing -> return () -- worker does not exist + Just sn -> modifyMVar_ (srvQ sn) $ \q -> + return $ setStateQ i Free upd q -- set free and upd hb ------------------------------------------------------------------------- -- Update heartbeat ------------------------------------------------------------------------- - updWorkerHb :: Identity -> IO () - updWorkerHb i = getCurrentTime >>= \now -> updWorker i (updHb updHim now) + updWorkerHb :: Registry -> Identity -> IO () + updWorkerHb r i = getCurrentTime >>= \now -> + updWorker r i (updHb updHim now) ------------------------------------------------------------------------- -- Generic worker update ------------------------------------------------------------------------- - updWorker :: Identity -> (WrkNode -> WrkNode) -> IO () - updWorker i f = do - mbW <- lookupW i - case mbW of - Nothing -> return () + updWorker :: Registry -> Identity -> (WrkNode -> WrkNode) -> IO () + updWorker r i f = withReg r $ \reg -> + case M.lookup i (rgWrk reg) of + Nothing -> return () -- worker does not exist Just sn -> modifyMVar_ (srvQ sn) $ \q -> - return $ updateQ i f q + return $ updateQ i f q -- update worker ------------------------------------------------------------------------- -- Get service by worker identity ------------------------------------------------------------------------- - getServiceName :: Identity -> IO (Maybe B.ByteString) - getServiceName i = do - mbW <- lookupW i - case mbW of + getServiceName :: Registry -> Identity -> IO (Maybe B.ByteString) + getServiceName r i = withReg r $ \reg -> + case M.lookup i (rgWrk reg) of Nothing -> return Nothing Just s -> return $ Just (srvName s) ------------------------------------------------------------------------- -- Check whether service exists ------------------------------------------------------------------------- - lookupService :: B.ByteString -> IO Bool - lookupService s = do - mbS <- lookupS s + lookupService :: Registry -> B.ByteString -> IO Bool + lookupService r s = do + mbS <- lookupS r s case mbS of Nothing -> return False Just _ -> return True @@ -235,16 +227,16 @@ -- and update its heartbeat -- - if free q is emtpy, take the first of the busy queue ------------------------------------------------------------------------- - getWorker :: B.ByteString -> IO (Maybe Identity) - getWorker s = do + getWorker :: Registry -> B.ByteString -> IO (Maybe Identity) + getWorker r s = do now <- getCurrentTime - mbS <- lookupS s + mbS <- lookupS r s case mbS of - Nothing -> return Nothing + Nothing -> return Nothing -- service does not exist Just sn -> do - bury now sn -- remove non-responsive workers + bury r now sn -- remove non-responsive workers modifyMVar (srvQ sn) $ \q -> - case firstFreeQ q of + case firstFreeQ q of -- get first of free Just (i,_) -> return (setStateQ i Busy (updHb updMe now) q, Just i) Nothing -> @@ -258,20 +250,20 @@ -- and heartbeat those that have been inactive -- for at least one heartbeat period ------------------------------------------------------------------------- - checkWorker :: IO [Identity] - checkWorker = do + checkWorker :: Registry -> IO [Identity] + checkWorker r = do now <- getCurrentTime - checkService >>= \mbS -> case mbS of - Nothing -> return () - Just s -> bury now s - concat <$> withMVar _srv (mapM (getWorkers now) . M.elems) + checkService r >>= \mbS -> case mbS of + Nothing -> return () + Just s -> bury r now s + srv <- M.elems <$> withReg r (return . rgSrv) + concat <$> mapM (getWorkers now) srv where getWorkers now s = do is <- withMVar (srvQ s) ( return . map fst . takeIfQ Free tst) - mapM_ (`updWorker` updSt) is + mapM_ (\i -> updWorker r i updSt) is return is - where updSt (i,w) = let hb = updMe now $ wrkHB w in (i, w{wrkHB = hb}) tst = checkMe now . wrkHB . snd @@ -284,59 +276,62 @@ -- if a lot of services have been removed since our last visit -- we will go through this many times ------------------------------------------------------------------------- - checkService :: IO (Maybe Service) - checkService = do - mbS <- modifyMVar _s getS + checkService :: Registry -> IO (Maybe Service) + checkService r = do + mbS <- modifyReg r $ \reg -> do + let (s, mbS) = getS (rgS reg) + return (reg {rgS = s}, mbS) case mbS of Nothing -> return Nothing Just s -> do - mbSrv <- lookupS s + mbSrv <- lookupS r s case mbSrv of - Nothing -> rm >> checkService + Nothing -> rm >> checkService r Just srv -> return $ Just srv where getS s = case S.viewl s of - EmptyL -> return (s, Nothing) - x :< xs -> - return (xs |> x, Just x) + EmptyL -> (s, Nothing) + x :< xs -> (xs |> x, Just x) rmTail s = case S.viewr s of - EmptyR -> return s - xs :> _ -> return xs - rm = modifyMVar_ _s rmTail + EmptyR -> s + xs :> _ -> xs + rm = modifyReg_ r $ \reg -> + return reg {rgS = rmTail (rgS reg)} ------------------------------------------------------------------------- -- Remove unresponsive workers ------------------------------------------------------------------------- - bury :: UTCTime -> Service -> IO () - bury now sn = do - q <- readMVar $ srvQ sn - mapM_ (remove . fst) $ findDeads $ qFree q - mapM_ (remove . fst) $ findDeads $ qBusy q - where findDeads = filter (not . alive now . wrkHB . snd) . toList + bury :: Registry -> UTCTime -> Service -> IO () + bury r now sn = do + q <- readMVar $ srvQ sn -- we use remove on registry, so don't block + mapM_ (remove r . fst) $ findDead $ qFree q + mapM_ (remove r . fst) $ findDead $ qBusy q + where findDead = filter (not . alive now . wrkHB . snd) . toList ------------------------------------------------------------------------- - -- Clean up global variables + -- Clean up registry ------------------------------------------------------------------------- - clean :: IO () - clean = do modifyMVar_ _wrk $ \_ -> return M.empty - modifyMVar_ _srv $ \_ -> return M.empty - modifyMVar_ _s $ \_ -> return S.empty + clean :: Registry -> IO () + clean r = modifyReg_ r $ \reg -> return reg { + rgWrk = M.empty, + rgSrv = M.empty, + rgS = S.empty} ------------------------------------------------------------------------- -- Number of workers ------------------------------------------------------------------------- - size :: IO Int - size = withMVar _wrk $ \t -> return $ M.size t + size :: Registry -> IO Int + size r = withReg r $ \reg -> return $ M.size (rgWrk reg) ------------------------------------------------------------------------- -- Per service: -- number of free workers -- number of busy workers ------------------------------------------------------------------------- - statPerService :: B.ByteString -> IO (B.ByteString, Int, Int) - statPerService s = do - mbS <- lookupS s + statPerService :: Registry -> B.ByteString -> IO (B.ByteString, Int, Int) + statPerService r s = do + mbS <- lookupS r s case mbS of Nothing -> return (s, 0, 0) Just sn -> perService (s, sn) @@ -346,8 +341,9 @@ -- number of free workers -- number of busy workers ------------------------------------------------------------------------- - stat :: IO [(B.ByteString, Int, Int)] - stat = withMVar _srv $ \t -> mapM perService $ M.assocs t + stat :: Registry -> IO [(B.ByteString, Int, Int)] + stat r = withReg r $ \reg -> + mapM perService $ M.assocs (rgSrv reg) ------------------------------------------------------------------------- -- Get stats per service @@ -360,9 +356,9 @@ ------------------------------------------------------------------------- -- Debug: print q ------------------------------------------------------------------------- - printQ :: B.ByteString -> IO () - printQ s = do - mbS <- lookupS s + printQ :: Registry -> B.ByteString -> IO () + printQ r s = do + mbS <- lookupS r s case mbS of Nothing -> return () Just x -> do q <- readMVar (srvQ x) @@ -377,53 +373,20 @@ -- initialising heartbeat -- inserting into service queue ------------------------------------------------------------------------- - initW :: Identity -> Service -> IO () - initW i sn = modifyMVar_ (srvQ sn) $ \q -> do - hb <- withMVar _hb newHeartbeat - let w = Worker { - wrkId = i, - wrkState = Free, - wrkHB = hb, - wrkQ = srvQ sn} - return $ insertQ (i,w) q + initW :: Identity -> Heartbeat -> Service -> IO () + initW i hb sn = modifyMVar_ (srvQ sn) $ \q -> do + let w = Worker { + wrkId = i, + wrkHB = hb, + wrkQ = srvQ sn} + return $ insertQ (i,w) q ------------------------------------------------------------------------- - -- Lookup worker, returning service - ------------------------------------------------------------------------- - lookupW :: Identity -> IO (Maybe Service) - lookupW i = withMVar _wrk $ \t -> return $ M.lookup i t - - ------------------------------------------------------------------------- -- Lookup service ------------------------------------------------------------------------- - lookupS :: B.ByteString -> IO (Maybe Service) - lookupS sn = withMVar _srv $ \t -> return $ M.lookup sn t - - ------------------------------------------------------------------------- - -- Insert worker - ------------------------------------------------------------------------- - insertW :: Identity -> Service -> IO () - insertW i sn = modifyMVar_ _wrk $ \t -> return $ M.insert i sn t - - ------------------------------------------------------------------------- - -- Insert service - ------------------------------------------------------------------------- - insertS :: B.ByteString -> Service -> IO () - insertS s sn = do - modifyMVar_ _srv $ \t -> return $ M.insert s sn t - modifyMVar_ _s $ \ss -> return $ ss |> s - - ------------------------------------------------------------------------- - -- Delete worker - ------------------------------------------------------------------------- - deleteW :: Identity -> IO () - deleteW i = modifyMVar_ _wrk $ \t -> return $ M.delete i t - - ------------------------------------------------------------------------- - -- Delete service - ------------------------------------------------------------------------- - deleteS :: B.ByteString -> IO () - deleteS s = modifyMVar_ _srv $ \t -> return $ M.delete s t + lookupS :: Registry -> B.ByteString -> IO (Maybe Service) + lookupS r sn = withMVar r $ \reg -> + return $ M.lookup sn (rgSrv reg) ------------------------------------------------------------------------- -- Queue: @@ -548,7 +511,7 @@ (w :< _) -> Just w ------------------------------------------------------------------------- - -- Take n that fulfil f with state s + -- Take those that fulfil f with state s ------------------------------------------------------------------------- takeIfQ :: State -> (WrkNode -> Bool) -> Queue -> [WrkNode] takeIfQ s f q = case s of