Control.Eff.Concurrent

Erlang style processes with message passing concurrency based on (more) extensible-effects.

Concurrent Processes with Message Passing Concurrency

newtype ProcessId

Each process is identified by a single process id, that stays constant throughout the life cycle of a process. Also, message sending relies on these values to address messages to processes.

Constructors

ProcessId

Fields

• _fromProcessId :: Int

Instances

Bounded ProcessId

Defined in Control.Eff.Concurrent.Process

Methods

minBound :: ProcessId

maxBound :: ProcessId

Enum ProcessId

Defined in Control.Eff.Concurrent.Process

Methods

succ :: ProcessId -> ProcessId

pred :: ProcessId -> ProcessId

toEnum :: Int -> ProcessId

fromEnum :: ProcessId -> Int

enumFrom :: ProcessId -> [ProcessId]

enumFromThen :: ProcessId -> ProcessId -> [ProcessId]

enumFromTo :: ProcessId -> ProcessId -> [ProcessId]

enumFromThenTo :: ProcessId -> ProcessId -> ProcessId -> [ProcessId]

Eq ProcessId

Defined in Control.Eff.Concurrent.Process

Methods

(==) :: ProcessId -> ProcessId -> Bool

(/=) :: ProcessId -> ProcessId -> Bool

Integral ProcessId

Defined in Control.Eff.Concurrent.Process

Methods

quot :: ProcessId -> ProcessId -> ProcessId

rem :: ProcessId -> ProcessId -> ProcessId

div :: ProcessId -> ProcessId -> ProcessId

mod :: ProcessId -> ProcessId -> ProcessId

quotRem :: ProcessId -> ProcessId -> (ProcessId, ProcessId)

divMod :: ProcessId -> ProcessId -> (ProcessId, ProcessId)

toInteger :: ProcessId -> Integer

Num ProcessId

Defined in Control.Eff.Concurrent.Process

Methods

(+) :: ProcessId -> ProcessId -> ProcessId

(-) :: ProcessId -> ProcessId -> ProcessId

(*) :: ProcessId -> ProcessId -> ProcessId

negate :: ProcessId -> ProcessId

abs :: ProcessId -> ProcessId

signum :: ProcessId -> ProcessId

fromInteger :: Integer -> ProcessId

Ord ProcessId

Defined in Control.Eff.Concurrent.Process

Methods

compare :: ProcessId -> ProcessId -> Ordering

(<) :: ProcessId -> ProcessId -> Bool

(<=) :: ProcessId -> ProcessId -> Bool

(>) :: ProcessId -> ProcessId -> Bool

(>=) :: ProcessId -> ProcessId -> Bool

max :: ProcessId -> ProcessId -> ProcessId

min :: ProcessId -> ProcessId -> ProcessId

Read ProcessId

Defined in Control.Eff.Concurrent.Process

Methods

readsPrec :: Int -> ReadS ProcessId

readList :: ReadS [ProcessId]

readPrec :: ReadPrec ProcessId

readListPrec :: ReadPrec [ProcessId]

Real ProcessId

Defined in Control.Eff.Concurrent.Process

Methods

toRational :: ProcessId -> Rational

Show ProcessId

Defined in Control.Eff.Concurrent.Process

Methods

showsPrec :: Int -> ProcessId -> ShowS

show :: ProcessId -> String

showList :: [ProcessId] -> ShowS

data SchedulerProxy :: [Type -> Type] -> Type where

Every function for Process things needs such a proxy value for the low-level effect list, i.e. the effects identified by r in Process r : r, this might be dependent on the scheduler implementation.

Constructors

SchedulerProxy :: SchedulerProxy q

Tell the typechecker what effects we have below Process

SP :: SchedulerProxy q

Like SchedulerProxy but shorter

type ConsProcess r = Process r ': r

Cons Process onto a list of effects.

data ResumeProcess v where

Every Process action returns it's actual result wrapped in this type. It will allow to signal errors as well as pass on normal results such as incoming messages.

Constructors

ShutdownRequested :: ResumeProcess v

The process is required to exit.

OnError :: String -> ResumeProcess v

The process is required to exit from an error condition, that cannot be recovered from.

ResumeWith :: a -> ResumeProcess a

