packages feed

nqe (empty) → 0.1.0.0

raw patch · 9 files changed

+577/−0 lines, 9 filesdep +asyncdep +basedep +bytestringsetup-changed

Dependencies added: async, base, bytestring, conduit, conduit-extra, containers, exceptions, hspec, lifted-async, lifted-base, monad-control, nqe, stm, stm-conduit, text, transformers-base

Files

+ README.md view
@@ -0,0 +1,3 @@+# Not Quite Erlang++Haskell framework for simple concurrent processes with mailboxes inspired by Erlang/OTP.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ UNLICENSE view
@@ -0,0 +1,24 @@+This is free and unencumbered software released into the public domain.++Anyone is free to copy, modify, publish, use, compile, sell, or+distribute this software, either in source code form or as a compiled+binary, for any purpose, commercial or non-commercial, and by any+means.++In jurisdictions that recognize copyright laws, the author or authors+of this software dedicate any and all copyright interest in the+software to the public domain. We make this dedication for the benefit+of the public at large and to the detriment of our heirs and+successors. We intend this dedication to be an overt act of+relinquishment in perpetuity of all present and future rights to this+software under copyright law.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR+OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,+ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR+OTHER DEALINGS IN THE SOFTWARE.++For more information, please refer to <http://unlicense.org/>
+ nqe.cabal view
@@ -0,0 +1,57 @@+name:                nqe+version:             0.1.0.0+synopsis:            Concurrency library in the style of Erlang/OTP+description:+    Follow the example of Erlang developing a minimalistic actor library with a+    distinct Haskell flavour. It does not implement Erlang/OTP-style processes+    exactly, but it is merely inspired by their philosophy.+homepage:            https://github.com/xenog/nqe#readme+license:             PublicDomain+license-file:        UNLICENSE+author:              Jean-Pierre Rupp+maintainer:          xenog@protonmail.com+category:            Control+build-type:          Simple+extra-source-files:  README.md+cabal-version:       >=1.10++library+  hs-source-dirs:      src+  exposed-modules:     Control.Concurrent.NQE+  other-modules:       Control.Concurrent.NQE.Process+                       Control.Concurrent.NQE.Network+                       Control.Concurrent.NQE.Supervisor+  build-depends:       base >= 4.7 && < 5+                     , bytestring >= 0.10 && < 0.11+                     , conduit >= 1.2 && < 1.3+                     , conduit-extra >= 1.1 && < 1.3+                     , containers >= 0.5 && < 0.6+                     , async >= 2.1 && < 2.3+                     , lifted-async >= 0.9 && < 0.10+                     , lifted-base >= 0.2 && < 0.3+                     , transformers-base >= 0.4 && < 0.5+                     , monad-control >= 1.0 && < 1.1+                     , stm >= 2.4 && < 3+  default-language:    Haskell2010++test-suite nqe-test+  type:                exitcode-stdio-1.0+  hs-source-dirs:      test+  main-is:             Spec.hs+  build-depends:       base+                     , nqe+                     , bytestring >= 0.10 && < 0.11+                     , conduit >= 1.2 && < 1.3+                     , conduit-extra >= 1.1 && < 1.3+                     , exceptions >= 0.8 && < 0.9+                     , async >= 2.1 && < 2.3+                     , hspec >= 2.4 && < 3+                     , stm >= 2.4 && < 3+                     , stm-conduit >= 3.0 && < 3.1+                     , text >= 1.2 && < 1.3+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  default-language:    Haskell2010++source-repository head+  type:     git+  location: https://github.com/xenog/nqe
+ src/Control/Concurrent/NQE.hs view
@@ -0,0 +1,13 @@+module Control.Concurrent.NQE+    ( module Control.Concurrent.NQE.Process+    , module Control.Concurrent.NQE.Network+    , module Control.Concurrent.NQE.Supervisor+    , module Control.Concurrent.STM+    , module Control.Concurrent.Async.Lifted.Safe+    ) where++import           Control.Concurrent.Async.Lifted.Safe+import           Control.Concurrent.NQE.Network+import           Control.Concurrent.NQE.Process+import           Control.Concurrent.NQE.Supervisor+import           Control.Concurrent.STM
+ src/Control/Concurrent/NQE/Network.hs view
@@ -0,0 +1,28 @@+{-# LANGUAGE FlexibleContexts      #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes            #-}+module Control.Concurrent.NQE.Network+    ( fromSource+    , withSource+    ) where++import           Control.Concurrent.Async.Lifted.Safe+import           Control.Concurrent.NQE.Process+import           Control.Monad.IO.Class+import           Control.Monad.Trans.Control+import           Data.Conduit++fromSource ::+       (MonadIO m, Mailbox mbox)+    => Source m msg+    -> mbox msg -- ^ will receive all messages+    -> m ()+fromSource src mbox = src $$ awaitForever (`send` mbox)++withSource ::+       (MonadIO m, MonadBaseControl IO m, Forall (Pure m), Mailbox mbox)+    => Source m msg+    -> mbox msg+    -> (Async () -> m a)+    -> m a+withSource src mbox = withAsync (fromSource src mbox)
+ src/Control/Concurrent/NQE/Process.hs view
@@ -0,0 +1,114 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleContexts          #-}+{-# LANGUAGE LambdaCase                #-}+{-# LANGUAGE MultiParamTypeClasses     #-}+{-# LANGUAGE RankNTypes                #-}+module Control.Concurrent.NQE.Process where++import           Control.Concurrent+import           Control.Concurrent.Async.Lifted.Safe+import           Control.Concurrent.STM+import           Control.Monad+import           Control.Monad.IO.Class+import           Control.Monad.Trans.Control++type Reply a = a -> STM ()+type Listen a = a -> STM ()++class Mailbox mbox where+    mailboxEmptySTM :: mbox msg -> STM Bool+    sendSTM :: msg -> mbox msg -> STM ()+    receiveSTM :: mbox msg -> STM msg+    requeueMsg :: msg -> mbox msg -> STM ()++instance Mailbox TQueue where+    mailboxEmptySTM = isEmptyTQueue+    sendSTM msg = (`writeTQueue` msg)+    receiveSTM = readTQueue+    requeueMsg msg = (`unGetTQueue` msg)++instance Mailbox TBQueue where+    mailboxEmptySTM = isEmptyTBQueue+    sendSTM msg = (`writeTBQueue` msg)+    receiveSTM = readTBQueue+    requeueMsg msg = (`unGetTBQueue` msg)++data Inbox msg =+    forall mbox. (Mailbox mbox) =>+                 Inbox (mbox msg)++instance Mailbox Inbox where+    mailboxEmptySTM (Inbox mbox) = mailboxEmptySTM mbox+    sendSTM msg (Inbox mbox) = msg `sendSTM` mbox+    receiveSTM (Inbox mbox) = receiveSTM mbox+    requeueMsg msg (Inbox mbox) = msg `requeueMsg` mbox++mailboxEmpty :: (MonadIO m, Mailbox mbox) => mbox msg -> m Bool+mailboxEmpty = liftIO . atomically . mailboxEmptySTM++send :: (MonadIO m, Mailbox mbox) => msg -> mbox msg -> m ()+send msg = liftIO . atomically . sendSTM msg++requeue :: (Mailbox mbox) => [msg] -> mbox msg -> STM ()+requeue xs mbox = mapM_ (`requeueMsg` mbox) xs++extractMsg ::+       (Mailbox mbox)+    => [(msg -> Maybe a, a -> b)]+    -> mbox msg+    -> STM b+extractMsg hs mbox = do+    msg <- receiveSTM mbox+    go [] msg hs+  where+    go acc msg [] = do+        msg' <- receiveSTM mbox+        go (msg : acc) msg' hs+    go acc msg ((f, action):fs) =+        case f msg of+            Just x -> do+                requeue acc mbox+                return $ action x+            Nothing -> go acc msg fs++query ::+       (MonadIO m, Mailbox mbox)+    => (Reply b -> msg)+    -> mbox msg+    -> m b+query f mbox = do+    box <- liftIO $ atomically newEmptyTMVar+    f (putTMVar box) `send` mbox+    liftIO . atomically $ takeTMVar box++dispatch ::+       (MonadIO m, Mailbox mbox)+    => [(msg -> Maybe a, a -> IO b)] -- ^ action to dispatch+    -> mbox msg -- ^ mailbox to read from+    -> m b+dispatch hs = liftIO . join . atomically . extractMsg hs++dispatchSTM :: (Mailbox mbox) => [msg -> Maybe a] -> mbox msg -> STM a+dispatchSTM = extractMsg . map (\x -> (x, id))++receive ::+       (MonadIO m, Mailbox mbox)+    => mbox msg+    -> m msg+receive = dispatch [(Just, return)]++receiveMatch :: (MonadIO m, Mailbox mbox) => mbox msg -> (msg -> Maybe a) -> m a+receiveMatch mbox f = dispatch [(f, return)] mbox++receiveMatchSTM :: (Mailbox mbox) => mbox msg -> (msg -> Maybe a) -> STM a+receiveMatchSTM mbox f = dispatchSTM [f] mbox++timeout ::+       forall m b. (MonadIO m, MonadBaseControl IO m, Forall (Pure m))+    => Int+    -> m b+    -> m (Maybe b)+timeout n action =+    race (liftIO $ threadDelay n) action >>= \case+        Left () -> return Nothing+        Right r -> return $ Just r
+ src/Control/Concurrent/NQE/Supervisor.hs view
@@ -0,0 +1,168 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleContexts          #-}+{-# LANGUAGE GADTs                     #-}+{-# LANGUAGE MultiParamTypeClasses     #-}+module Control.Concurrent.NQE.Supervisor+    ( SupervisorMessage(..)+    , Strategy(..)+    , supervisor+    , addChild+    , removeChild+    , stopSupervisor+    ) where++import           Control.Applicative+import           Control.Concurrent.Async.Lifted.Safe+import           Control.Concurrent.NQE.Process+import           Control.Concurrent.STM+import           Control.Exception.Lifted+import           Control.Monad+import           Control.Monad.Base+import           Control.Monad.IO.Class+import           Control.Monad.Trans.Control++data SupervisorMessage+    = AddChild (IO ())+               (Reply (Async ()))+    | RemoveChild (Async ())+    | StopSupervisor++data Strategy+    = Notify ((Async (), Either SomeException ()) -> STM ())+    | KillAll+    | IgnoreGraceful+    | IgnoreAll++supervisor ::+       (MonadIO m, MonadBaseControl IO m, Forall (Pure m), Mailbox mbox)+    => Strategy+    -> mbox SupervisorMessage+    -> [m ()]+    -> m ()+supervisor strat mbox children = do+    state <- liftIO $ newTVarIO []+    finally (go state) (down state)+  where+    go state = do+        mapM_ (startChild state) children+        loop state+    loop state = do+        e <-+            liftIO . atomically $+            Right <$> receiveSTM mbox <|> Left <$> waitForChild state+        again <- case e of+            Right m -> processMessage state m+            Left x  -> processDead state strat x+        when again $ loop state+    down state = do+        as <- liftIO . atomically $ readTVar state+        mapM_ cancel as++waitForChild :: TVar [Async ()] -> STM (Async (), Either SomeException ())+waitForChild state = do+    as <- readTVar state+    waitAnyCatchSTM as++processMessage ::+       (MonadIO m, MonadBaseControl IO m, Forall (Pure m))+    => TVar [Async ()]+    -> SupervisorMessage+    -> m Bool+processMessage state (AddChild ch r) = do+    a <- async $ liftIO ch+    liftIO . atomically $ do+        modifyTVar' state (a:)+        r a+    return True++processMessage state (RemoveChild a) = do+    liftIO . atomically $ modifyTVar' state (filter (/= a))+    cancel a+    return True++processMessage state StopSupervisor = do+    as <- liftIO $ readTVarIO state+    forM_ as (stopChild state)+    return False++processDead ::+       (MonadIO m, MonadBaseControl IO m)+    => TVar [Async ()]+    -> Strategy+    -> (Async (), Either SomeException ())+    -> m Bool+processDead state IgnoreAll (a, _) = do+    liftIO . atomically $ modifyTVar' state (filter (/= a))+    return True++processDead state KillAll (a, e) = do+    as <- liftIO . atomically $ do+        modifyTVar' state (filter (/= a))+        readTVar state+    mapM_ (stopChild state) as+    case e of+        Left x   -> throw x+        Right () -> return False++processDead state IgnoreGraceful (a, Right ()) = do+    liftIO . atomically $ modifyTVar' state (filter (/= a))+    return True++processDead state IgnoreGraceful (a, Left e) = do+    as <- liftIO . atomically $ do+        modifyTVar' state (filter (/= a))+        readTVar state+    mapM_ (stopChild state) as+    throw e++processDead state (Notify notif) (a, e) = do+    x <-+        liftIO . atomically $ do+            modifyTVar' state (filter (/= a))+            catchSTM (notif (a, e) >> return Nothing) $ \x ->+                return $ Just (x :: SomeException)+    case x of+        Nothing -> return True+        Just ex -> do+            as <- liftIO $ readTVarIO state+            forM_ as (stopChild state)+            throwIO ex++startChild ::+       (MonadIO m, MonadBaseControl IO m, Forall (Pure m))+    => TVar [Async ()]+    -> m ()+    -> m (Async ())+startChild state run = do+    a <- async run+    liftIO . atomically $ modifyTVar' state (a:)+    return a++stopChild ::+       (MonadIO m, MonadBase IO m) => TVar [Async ()] -> Async () -> m ()+stopChild state a = do+    isChild <-+        liftIO . atomically $ do+            cur <- readTVar state+            let new = filter (/= a) cur+            writeTVar state new+            return $ cur /= new+    when isChild $ cancel a++addChild ::+       (MonadIO m, Mailbox mbox)+    => mbox SupervisorMessage+    -> IO ()+    -> m (Async ())+addChild mbox action = AddChild action `query` mbox++removeChild ::+       (MonadIO m, Mailbox mbox)+    => mbox SupervisorMessage+    -> Async ()+    -> m ()+removeChild mbox child = RemoveChild child `send` mbox++stopSupervisor ::+       (MonadIO m, Mailbox mbox) => mbox SupervisorMessage -> m ()+stopSupervisor mbox = StopSupervisor `send` mbox
+ test/Spec.hs view
@@ -0,0 +1,168 @@+{-# LANGUAGE LambdaCase          #-}+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE ScopedTypeVariables #-}+import           Control.Concurrent     hiding (yield)+import           Control.Concurrent.NQE+import           Control.Exception+import           Control.Monad+import           Control.Monad.Catch+import           Control.Monad.IO.Class+import           Data.ByteString        (ByteString)+import           Data.Conduit+import           Data.Conduit.Text      (decode, encode, utf8)+import qualified Data.Conduit.Text      as CT+import           Data.Conduit.TMChan+import           Data.Text              (Text)+import           Test.Hspec++data Pong = Pong deriving (Eq, Show)+newtype Ping = Ping (Pong -> STM ())++data TestError+    = TestError1+    | TestError2+    | TestError3+    deriving (Show, Eq)+instance Exception TestError++pong :: TQueue Ping -> IO ()+pong mbox =+    forever $ do+        Ping reply <- receive mbox+        atomically (reply Pong)++encoder :: MonadThrow m => Conduit Text m ByteString+encoder = encode utf8++decoder :: MonadThrow m => Conduit ByteString m Text+decoder = decode utf8 =$= CT.lines++conduits ::+       IO ( Source IO ByteString+          , Sink ByteString IO ()+          , Source IO ByteString+          , Sink ByteString IO ())+conduits = do+    inChan <- newTBMChanIO 2048+    outChan <- newTBMChanIO 2048+    return ( sourceTBMChan inChan+           , sinkTBMChan outChan True+           , sourceTBMChan outChan+           , sinkTBMChan inChan True+           )++pongServer ::+       Source IO ByteString+    -> Sink ByteString IO ()+    -> (Async () -> IO a)+    -> IO a+pongServer source sink go = do+    mbox <- newTQueueIO+    withAsync (action mbox) go+  where+    action mbox = withSource src mbox . const $ processor mbox $$ snk+    src = source =$= decoder+    snk = encoder =$= sink+    processor mbox =+        forever $+        liftIO (receive mbox) >>= \case+            ("ping" :: Text) -> yield ("pong\n" :: Text)+            _ -> return ()++pongClient :: Source IO ByteString+           -> Sink ByteString IO ()+           -> IO Text+pongClient source sink = do+    mbox <- newTQueueIO+    withAsync (action mbox) go+  where+    action mbox =+        withSource src mbox $ const $ processor mbox+    go = wait+    src = source =$= decoder+    snk = encoder =$= sink+    processor mbox = do+        yield ("ping\n" :: Text) $$ snk+        receive mbox++main :: IO ()+main =+    hspec $ do+        describe "two communicating processes" $ do+            it "exchange ping/pong messages" $ do+                mbox <- newTQueueIO+                g <- withAsync (pong mbox) $ const $ query Ping mbox+                g `shouldBe` Pong+        describe "network process" $ do+            it "responds to a ping" $ do+                (source1, sink1, source2, sink2) <- conduits+                msg <-+                    pongServer source1 sink1 $ const $ pongClient source2 sink2+                msg `shouldBe` "pong"+        describe "utilities" $ do+            it "timeout action fails" $ do+                n <- timeout 0xbeef (threadDelay 0xdeadbeef)+                n `shouldBe` Nothing+            it "timeout action succeeds" $ do+                n <- timeout 0xdeadbeef (return 0xbeef)+                n `shouldBe` Just 0xbeef+        describe "supervisor" $ do+            let p1 m = forever $ receive m >>= \r -> atomically $ r ()+                p2 = query id+            it "all processes end without failure" $ do+                mbox <- newTQueueIO+                sup <- newTQueueIO+                g <- async $ supervisor KillAll sup [p1 mbox, p2 mbox]+                wait g `shouldReturn` ()+            it "one process crashes" $ do+                mbox <- newTQueueIO+                sup <- newTQueueIO+                g <-+                    async $+                    supervisor+                        IgnoreGraceful+                        sup+                        [p1 mbox, p2 mbox >> throw TestError1]+                wait g `shouldThrow` (== TestError1)+            it "both processes crash" $ do+                sup <- newTQueueIO+                g <-+                    async $+                    supervisor+                        IgnoreGraceful+                        sup+                        [throw TestError1, throw TestError2]+                wait g `shouldThrow` (\e -> e == TestError1 || e == TestError2)+            it "process crashes ignored" $ do+                sup <- newTQueueIO+                g <-+                    async $+                    supervisor+                        IgnoreAll+                        sup+                        [throw TestError1, throw TestError2]+                stopSupervisor sup+                wait g `shouldReturn` ()+            it "monitors processes" $ do+                sup <- newTQueueIO+                mon <- newTQueueIO+                g <-+                    async $+                    supervisor+                        (Notify (writeTQueue mon))+                        sup+                        [throw TestError1, throw TestError2]+                (t1, t2) <-+                    atomically $ (,) <$> readTQueue mon <*> readTQueue mon+                let er e =+                        case e of+                            Right () -> False+                            Left x ->+                                case fromException x of+                                    Just TestError1 -> True+                                    Just TestError2 -> True+                                    _ -> False+                snd t1 `shouldSatisfy` er+                snd t2 `shouldSatisfy` er+                stopSupervisor sup+                wait g `shouldReturn` ()