extensible-effects-concurrent (empty) → 0.1.0.0
raw patch · 11 files changed
+688/−0 lines, 11 filesdep +QuickCheckdep +basedep +containerssetup-changed
Dependencies added: QuickCheck, base, containers, directory, extensible-effects, extensible-effects-concurrent, filepath, lens, mtl, parallel, process, random, stm, tagged, time, transformers
Files
- ChangeLog.md +3/−0
- LICENSE +30/−0
- README.md +9/−0
- Setup.hs +2/−0
- extensible-effects-concurrent.cabal +80/−0
- src/Control/Eff/Interactive.hs +74/−0
- src/Control/Eff/Processes.hs +72/−0
- src/Control/Eff/Processes/Examples.hs +68/−0
- src/Control/Eff/Processes/STM.hs +198/−0
- src/Control/Eff/Processes/Server.hs +150/−0
- test/Spec.hs +2/−0
+ ChangeLog.md view
@@ -0,0 +1,3 @@+# Changelog for extensible-effects-concurrent++## Unreleased changes
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Sven Heyll (c) 2018++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Sven Heyll nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,9 @@+# extensible-effects-concurrent++[](https://travis-ci.org/sheyll/extensible-effects-concurrent) ++[](http://hackage.haskell.org/package/extensible-effects-concurrent) ++[](http://stackage.org/lts/package/extensible-effects-concurrent)++
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ extensible-effects-concurrent.cabal view
@@ -0,0 +1,80 @@+name: extensible-effects-concurrent+version: 0.1.0.0+description: Please see the README on GitHub at <https://github.com/sheyll/extensible-effects-concurrent#readme>+homepage: https://github.com/sheyll/extensible-effects-concurrent#readme+bug-reports: https://github.com/sheyll/extensible-effects-concurrent/issues+author: Sven Heyll+maintainer: sven.heyll@gmail.com+copyright: Copyright Sven Heyll+license: BSD3+license-file: LICENSE+build-type: Simple+cabal-version: >= 1.10++extra-source-files:+ ChangeLog.md+ README.md++source-repository head+ type: git+ location: https://github.com/sheyll/extensible-effects-concurrent++library+ hs-source-dirs:+ src+ build-depends:+ base >=4.9 && <5,+ directory,+ filepath,+ time,+ mtl,+ containers,+ QuickCheck <2.11,+ lens,+ transformers,+ parallel,+ process,+ random,+ extensible-effects,+ stm,+ tagged+ exposed-modules:+ Control.Eff.Interactive,+ Control.Eff.Processes,+ Control.Eff.Processes.STM,+ Control.Eff.Processes.Server+ other-modules:+ Paths_extensible_effects_concurrent,+ Control.Eff.Processes.Examples+ other-extensions:+ ConstraintKinds,+ DeriveFoldable,+ DeriveFunctor,+ DeriveFunctor,+ DeriveGeneric,+ DeriveTraversable,+ FlexibleContexts,+ GeneralizedNewtypeDeriving,+ RankNTypes,+ ScopedTypeVariables,+ StandaloneDeriving,+ TemplateHaskell,+ TupleSections,+ TypeApplications,+ TypeInType,+ TypeOperators+ default-language: Haskell2010+++test-suite extensible-effects-concurrent-test+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ hs-source-dirs:+ test+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends:+ base >=4.9 && <5+ , extensible-effects-concurrent+ other-modules:+ Paths_extensible_effects_concurrent+ default-language: Haskell2010
+ src/Control/Eff/Interactive.hs view
@@ -0,0 +1,74 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE GADTs #-}+module Control.Eff.Interactive+ ( Interaction(..)+ , printLine+ , printStep+ , promptStep+ , step+ , Interactive(..)+ , interactiveInterpreter+ , runInteractionIOE+ , runInteractionIO+ )+where++import Control.Eff+import Control.Eff.Lift+import Control.Eff.Operational+import Control.Monad ( void )+import GHC.Stack++data Interaction a where+ PrintLine :: String -> Interaction ()+ ReadLine :: (String -> a) -> Interaction a+ Step :: Interaction ()++printLine+ :: (Member (Program Interaction) r, HasCallStack) => String -> Eff r ()+printLine = singleton . PrintLine++printStep+ :: (Member (Program Interaction) r, HasCallStack) => String -> Eff r ()+printStep m = do+ printLine m+ step++step :: (Member (Program Interaction) r, HasCallStack) => Eff r ()+step = do+ printLine "\nPress ENTER to continue."+ singleton Step++promptStep+ :: (Member (Program Interaction) r, HasCallStack)+ => String+ -> (String -> a)+ -> Eff r a+promptStep m p = do+ singleton (PrintLine m)+ singleton (ReadLine p)++class Interactive f where+ singleSteps :: (Member (Program Interaction) r, HasCallStack) => f a -> Eff r a++interactiveInterpreter+ :: (HasCallStack, Member (Program Interaction) r, Interactive f)+ => Eff (Program f ': r) a+ -> Eff r a+interactiveInterpreter = runProgram singleSteps++runInteractionIOE+ :: (SetMember Lift (Lift IO) r, HasCallStack)+ => Eff (Program Interaction ': r) a+ -> Eff r a+runInteractionIOE = runProgram go+ where+ go :: (SetMember Lift (Lift IO) r, HasCallStack) => Interaction a -> Eff r a+ go (PrintLine m) = lift (putStrLn m)+ go Step = lift (void getLine)+ go (ReadLine parse) = lift (parse <$> getLine)++runInteractionIO :: Eff '[Program Interaction, Lift IO] a -> IO a+runInteractionIO = runLift . runInteractionIOE
+ src/Control/Eff/Processes.hs view
@@ -0,0 +1,72 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE GADTs #-}+module Control.Eff.Processes+ ( ProcessId(..)+ , self+ , sendMessage+ , receiveMessage+ , Process(..)+ , fromProcessId+ )+where++import Control.Eff+import Control.Lens+import Data.Dynamic+import Data.Proxy+import Text.Printf++-- * Process Types++newtype ProcessId = ProcessId { _fromProcessId :: Int }+ deriving (Eq,Ord,Typeable,Bounded,Num, Enum, Integral, Real)++instance Read ProcessId where+ readsPrec _ ('<':'0':'.':rest1) =+ case reads rest1 of+ [(c, '.':'0':'>':rest2)] -> [(ProcessId c, rest2)]+ _ -> []+ readsPrec _ _ = []++instance Show ProcessId where+ show (ProcessId c) =+ printf "<0.%d.0>" c++makeLenses ''ProcessId++-- * Process Effect++data Process b where+ SendMessage :: Typeable m+ => ProcessId+ -> m+ -> Process Bool+ ReceiveMessage+ :: forall e m . (Typeable m)+ => (m -> e)+ -> Process (Maybe e)+ SelfPid :: Process ProcessId++-- * Process Effects++self :: Member Process r => Eff r ProcessId+self = send SelfPid++sendMessage+ :: forall o r . (Member Process r, Typeable o) => ProcessId -> o -> Eff r Bool+sendMessage pid message = send (SendMessage pid message)++receiveMessage+ :: forall o r . (Member Process r, Typeable o) => Proxy o -> Eff r (Maybe o)+receiveMessage _ = send (ReceiveMessage id)
+ src/Control/Eff/Processes/Examples.hs view
@@ -0,0 +1,68 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE GADTs #-}+module Control.Eff.Processes.Examples where++import GHC.Stack+import Control.Eff+import Control.Eff.Lift+import Control.Monad+import Data.Dynamic+import Data.Void++import Control.Eff.Processes+import Control.Eff.Processes.Server+import Control.Eff.Processes.STM++data TestServer+ deriving Typeable++data instance Api TestServer x where+ SayHello :: String -> Api TestServer Bool+ Shout :: String -> Api TestServer Void+ deriving Typeable++example :: (HasCallStack, HasScheduler r, Member Process r, SetMember Lift (Lift IO) r) => Eff r ()+example = do+ me <- self+ lift (putStrLn ("I am " ++ show me))+ server <- asServer @TestServer <$> spawn testa+ lift (putStrLn ("Started server " ++ show server))+ let go = do+ x <- lift (putStr ">>> " >> getLine)+ res <- call server (SayHello x)+ lift (putStrLn ("Result: " ++ show res))+ pid2 <- asServer @TestServer <$>+ spawn (do Just m <- serve (\(Shout m) k -> (m, k undefined))+ cast_ server (Shout ("Who: " ++ m)))+ x2 <- lift (putStr "<<< " >> getLine)+ cast_ pid2 (Shout x2)+ case x2 of+ ('q':_) -> return ()+ _ -> go+ go++testa :: (HasCallStack, Member Process r, SetMember Lift (Lift IO) r) => Eff r ()+testa =+ forever $+ do me <- self+ Just msg <- serve (testaRespond me)+ lift (putStrLn msg)+ return ()++testaRespond :: ProcessId -> Api TestServer x -> (x -> z) -> (String, z)+testaRespond me (SayHello x) k =+ (show me ++ " Got Hello: " ++ x, k (length x > 3))+testaRespond me (Shout x) k =+ (show me ++ " Shouting: " ++ x, k undefined )
+ src/Control/Eff/Processes/STM.hs view
@@ -0,0 +1,198 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE GADTs #-}+module Control.Eff.Processes.STM+ ( ProcessIOEffects+ , HasScheduler+ , Scheduler+ -- , nextPid, processTable+ , ProcessInfo+ , processId+ -- , messageQ+ , runScheduler+ , getProcessInfo+ , spawn+ , dispatchMessages+ -- , withMessageQueue+ )+where++import GHC.Stack+import Data.Kind+import Control.Concurrent as Concurrent+import Control.Concurrent.STM as STM+import Control.Eff+import Control.Eff.Lift+import Control.Eff.Processes+import Control.Eff.Reader.Strict as Reader+import Control.Lens+import qualified Control.Monad.State as Mtl+import Data.Dynamic+import Data.Typeable ( typeRep )+import Data.Map ( Map )+import qualified Data.Map as Map++-- * Process Scheduling++data ProcessInfo =+ ProcessInfo { _processId :: ProcessId+ , _messageQ :: STM.TQueue Dynamic+ }++makeLenses ''ProcessInfo++instance Show ProcessInfo where+ show p =+ "Process: " ++ show (p ^. processId)++data Scheduler =+ Scheduler { _nextPid :: ProcessId+ , _processTable :: Map ProcessId ProcessInfo+ }+ deriving Show+makeLenses ''Scheduler++type family Members (es :: [* -> *]) (r :: [* -> *]) :: Constraint where+ Members '[] r = ()+ Members (e ': es) r = (Member e r, Members es r)++type HasScheduler r = ( HasCallStack+ , SetMember Lift (Lift IO) r+ , Member (Reader (STM.TVar Scheduler)) r)++type ProcessIOEffects = '[Process, Reader (STM.TVar Scheduler), Lift IO]++runScheduler+ :: (SetMember Lift (Lift IO) r, HasCallStack)+ => Eff (Reader (STM.TVar Scheduler) ': r) ()+ -> Eff r ()+runScheduler e = do+ v <- lift (newTVarIO (Scheduler 1 Map.empty))+ runReader e v++getProcessInfo :: HasScheduler r => ProcessId -> Eff r (Maybe ProcessInfo)+getProcessInfo pid = do+ p <- getScheduler+ return (p ^. processTable . at pid)++-- ** Process execution++spawn :: HasScheduler r => Eff ProcessIOEffects () -> Eff r ProcessId+spawn mfa = do+ processes <- ask+ pidVar <- lift newEmptyTMVarIO+ _threadId <- lift+ (Concurrent.forkIO+ (runLift+ (runReader+ (dispatchMessages+ (do+ pid <- self+ lift (atomically (STM.putTMVar pidVar pid))+ mfa+ )+ )+ processes+ )+ )+ )+ lift (atomically (STM.takeTMVar pidVar))++dispatchMessages+ :: forall r a+ . (HasScheduler r, HasCallStack)+ => Eff (Process ': r) a+ -> Eff r a+dispatchMessages processAction = withMessageQueue+ (\pinfo -> handle_relay return (go pinfo) processAction)+ where+ go+ :: forall v+ . HasCallStack+ => ProcessInfo+ -> Process v+ -> (v -> Eff r a)+ -> Eff r a+ go (ProcessInfo selfPidInt _) SelfPid k = k selfPidInt+ go _ (SendMessage toPid reqIn) k = do+ psVar <- getSchedulerTVar+ lift+ (atomically+ (do+ p <- readTVar psVar+ let mto = p ^. processTable . at toPid+ case mto of+ Just toProc ->+ let dReq = toDyn reqIn+ in do+ writeTQueue (toProc ^. messageQ) dReq+ return True+ Nothing -> return False+ )+ )+ >>= k+ go (ProcessInfo selfPidInt channel) (ReceiveMessage onMsg) k = do+ mDynMsg <- lift (atomically (readTQueue channel))+ case fromDynamic mDynMsg of+ Just req -> let result = onMsg req in k (Just result)+ nix -> do+ lift+ (putStrLn+ ( show selfPidInt+ ++ " got unexpected msg: "+ ++ show mDynMsg+ ++ " expected: "+ ++ show (typeRep nix)+ )+ )+ k Nothing++withMessageQueue :: HasScheduler r => (ProcessInfo -> Eff r a) -> Eff r a+withMessageQueue m = do+ pinfo <- createQueue+ res <- m pinfo+ destroyQueue pinfo+ return res+ where+ createQueue = overScheduler+ (do+ pid <- nextPid <<+= 1+ channel <- Mtl.lift newTQueue+ let pinfo = ProcessInfo pid channel+ processTable . at pid .= Just pinfo+ return pinfo+ )+ destroyQueue pinfo =+ overScheduler (processTable . at (pinfo ^. processId) .= Nothing)+++overScheduler :: HasScheduler r => Mtl.StateT Scheduler STM.STM a -> Eff r a+overScheduler stAction = do+ psVar <- ask+ lift+ (STM.atomically+ (do+ ps <- STM.readTVar psVar+ (result, psModified) <- Mtl.runStateT stAction ps+ STM.writeTVar psVar psModified+ return result+ )+ )++getSchedulerTVar :: HasScheduler r => Eff r (TVar Scheduler)+getSchedulerTVar = ask++getScheduler :: HasScheduler r => Eff r Scheduler+getScheduler = do+ processesVar <- getSchedulerTVar+ lift (atomically (readTVar processesVar))
+ src/Control/Eff/Processes/Server.hs view
@@ -0,0 +1,150 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE GADTs #-}++-- | Type safe /server/ API processes++module Control.Eff.Processes.Server+ ( Api+ , Server(..)+ , fromServer+ , proxyAsServer+ , asServer+ , cast+ , cast_+ , call+ , serve+ )+where++import GHC.Stack+import Data.Kind+import Control.Eff+import Control.Lens+import Control.Monad+import Data.Dynamic+import Data.Typeable ( typeRep )+import Data.Proxy++import Control.Eff.Processes++-- | This data family defines an API implemented by a server.+-- The first parameter is the API /index/ and the second parameter+-- (the @* -> *@)+data family Api o :: * -> *++newtype Server o = Server { _fromServer :: ProcessId }+ deriving (Eq,Ord,Typeable)++instance Read (Server o) where+ readsPrec _ ('[':'#':rest1) =+ case reads (dropWhile (/= '⇒') rest1) of+ [(c, ']':rest2)] -> [(Server c, rest2)]+ _ -> []+ readsPrec _ _ = []++instance Typeable o => Show (Server o) where+ show s@(Server c) =+ "[#" ++ show (typeRep s) ++ "⇒" ++ show c ++ "]"++makeLenses ''Server++proxyAsServer :: proxy api -> ProcessId -> Server api+proxyAsServer = const Server++asServer :: ProcessId -> Server api+asServer = Server++data Request p where+ Call :: forall p x . (Typeable p, Typeable x, Typeable (Api p x)) => Maybe ProcessId -> Api p x -> Request p+ Cast :: forall p x . (Typeable p, Typeable x, Typeable (Api p x)) => Api p x -> Request p+ deriving Typeable++data Response p x where+ Response :: (Typeable p, Typeable x) => Proxy p -> x -> Response p x+ deriving Typeable++cast+ :: forall r o result+ . ( HasCallStack+ , Member Process r+ , Typeable o+ , Typeable result+ , Typeable (Api o result)+ )+ => Server o+ -> Api o result+ -> Eff r Bool+cast (Server pid) callMsg = sendMessage pid (Cast callMsg)++cast_+ :: forall r o result+ . ( HasCallStack+ , Member Process r+ , Typeable o+ , Typeable result+ , Typeable (Api o result)+ )+ => Server o+ -> Api o result+ -> Eff r ()+cast_ = ((.) . (.)) void cast++call+ :: forall result o r+ . ( Member Process r+ , Typeable o+ , Typeable (Api o result)+ , Typeable result+ , HasCallStack+ )+ => Server o+ -> Api o result+ -> Eff r (Maybe result)+call (Server pidInt) req = do+ fromPid <- self+ let requestMessage = Call (Just fromPid) req+ wasSent <- sendMessage pidInt requestMessage+ if wasSent+ then+ let responseMessage :: Response o result -> result+ responseMessage (Response _pxReply reply) = reply+ in send (ReceiveMessage responseMessage)+ else fail "Could not send request message " >> return Nothing++serve+ :: forall r p e+ . (Member Process r, Typeable p, HasCallStack)+ => ( forall x+ . (Typeable (Api p x), Typeable x)+ => Api p x+ -> (x -> Maybe (Process Bool))+ -> (e, Maybe (Process Bool))+ )+ -> Eff r (Maybe e)+serve handle = do+ mrequest <- send (ReceiveMessage receiveCallReq)+ case mrequest of+ Just (result, mReplyCast) -> do+ mapM_ (void . send) mReplyCast+ return (Just result)+ _ -> fail "sdf4446" >> return Nothing+ where+ receiveCallReq :: Request p -> (e, Maybe (Process Bool))+ receiveCallReq (Cast request ) = handle request (const Nothing)+ receiveCallReq (Call mFromPid request) = handle request (packReply request)+ where+ packReply :: Typeable x => Api p x -> x -> Maybe (Process Bool)+ packReply _ reply = do+ fromPid <- mFromPid+ return (SendMessage fromPid (Response (Proxy :: Proxy p) reply))
+ test/Spec.hs view
@@ -0,0 +1,2 @@+main :: IO ()+main = putStrLn "Test suite not yet implemented"