The process may resume to do work, using the given result.

RetryLastAction :: ResumeProcess v

This indicates that the action did not complete, and maybe retried

Instances

Functor ResumeProcess

Defined in Control.Eff.Concurrent.Process

Methods

fmap :: (a -> b) -> ResumeProcess a -> ResumeProcess b

(<$) :: a -> ResumeProcess b -> ResumeProcess a

Foldable ResumeProcess

Defined in Control.Eff.Concurrent.Process

Methods

fold :: Monoid m => ResumeProcess m -> m

foldMap :: Monoid m => (a -> m) -> ResumeProcess a -> m

foldr :: (a -> b -> b) -> b -> ResumeProcess a -> b

foldr' :: (a -> b -> b) -> b -> ResumeProcess a -> b

foldl :: (b -> a -> b) -> b -> ResumeProcess a -> b

foldl' :: (b -> a -> b) -> b -> ResumeProcess a -> b

foldr1 :: (a -> a -> a) -> ResumeProcess a -> a

foldl1 :: (a -> a -> a) -> ResumeProcess a -> a

toList :: ResumeProcess a -> [a]

null :: ResumeProcess a -> Bool

length :: ResumeProcess a -> Int

elem :: Eq a => a -> ResumeProcess a -> Bool

maximum :: Ord a => ResumeProcess a -> a

minimum :: Ord a => ResumeProcess a -> a

sum :: Num a => ResumeProcess a -> a

product :: Num a => ResumeProcess a -> a

Traversable ResumeProcess

Defined in Control.Eff.Concurrent.Process

Methods

traverse :: Applicative f => (a -> f b) -> ResumeProcess a -> f (ResumeProcess b)

sequenceA :: Applicative f => ResumeProcess (f a) -> f (ResumeProcess a)

mapM :: Monad m => (a -> m b) -> ResumeProcess a -> m (ResumeProcess b)

sequence :: Monad m => ResumeProcess (m a) -> m (ResumeProcess a)

NFData1 ResumeProcess

Defined in Control.Eff.Concurrent.Process

Methods

liftRnf :: (a -> ()) -> ResumeProcess a -> ()

Eq v => Eq (ResumeProcess v)

Defined in Control.Eff.Concurrent.Process

Methods

(==) :: ResumeProcess v -> ResumeProcess v -> Bool

(/=) :: ResumeProcess v -> ResumeProcess v -> Bool

Ord v => Ord (ResumeProcess v)

Defined in Control.Eff.Concurrent.Process

Methods

compare :: ResumeProcess v -> ResumeProcess v -> Ordering

(<) :: ResumeProcess v -> ResumeProcess v -> Bool

(<=) :: ResumeProcess v -> ResumeProcess v -> Bool

(>) :: ResumeProcess v -> ResumeProcess v -> Bool

(>=) :: ResumeProcess v -> ResumeProcess v -> Bool

max :: ResumeProcess v -> ResumeProcess v -> ResumeProcess v

min :: ResumeProcess v -> ResumeProcess v -> ResumeProcess v

Show v => Show (ResumeProcess v)

Defined in Control.Eff.Concurrent.Process

Methods

showsPrec :: Int -> ResumeProcess v -> ShowS

show :: ResumeProcess v -> String

showList :: [ResumeProcess v] -> ShowS

Generic (ResumeProcess v)

Defined in Control.Eff.Concurrent.Process

Associated Types

type Rep (ResumeProcess v) :: * -> *

Methods

from :: ResumeProcess v -> Rep (ResumeProcess v) x

to :: Rep (ResumeProcess v) x -> ResumeProcess v

NFData a => NFData (ResumeProcess a)

Defined in Control.Eff.Concurrent.Process

Methods

rnf :: ResumeProcess a -> ()

Generic1 ResumeProcess

Defined in Control.Eff.Concurrent.Process

Associated Types

type Rep1 ResumeProcess :: k -> *

Methods

from1 :: ResumeProcess a -> Rep1 ResumeProcess a

to1 :: Rep1 ResumeProcess a -> ResumeProcess a

type Rep (ResumeProcess v)

Defined in Control.Eff.Concurrent.Process

