diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,11 +1,15 @@ # Changelog for extensible-effects-concurrent -## Unreleased Changes +## 0.1.1.0 +* Substantial API reoganization +* Rename/Move modules + +## 0.1.0.1 + * Stack/Cabal/Github Cosmetics * Travis build job ## 0.1.0.0 * Initial Version - diff --git a/extensible-effects-concurrent.cabal b/extensible-effects-concurrent.cabal --- a/extensible-effects-concurrent.cabal +++ b/extensible-effects-concurrent.cabal @@ -1,5 +1,5 @@ name: extensible-effects-concurrent -version: 0.1.0.1 +version: 0.1.1.0 description: Please see the README on GitHub at synopsis: Message passing concurrency as extensible-effect homepage: https://github.com/sheyll/extensible-effects-concurrent#readme @@ -41,12 +41,12 @@ tagged exposed-modules: Control.Eff.Interactive, - Control.Eff.Processes, - Control.Eff.Processes.STM, - Control.Eff.Processes.Server + Control.Eff.Concurrent.GenServer, + Control.Eff.Concurrent.MessagePassing, + Control.Eff.Concurrent.Dispatcher other-modules: Paths_extensible_effects_concurrent, - Control.Eff.Processes.Examples + Control.Eff.Concurrent.Examples other-extensions: ConstraintKinds, DeriveFoldable, diff --git a/src/Control/Eff/Concurrent/Dispatcher.hs b/src/Control/Eff/Concurrent/Dispatcher.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Dispatcher.hs @@ -0,0 +1,206 @@ +{-# 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.Concurrent.Dispatcher + ( ProcIO + , HasProcesses + , Scheduler + -- , nextPid, processTable + , ProcessInfo + , processId + -- , messageQ + , runProcesses + , 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.Concurrent.MessagePassing +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 + +-- * MessagePassing Scheduling + +data ProcessInfo = + ProcessInfo { _processId :: ProcessId + , _messageQ :: STM.TQueue Dynamic + } + +makeLenses ''ProcessInfo + +instance Show ProcessInfo where + show p = + "ProcessInfo: " ++ 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 HasProcesses r = ( HasCallStack + , SetMember Lift (Lift IO) r + , Member (Reader (STM.TVar Scheduler)) r) + +type ProcIO = '[MessagePassing, Process, Reader (STM.TVar Scheduler), Lift IO] + +runProcesses + :: (SetMember Lift (Lift IO) r, HasCallStack) + => Eff (Reader (STM.TVar Scheduler) ': r) () + -> Eff r () +runProcesses e = do + v <- lift (newTVarIO (Scheduler 1 Map.empty)) + runReader e v + +getProcessInfo :: HasProcesses r => ProcessId -> Eff r (Maybe ProcessInfo) +getProcessInfo pid = do + p <- getScheduler + return (p ^. processTable . at pid) + +-- ** MessagePassing execution + +spawn :: HasProcesses r => Eff ProcIO () -> 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 + . (HasProcesses r, HasCallStack) + => Eff (MessagePassing ': Process ': r) a + -> Eff r a +dispatchMessages processAction = withMessageQueue + (\pinfo -> + handle_relay return (goProc pinfo) (handle_relay return (go pinfo) processAction)) + where + go + :: forall v + . HasCallStack + => ProcessInfo + -> MessagePassing v + -> (v -> Eff (Process ': r) a) + -> Eff (Process ': r) a + 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 + goProc + :: forall v + . HasCallStack + => ProcessInfo + -> Process v + -> (v -> Eff r a) + -> Eff r a + goProc (ProcessInfo selfPidInt _) SelfPid k = k selfPidInt + +withMessageQueue :: HasProcesses 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 :: HasProcesses 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 :: HasProcesses r => Eff r (TVar Scheduler) +getSchedulerTVar = ask + +getScheduler :: HasProcesses r => Eff r Scheduler +getScheduler = do + processesVar <- getSchedulerTVar + lift (atomically (readTVar processesVar)) diff --git a/src/Control/Eff/Concurrent/Examples.hs b/src/Control/Eff/Concurrent/Examples.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Examples.hs @@ -0,0 +1,75 @@ +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE ConstraintKinds #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# 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.Concurrent.Examples where + +import GHC.Stack +import Control.Eff +import Control.Eff.Lift +import Control.Monad +import Data.Dynamic + +import Control.Eff.Concurrent.MessagePassing +import Control.Eff.Concurrent.GenServer +import Control.Eff.Concurrent.Dispatcher + +data TestApi + deriving Typeable + +data instance Api TestApi x where + SayHello :: String -> Api TestApi ('Synchronous Bool) + Shout :: String -> Api TestApi 'Asynchronous + deriving (Typeable) + +deriving instance Show (Api TestApi x) + +runExample :: IO () +runExample = runLift $ runProcesses $ dispatchMessages example + +example + :: ( HasCallStack + , HasProcesses r + , Member MessagePassing r + , Member Process r + , SetMember Lift (Lift IO) r) + => Eff r () +example = do + me <- self + lift (putStrLn ("I am " ++ show me)) + server <- asServer @TestApi <$> spawn testServerLoop + lift (putStrLn ("Started server " ++ show server)) + let go = do + x <- lift (putStr ">>> " >> getLine) + res <- call server (SayHello x) + lift (putStrLn ("Result: " ++ show res)) + case x of + ('q':_) -> return () + _ -> go + go + +testServerLoop + :: forall r. (HasCallStack, Member MessagePassing r, Member Process r, SetMember Lift (Lift IO) r) + => Eff r () +testServerLoop = forever $ serve_ $ ApiHandler handleCast handleCall + where + handleCast :: Api TestApi 'Asynchronous -> Eff r () + handleCast (Shout x) = do + me <- self + lift (putStrLn (show me ++ " Shouting: " ++ x)) + handleCall :: Api TestApi ('Synchronous x) -> (x -> Eff r Bool) -> Eff r () + handleCall (SayHello x) reply = do + me <- self + lift (putStrLn (show me ++ " Got Hello: " ++ x)) + void (reply (length x > 3)) diff --git a/src/Control/Eff/Concurrent/GenServer.hs b/src/Control/Eff/Concurrent/GenServer.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/GenServer.hs @@ -0,0 +1,211 @@ +{-# 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 #-} + +-- | Type safe /server/ API processes + +module Control.Eff.Concurrent.GenServer + ( Api + , Synchronicity(..) + , ApiHandler(ApiHandler) + , Server(..) + , fromServer + , proxyAsServer + , asServer + , cast + , cast_ + , call + , serve + , serve_ + , unhandledCallError + , unhandledCastError + ) +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.Concurrent.MessagePassing + +-- | 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 :: Synchronicity -> * + +data Synchronicity = + Synchronous Type | Asynchronous + deriving (Typeable) + +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 ('Synchronous x))) + => ProcessId -> Api p ('Synchronous x) -> Request p + Cast :: forall p . (Typeable p, Typeable (Api p 'Asynchronous)) + => Api p 'Asynchronous -> 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 + . ( HasCallStack + , Member MessagePassing r + , Typeable o + , Typeable (Api o 'Asynchronous) + ) + => Server o + -> Api o 'Asynchronous + -> Eff r Bool +cast (Server pid) callMsg = sendMessage pid (Cast callMsg) + +cast_ + :: forall r o + . ( HasCallStack + , Member MessagePassing r + , Typeable o + , Typeable (Api o 'Asynchronous) + ) + => Server o + -> Api o 'Asynchronous + -> Eff r () +cast_ = ((.) . (.)) void cast + +call + :: forall result o r + . ( Member MessagePassing r + , Member Process r + , Typeable o + , Typeable (Api o ('Synchronous result)) + , Typeable result + , HasCallStack + ) + => Server o + -> Api o ('Synchronous result) + -> Eff r (Maybe result) +call (Server pidInt) req = do + fromPid <- self + let requestMessage = Call fromPid req + wasSent <- sendMessage pidInt requestMessage + if wasSent + then + let extractResult :: Response o result -> result + extractResult (Response _pxResult result) = result + in do mResp <- receiveMessage (Proxy @(Response o result)) + return (extractResult <$> mResp) + else fail "Could not send request message " >> return Nothing + +data ApiHandler p r e where + ApiHandler :: + { _handleCast + :: (Typeable p, Typeable (Api p 'Asynchronous), HasCallStack) + => Api p 'Asynchronous -> Eff r e + , _handleCall + :: forall x . (Typeable p, Typeable (Api p ('Synchronous x)), Typeable x, HasCallStack) + => Api p ('Synchronous x) -> (x -> Eff r Bool) -> Eff r e + } -> ApiHandler p r e + +serve_ + :: forall r p + . (Typeable p, Member MessagePassing r, Member Process r, HasCallStack) + => ApiHandler p r () + -> Eff r () +serve_ = void . serve + +serve + :: forall r p e + . (Typeable p, Member MessagePassing r, Member Process r, HasCallStack) + => ApiHandler p r e + -> Eff r (Maybe e) +serve (ApiHandler handleCast handleCall) = do + mReq <- receiveMessage (Proxy @(Request p)) + mapM receiveCallReq mReq + where + receiveCallReq :: Request p -> Eff r e + receiveCallReq (Cast request ) = handleCast request + receiveCallReq (Call fromPid request) = handleCall request + (sendReply request) + where + sendReply :: Typeable x => Api p ('Synchronous x) -> x -> Eff r Bool + sendReply _ reply = sendMessage fromPid (Response (Proxy :: Proxy p) reply) + +unhandledCallError + :: ( Show (Api p ('Synchronous x)) + , Typeable p + , Typeable (Api p ('Synchronous x)) + , Typeable x + , HasCallStack + , Member Process r + ) + => Api p ('Synchronous x) + -> (x -> Eff r Bool) + -> Eff r e +unhandledCallError api _ = do + me <- self + fail + ( show me + ++ " Unhandled call: (" + ++ show api + ++ " :: " + ++ show (typeRep api) + ++ ")" + ) + +unhandledCastError + :: ( Show (Api p 'Asynchronous) + , Typeable p + , Typeable (Api p 'Asynchronous) + , HasCallStack + , Member Process r + ) + => Api p 'Asynchronous + -> Eff r e +unhandledCastError api = do + me <- self + fail + ( show me + ++ " Unhandled cast: (" + ++ show api + ++ " :: " + ++ show (typeRep api) + ++ ")" + ) diff --git a/src/Control/Eff/Concurrent/MessagePassing.hs b/src/Control/Eff/Concurrent/MessagePassing.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/MessagePassing.hs @@ -0,0 +1,99 @@ +{-# 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.Concurrent.MessagePassing + ( ProcessId(..) + , fromProcessId + , Process(..) + , self + , MessagePassing(..) + , sendMessage + , kill + , receiveMessage + ) +where + +import GHC.Stack +import Control.Eff +import Control.Lens +import Data.Dynamic +import Data.Proxy +import Data.Void +import Text.Printf + + +-- * Process Effects + +data Process b where + SelfPid :: Process ProcessId + -- LinkProcesses :: ProcessId -> ProcessId -> Process () + +self :: Member Process r => Eff r ProcessId +self = send SelfPid + +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 + + +-- * MessagePassing Effect + +data MessagePassing b where + SendMessage :: Typeable m + => ProcessId + -> Message m + -> MessagePassing Bool + ReceiveMessage + :: forall e m . (Typeable m, Typeable (Message m)) + => (Message m -> e) + -> MessagePassing (Maybe e) + +data Message m where + Shutdown :: Message Void + Message :: m -> Message m + deriving Typeable + +sendMessage + :: forall o r + . (HasCallStack, Member MessagePassing r, Typeable o) + => ProcessId + -> o + -> Eff r Bool +sendMessage pid message = send (SendMessage pid (Message message)) + +kill :: (HasCallStack, Member MessagePassing r) + => ProcessId -> Eff r Bool +kill pid = send (SendMessage pid Shutdown) + +receiveMessage + :: forall o r . (HasCallStack, Member MessagePassing r, Member Process r, Typeable o) + => Proxy o -> Eff r (Maybe o) +receiveMessage _ = do + me <- self + mRes <- send (ReceiveMessage id) + case mRes of + Just Shutdown -> fail (show me ++ " SHUTDOWN") + Just (Message m) -> return (Just m) + Nothing -> return Nothing diff --git a/src/Control/Eff/Processes.hs b/src/Control/Eff/Processes.hs deleted file mode 100644 --- a/src/Control/Eff/Processes.hs +++ /dev/null @@ -1,72 +0,0 @@ -{-# 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) diff --git a/src/Control/Eff/Processes/Examples.hs b/src/Control/Eff/Processes/Examples.hs deleted file mode 100644 --- a/src/Control/Eff/Processes/Examples.hs +++ /dev/null @@ -1,68 +0,0 @@ -{-# 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 ) diff --git a/src/Control/Eff/Processes/STM.hs b/src/Control/Eff/Processes/STM.hs deleted file mode 100644 --- a/src/Control/Eff/Processes/STM.hs +++ /dev/null @@ -1,198 +0,0 @@ -{-# 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)) diff --git a/src/Control/Eff/Processes/Server.hs b/src/Control/Eff/Processes/Server.hs deleted file mode 100644 --- a/src/Control/Eff/Processes/Server.hs +++ /dev/null @@ -1,150 +0,0 @@ -{-# 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))