patterns 0.1.0 → 0.1.1

raw patch · 8 files changed

+218/−251 lines, 8 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• patterns.cabal +8/−1
• src/Network/Mom/Patterns.hs +1/−1
• src/Network/Mom/Patterns/Basic/Client.hs +15/−2
• src/Network/Mom/Patterns/Broker.hs +0/−6
• src/Network/Mom/Patterns/Broker/Broker.hs +36/−52
• src/Network/Mom/Patterns/Broker/Client.hs +15/−6
• src/Network/Mom/Patterns/Broker/Common.hs +0/−3
• src/broker/Registry.hs +143/−180

@@ -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.    . 

@@ -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 

@@ -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

@@ -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

@@ -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 ())- 

@@ -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

@@ -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

@@ -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