type Rep (ResumeProcess v) = D1 (MetaData "ResumeProcess" "Control.Eff.Concurrent.Process" "extensible-effects-concurrent-0.6.2-Dn0edfhM7NrCsGRg2jlKcC" False) ((C1 (MetaCons "ShutdownRequested" PrefixI False) (U1 :: * -> *) :+: C1 (MetaCons "OnError" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 String))) :+: (C1 (MetaCons "ResumeWith" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 v)) :+: C1 (MetaCons "RetryLastAction" PrefixI False) (U1 :: * -> *)))

type Rep1 ResumeProcess

Defined in Control.Eff.Concurrent.Process

type Rep1 ResumeProcess = D1 (MetaData "ResumeProcess" "Control.Eff.Concurrent.Process" "extensible-effects-concurrent-0.6.2-Dn0edfhM7NrCsGRg2jlKcC" False) ((C1 (MetaCons "ShutdownRequested" PrefixI False) (U1 :: * -> *) :+: C1 (MetaCons "OnError" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 String))) :+: (C1 (MetaCons "ResumeWith" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) Par1) :+: C1 (MetaCons "RetryLastAction" PrefixI False) (U1 :: * -> *)))

data Process (r :: [Type -> Type]) b where

The process effect is the basis for message passing concurrency. This effect describes an interface for concurrent, communicating isolated processes identified uniquely by a process-id.

Processes can raise exceptions that can be caught, exit gracefully or with an error, or be killed by other processes, with the option of ignoring the shutdown request.

Process Scheduling is implemented in different modules. All scheduler implementations should follow some basic rules:

• fair scheduling
• sending a message does not block
• receiving a message does block
• spawning a child blocks only a very moment
• a newly spawned process shall be scheduled before the parent process after
• the spawn
• when the first process exists, all process should be killed immediately

Constructors

YieldProcess :: Process r (ResumeProcess ())

In cooperative schedulers, this will give processing time to the scheduler. Every other operation implicitly serves the same purpose.

SelfPid :: Process r (ResumeProcess ProcessId)

Return the current ProcessId

Spawn :: Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId)

Start a new process, the new process will execute an effect, the function will return immediately with a ProcessId.

Shutdown :: Process r a

Process exit, this is the same as if the function that was applied to a spawn function returned.

ExitWithError :: String -> Process r b

Exit the process due to an error, this cannot be caught.

RaiseError :: String -> Process r b

Raise an error, that can be handled.

SendShutdown :: ProcessId -> Process r (ResumeProcess Bool)

Request that another a process exits. The targeted process is interrupted and gets a ShutdownRequested, the target process may decide to ignore the shutdown requests.

SendMessage :: ProcessId -> Dynamic -> Process r (ResumeProcess Bool)

Send a message to a process addressed by the ProcessId. Sending a message should **always succeed** and return **immediately**, even if the destination process does not exist, or does not accept messages of the given type.

ReceiveMessage :: Process r (ResumeProcess Dynamic)

Receive a message. This should block until an a message was received. The pure function may convert the incoming message into something, and the result is returned as ProcessMessage value. Another reason why this function returns, is if a process control message was sent to the process. This can only occur from inside the runtime system, aka the effect handler implementation. (Currently there is one in ForkIOScheduler.)

executeAndResume :: forall r q v. (SetMember Process (Process q) r, HasCallStack) => Process q (ResumeProcess v) -> Eff r v

Execute a Process action and resume the process, retry the action or exit the process when a shutdown was requested.

executeAndCatch :: forall q r v. (SetMember Process (Process q) r, HasCallStack) => SchedulerProxy q -> Eff r (ResumeProcess v) -> Eff r (Either String v)

Execute a and action and resume the process, retry the action, shutdown the process or return an error.

thisSchedulerProxy :: Eff (Process r ': r) (SchedulerProxy r)

Return a SchedulerProxy for a Process effect.

yieldProcess :: forall r q. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r ()

Use executeAndResume to execute YieldProcess. Refer to YieldProcess for more information.

sendMessage :: forall r q. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> ProcessId -> Dynamic -> Eff r ()

Send a message to a process addressed by the ProcessId. See SendMessage.

sendMessageChecked :: forall r q. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> ProcessId -> Dynamic -> Eff r Bool

Send a message to a process addressed by the ProcessId. See SendMessage. Return True if the process existed. I you don't care, just sendMessage instead.

sendMessageAs :: forall o r q. (HasCallStack, SetMember Process (Process q) r, Typeable o) => SchedulerProxy q -> ProcessId -> o -> Eff r ()

Send a message to a process addressed by the ProcessId. See SendMessage.

sendShutdown :: forall r q. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> ProcessId -> Eff r ()

Exit a process addressed by the ProcessId. See SendShutdown.

sendShutdownChecked :: forall r q. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> ProcessId -> Eff r Bool

Like sendShutdown, but also return True iff the process to exit exists.

spawn :: forall r q. (HasCallStack, SetMember Process (Process q) r) => Eff (Process q ': q) () -> Eff r ProcessId

Start a new process, the new process will execute an effect, the function will return immediately with a ProcessId.

spawn_ :: forall r q. (HasCallStack, SetMember Process (Process q) r) => Eff (Process q ': q) () -> Eff r ()

Like spawn but return ().

receiveMessage :: forall r q. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r Dynamic

Block until a message was received.

receiveMessageAs :: forall a r q. (HasCallStack, Typeable a, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a

Receive and cast the message to some Typeable instance.

receiveLoop :: forall r q. (SetMember Process (Process q) r, HasCallStack) => SchedulerProxy q -> (Either (Maybe String) Dynamic -> Eff r ()) -> Eff r ()

Enter a loop to receive messages and pass them to a callback, until the function returns False.

self :: (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r ProcessId

Returns the ProcessId of the current process.

exitNormally :: forall r q a. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a

Exit the process.

exitWithError :: forall r q a. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> String -> Eff r a

Exit the process with an error.

raiseError :: forall r q b. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> String -> Eff r b

Thrown an error, can be caught by catchRaisedError.

catchRaisedError :: forall r q w. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> (String -> Eff r w) -> Eff r w -> Eff r w

Catch and handle an error raised by raiseError. Works independent of the handler implementation.

ignoreProcessError :: (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a -> Eff r (Either String a)

Like catchRaisedError it catches raiseError, but instead of invoking a user provided handler, the result is wrapped into an Either.

fromProcessId :: Iso' ProcessId Int

Data Types and Functions for APIs (aka Protocols)

newtype Server api

This is a tag-type that wraps around a ProcessId and holds an Api index type.

Constructors

Server

Fields

• _fromServer :: ProcessId

Instances

Eq (Server api)

Defined in Control.Eff.Concurrent.Api

Methods

(==) :: Server api -> Server api -> Bool

(/=) :: Server api -> Server api -> Bool

Ord (Server api)

Defined in Control.Eff.Concurrent.Api

Methods

compare :: Server api -> Server api -> Ordering

(<) :: Server api -> Server api -> Bool

(<=) :: Server api -> Server api -> Bool

(>) :: Server api -> Server api -> Bool

(>=) :: Server api -> Server api -> Bool

max :: Server api -> Server api -> Server api

min :: Server api -> Server api -> Server api

Read (Server api)

Defined in Control.Eff.Concurrent.Api

Methods

readsPrec :: Int -> ReadS (Server api)

readList :: ReadS [Server api]

readPrec :: ReadPrec (Server api)

readListPrec :: ReadPrec [Server api]

Typeable api => Show (Server api)

Defined in Control.Eff.Concurrent.Api

Methods

showsPrec :: Int -> Server api -> ShowS

show :: Server api -> String

showList :: [Server api] -> ShowS

data Synchronicity

The (promoted) constructors of this type specify (at the type level) the reply behavior of a specific constructor of an Api instance.

Constructors

Synchronous Type

Specify that handling a request is a blocking operation with a specific return type, e.g. ('Synchronous (Either RentalError RentalId))

Asynchronous

Non-blocking, asynchronous, request handling

data family Api (api :: Type) (reply :: Synchronicity)

This data family defines an API, a communication interface description between at least two processes. The processes act as servers or client(s) regarding a specific instance of this type.

The first parameter is usually a user defined phantom type that identifies the Api instance.

The second parameter specifies if a specific constructor of an (GADT-like) Api instance is Synchronous, i.e. returns a result and blocks the caller or if it is Asynchronous

Example:

data BookShop deriving Typeable

data instance Api BookShop r where
  RentBook  :: BookId   -> Api BookShop ('Synchronous (Either RentalError RentalId))
  BringBack :: RentalId -> Api BookShop 'Asynchronous

type BookId = Int
type RentalId = Int
type RentalError = String

Instances

Show (Observation o) => Show (Api (CallbackObserver o) r)

Defined in Control.Eff.Concurrent.Api.Observer

Methods

showsPrec :: Int -> Api (CallbackObserver o) r -> ShowS

show :: Api (CallbackObserver o) r -> String

showList :: [Api (CallbackObserver o) r] -> ShowS

data Api (CallbackObserver o) r

Defined in Control.Eff.Concurrent.Api.Observer

data Api (CallbackObserver o) r where

• CbObserved :: Api (CallbackObserver o) Asynchronous

fromServer :: forall api api. Iso (Server api) (Server api) ProcessId ProcessId

proxyAsServer :: proxy api -> ProcessId -> Server api

Tag a ProcessId with an Api type index to mark it a Server process handling that API

asServer :: forall api. ProcessId -> Server api

Tag a ProcessId with an Api type index to mark it a Server process handling that API

Client Functions for Consuming APIs

type ServesApi o r q = (Typeable o, SetMember Process (Process q) r, Member (Reader (Server o)) r)

Instead of passing around a Server value and passing to functions like cast or call, a Server can provided by a Reader effect, if there is only a single server for a given Api instance. This type alias is convenience to express that an effect has Process and a reader for a Server.

castChecked :: forall r q o. (HasCallStack, SetMember Process (Process q) r, Typeable o, Typeable (Api o Asynchronous)) => SchedulerProxy q -> Server o -> Api o Asynchronous -> Eff r Bool

Send an Api request that has no return value and return as fast as possible. The type signature enforces that the corresponding Api clause is Asynchronous. Return True if the message was sent to the process. Note that this is totally not the same as that the request was successfully handled. If that is important, use call instead.

cast :: forall r q o. (HasCallStack, SetMember Process (Process q) r, Typeable o, Typeable (Api o Asynchronous)) => SchedulerProxy q -> Server o -> Api o Asynchronous -> Eff r ()

Send an Api request that has no return value and return as fast as possible. The type signature enforces that the corresponding Api clause is Asynchronous.

call :: forall result api r q. (SetMember Process (Process q) r, Typeable api, Typeable (Api api (Synchronous result)), Typeable result, HasCallStack) => SchedulerProxy q -> Server api -> Api api (Synchronous result) -> Eff r result

Send an Api request and wait for the server to return a result value.

The type signature enforces that the corresponding Api clause is Synchronous.

registerServer :: Server o -> Eff (Reader (Server o) ': r) a -> Eff r a

Run a reader effect that contains the one server handling a specific Api instance.

callRegistered :: (Typeable reply, ServesApi o r q) => SchedulerProxy q -> Api o (Synchronous reply) -> Eff r reply

Like call but take the Server from the reader provided by registerServer.

callRegisteredA :: forall r q o f reply. (Alternative f, Typeable f, Typeable reply, ServesApi o r q) => SchedulerProxy q -> Api o (Synchronous (f reply)) -> Eff r (f reply)

Like callRegistered but also catch errors raised if e.g. the server crashed. By allowing Alternative instances to contain the reply, application level errors can be combined with errors rising from inter process communication.

castRegistered :: (Typeable o, ServesApi o r q) => SchedulerProxy q -> Api o Asynchronous -> Eff r ()

Like cast but take the Server from the reader provided by registerServer.

Server Functions for Providing APIs

data UnhandledRequest

An exception that is used by the mechanism that chains together multiple different ApiHandler allowing a single process to implement multiple Apis. This exception is thrown by requestFromDynamic. This exception can be caught with catchUnhandled, this way, several distinct ApiHandler can be tried until one fits or until the exitUnhandled is invoked.

data ApiHandler p r where

A record of callbacks, handling requests sent to a server Process, all belonging to a specific Api family instance.

Constructors

ApiHandler :: {..} -> ApiHandler p r

Fields

• _handleCast :: HasCallStack => Api p Asynchronous -> Eff r ()

  A cast will not return a result directly. This is used for async methods.

• _handleCall :: forall x. HasCallStack => Api p (Synchronous x) -> (x -> Eff r ()) -> Eff r ()

  A call is a blocking operation, the caller is blocked until this handler calls the reply continuation.

• _handleTerminate :: HasCallStack => Maybe String -> Eff r ()

  This callback is called with Nothing if the process exits peacefully, or Just "error message..." if the process exits with an error. This function is responsible to exit the process if necessary. The default behavior is defined in defaultTermination.

serve :: forall r q p. (Typeable p, SetMember Process (Process q) r, HasCallStack) => SchedulerProxy q -> ApiHandler p r -> Eff r ()

Receive and process incoming requests until the process exits, using an ApiHandler.

unhandledCallError :: forall p x r q. (Show (Api p (Synchronous x)), Typeable p, HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Api p (Synchronous x) -> (x -> Eff r ()) -> Eff r ()

A default handler to use in _handleCall in ApiHandler. It will call raiseError with a nice error message.

unhandledCastError :: forall p r q. (Show (Api p Asynchronous), Typeable p, HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Api p Asynchronous -> Eff r ()

A default handler to use in _handleCast in ApiHandler. It will call raiseError with a nice error message.

defaultTermination :: forall q r. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Maybe String -> Eff r ()

Exit the process either normally of the error message is Nothing or with exitWithError otherwise.

serveBoth :: forall r q p1 p2. (Typeable p1, Typeable p2, SetMember Process (Process q) r, HasCallStack) => SchedulerProxy q -> ApiHandler p1 r -> ApiHandler p2 r -> Eff r ()

serve two ApiHandlers at once. The first handler is used for termination handling.

serve3 :: forall r q p1 p2 p3. (Typeable p1, Typeable p2, Typeable p3, SetMember Process (Process q) r, HasCallStack) => SchedulerProxy q -> ApiHandler p1 r -> ApiHandler p2 r -> ApiHandler p3 r -> Eff r ()

serve three ApiHandlers at once. The first handler is used for termination handling.

tryApiHandler :: forall r q p. (Typeable p, SetMember Process (Process q) r, HasCallStack) => SchedulerProxy q -> ApiHandler p r -> Dynamic -> Eff (Exc UnhandledRequest ': r) ()

The basic building block of the combination of ApiHandlers is this function, which can not only be passed to receiveLoop, but also throws an UnhandledRequest exception if requestFromDynamic failed, such that multiple invokation of this function for different ApiHandlers can be tried, by using catchUnhandled.

tryApiHandler px handlers1 message
  `catchUnhandled`
tryApiHandler px handlers2
  `catchUnhandled`
tryApiHandler px handlers3

catchUnhandled :: forall r a. (Member (Exc UnhandledRequest) r, HasCallStack) => Eff r a -> (Dynamic -> Eff r a) -> Eff r a

If requestFromDynamic failes to cast the message to a Request for a certain ApiHandler it throws an UnhandledRequest exception. That exception is caught by this function and the raw message is given to the handler function. This is the basis for chaining ApiHandlers.

ensureAllHandled :: forall r q. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff (Exc UnhandledRequest ': r) () -> Eff r ()

Catch UnhandledRequests and terminate the process with an error, if necessary.

requestFromDynamic :: forall r a. (HasCallStack, Typeable a, Member (Exc UnhandledRequest) r) => Dynamic -> Eff r a

Cast a Dynamic value, and if that fails, throw an UnhandledRequest error.

exitUnhandled :: forall r q. (SetMember Process (Process q) r, HasCallStack) => SchedulerProxy q -> Dynamic -> Eff r ()

If an incoming message could not be casted to a Request corresponding to an ApiHandler (e.g. with requestFromDynamic) one should use this function to exit the process with a corresponding error.

Observer Functions for Events and Event Listener

data CallbackObserver o

An Observer that schedules the observations to an effectful callback.

Instances

Show (Observation o) => Show (Api (CallbackObserver o) r)

Defined in Control.Eff.Concurrent.Api.Observer

Methods

showsPrec :: Int -> Api (CallbackObserver o) r -> ShowS

show :: Api (CallbackObserver o) r -> String

showList :: [Api (CallbackObserver o) r] -> ShowS

Observable o => Observer (CallbackObserver o) o

Defined in Control.Eff.Concurrent.Api.Observer

Methods

observationMessage :: Server o -> Observation o -> Api (CallbackObserver o) Asynchronous

data Api (CallbackObserver o) r

Defined in Control.Eff.Concurrent.Api.Observer

data Api (CallbackObserver o) r where

• CbObserved :: Api (CallbackObserver o) Asynchronous

data Observers o

Internal state for manageobservers

data SomeObserver o where

An existential wrapper around a Server of an Observer. Needed to support different types of observers to observe the same Observable in a general fashion.

Constructors

SomeObserver :: (Show (Server p), Typeable p, Observer p o) => Server p -> SomeObserver o

Instances

Eq (SomeObserver o)

Defined in Control.Eff.Concurrent.Api.Observer

Methods

(==) :: SomeObserver o -> SomeObserver o -> Bool

(/=) :: SomeObserver o -> SomeObserver o -> Bool

Ord (SomeObserver o)

Defined in Control.Eff.Concurrent.Api.Observer

Methods

compare :: SomeObserver o -> SomeObserver o -> Ordering

(<) :: SomeObserver o -> SomeObserver o -> Bool

(<=) :: SomeObserver o -> SomeObserver o -> Bool

(>) :: SomeObserver o -> SomeObserver o -> Bool

(>=) :: SomeObserver o -> SomeObserver o -> Bool

max :: SomeObserver o -> SomeObserver o -> SomeObserver o

min :: SomeObserver o -> SomeObserver o -> SomeObserver o

Show (SomeObserver o)

Defined in Control.Eff.Concurrent.Api.Observer

Methods

showsPrec :: Int -> SomeObserver o -> ShowS

show :: SomeObserver o -> String

showList :: [SomeObserver o] -> ShowS

class (Typeable o, Typeable (Observation o)) => Observable o where

An Api index that supports registration and de-registration of Observers.

Minimal complete definition

registerObserverMessage, forgetObserverMessage

Associated Types

data Observation o

Type of observations visible on this observable

Methods

registerObserverMessage :: SomeObserver o -> Api o Asynchronous

Return the Api value for the cast_ that registeres an observer

forgetObserverMessage :: SomeObserver o -> Api o Asynchronous

Return the Api value for the cast_ that de-registeres an observer

class (Typeable p, Observable o) => Observer p o where

An Api index that support observation of the another Api that is Observable.

Minimal complete definition

observationMessage

Methods

observationMessage :: Server o -> Observation o -> Api p Asynchronous

Wrap the Observation and the ProcessId (i.e. the Server) that caused the observation into a Api value that the Observable understands.

Instances

Observable o => Observer (CallbackObserver o) o

Defined in Control.Eff.Concurrent.Api.Observer

Methods

observationMessage :: Server o -> Observation o -> Api (CallbackObserver o) Asynchronous

notifyObserver :: (SetMember Process (Process q) r, Observable o, Observer p o, HasCallStack) => SchedulerProxy q -> Server p -> Server o -> Observation o -> Eff r ()

Send an Observation to an Observer

registerObserver :: (SetMember Process (Process q) r, Observable o, Observer p o, HasCallStack) => SchedulerProxy q -> Server p -> Server o -> Eff r ()

Send the registerObserverMessage

forgetObserver :: (SetMember Process (Process q) r, Observable o, Observer p o) => SchedulerProxy q -> Server p -> Server o -> Eff r ()

Send the forgetObserverMessage

notifySomeObserver :: (SetMember Process (Process q) r, Observable o, HasCallStack) => SchedulerProxy q -> Server o -> Observation o -> SomeObserver o -> Eff r ()

Send an Observation to SomeObserver.

manageObservers :: Eff (State (Observers o) ': r) a -> Eff r a

Keep track of registered Observers Observers can be added and removed, and an Observation can be sent to all registerd observers at once.

addObserver :: (SetMember Process (Process q) r, Member (State (Observers o)) r, Observable o) => SomeObserver o -> Eff r ()

Add an Observer to the Observers managed by manageObservers.

removeObserver :: (SetMember Process (Process q) r, Member (State (Observers o)) r, Observable o) => SomeObserver o -> Eff r ()

Delete an Observer from the Observers managed by manageObservers.

notifyObservers :: forall o r q. (Observable o, SetMember Process (Process q) r, Member (State (Observers o)) r) => SchedulerProxy q -> Observation o -> Eff r ()

Send an Observation to all SomeObservers in the Observers state.

spawnCallbackObserver :: forall o r q. (SetMember Process (Process q) r, Typeable o, Show (Observation o), Observable o, Member (Logs LogMessage) q) => SchedulerProxy q -> (Server o -> Observation o -> Eff (Process q ': q) Bool) -> Eff r (Server (CallbackObserver o))

Start a new process for an Observer that schedules all observations to an effectful callback.

spawnLoggingObserver :: forall o r q. (SetMember Process (Process q) r, Typeable o, Show (Observation o), Observable o, Member (Logs LogMessage) q) => SchedulerProxy q -> Eff r (Server (CallbackObserver o))

Use spawnCallbackObserver to create a universal logging observer, using the Show instance of the Observation. | Start a new process for an Observer that schedules all observations to an effectful callback.

Since: extensible-effects-concurrent-0.3.0.0

Scheduler Process Effect Handler

Concurrent Scheduler

type SchedulerIO = '[Reader SchedulerVar, Logs LogMessage, Lift IO]

The concrete list of Effects for this scheduler implementation. See HasSchedulerIO

type HasSchedulerIO r = (HasCallStack, SetMember Lift (Lift IO) r, Member (Logs LogMessage) r, Member (Reader SchedulerVar) r)

An alias for the constraints for the effects essential to this scheduler implementation, i.e. these effects allow spawning new Processes. See SchedulerIO

data SchedulerError

A sum-type with errors that can occur when scheduleing messages.

Constructors

ProcessNotFound ProcessId

No process info was found for a ProcessId during internal processing. NOTE: This is **ONLY** caused by internal errors, probably by an incorrect MessagePassing handler in this module. **Sending a message to a process ALWAYS succeeds!** Even if the process does not exist.

ProcessRaisedError String

A process called raiseError.

ProcessExitError String

A process called exitWithError.

ProcessShuttingDown

A process exits.

SchedulerShuttingDown

An action was not performed while the scheduler was exiting.

Instances

Show SchedulerError

Defined in Control.Eff.Concurrent.Process.ForkIOScheduler

Methods

showsPrec :: Int -> SchedulerError -> ShowS

show :: SchedulerError -> String

showList :: [SchedulerError] -> ShowS

Exception SchedulerError

Defined in Control.Eff.Concurrent.Process.ForkIOScheduler

Methods

toException :: SchedulerError -> SomeException

fromException :: SomeException -> Maybe SchedulerError

displayException :: SchedulerError -> String

forkIoScheduler :: SchedulerProxy SchedulerIO

A SchedulerProxy for SchedulerIO

schedule :: Eff (ConsProcess SchedulerIO) () -> LogChannel LogMessage -> IO ()

This is the main entry point to running a message passing concurrency application. This function takes a Process on top of the SchedulerIO effect and a LogChannel for concurrent logging.

defaultMain :: Eff (ConsProcess SchedulerIO) () -> IO ()

Start the message passing concurrency system then execute a Process on top of SchedulerIO effect. All logging is sent to standard output.

defaultMainWithLogChannel :: LogChannel LogMessage -> Eff (ConsProcess SchedulerIO) () -> IO ()

Start the message passing concurrency system then execute a Process on top of SchedulerIO effect. All logging is sent to standard output.

Single Threaded Scheduler

schedulePure :: Eff (ConsProcess '[]) a -> Either String a

Like schedule but pure. The yield effect is just return (). schedulePure == runIdentity . scheduleM (Identity . run) (return ())

Since: extensible-effects-concurrent-0.3.0.2

Utilities

Logging Effect

module Control.Eff.Log

Preventing Space Leaks

module Control.Eff.Loop