Actor Types

structure Lapis.Concurrent.Actor.ActorRef (Msg : Type) : Type

A handle to send messages to an actor. Can be freely cloned and shared across threads.

• mailbox : Channel.Unbounded Msg

def Lapis.Concurrent.Actor.ActorRef.send {Msg : Type} (ref : ActorRef Msg) (msg : Msg) : IO Unit

Send a message to the actor

Equations

• ref.send msg = ref.mailbox.send msg

def Lapis.Concurrent.Actor.ActorRef.call {Resp Msg : Type} [Nonempty Resp] (ref : ActorRef Msg) (mkMsg : Channel.Oneshot Resp → Msg) : IO Resp

Send a message and get a response via oneshot channel

Equations

• ref.call mkMsg = do let reply ← Lapis.Concurrent.Channel.Oneshot.new ref.send (mkMsg reply) reply.recv

def Lapis.Concurrent.Actor.ActorRef.trySend {Msg : Type} (ref : ActorRef Msg) (msg : Msg) : IO Bool

Try to send without blocking (always succeeds for unbounded)

Equations

• ref.trySend msg = do ref.mailbox.send msg pure true

inductive Lapis.Concurrent.Actor.ActorStatus : Type

Actor lifecycle state

• running : ActorStatus
• stopping : ActorStatus
• stopped : ActorStatus

def Lapis.Concurrent.Actor.instBEqActorStatus.beq : ActorStatus → ActorStatus → Bool

Equations

• Lapis.Concurrent.Actor.instBEqActorStatus.beq x✝ y✝ = (x✝.ctorIdx == y✝.ctorIdx)

instance Lapis.Concurrent.Actor.instBEqActorStatus : BEq ActorStatus

Equations

• Lapis.Concurrent.Actor.instBEqActorStatus = { beq := Lapis.Concurrent.Actor.instBEqActorStatus.beq }

instance Lapis.Concurrent.Actor.instReprActorStatus : Repr ActorStatus

Equations

• Lapis.Concurrent.Actor.instReprActorStatus = { reprPrec := Lapis.Concurrent.Actor.instReprActorStatus.repr }

def Lapis.Concurrent.Actor.instReprActorStatus.repr : ActorStatus → Nat → Std.Format

Equations

• One or more equations did not get rendered due to their size.

structure Lapis.Concurrent.Actor.ActorConfig : Type

Configuration for actor behavior

• name : String

  Name for debugging/logging

• restartOnPanic : Bool

  Whether to restart on handler panic

structure Lapis.Concurrent.Actor.Actor (Msg State : Type) : Type

A running actor instance

• ref : ActorRef Msg

  Reference to send messages

• task : Task (Except IO.Error Unit)

  The running task

• status : IO.Ref ActorStatus

  Actor status

• config : ActorConfig

  Configuration

def Lapis.Concurrent.Actor.Actor.isRunning {Msg State : Type} (actor : Actor Msg State) : IO Bool

Check if the actor is still running

Equations

• actor.isRunning = do let status ← ST.Ref.get actor.status pure (status == Lapis.Concurrent.Actor.ActorStatus.running)

def Lapis.Concurrent.Actor.Actor.stop {Msg State : Type} (actor : Actor Msg State) : IO Unit

Request the actor to stop (gracefully)

Equations

• actor.stop = ST.Ref.set actor.status Lapis.Concurrent.Actor.ActorStatus.stopping

def Lapis.Concurrent.Actor.Actor.join {Msg State : Type} (actor : Actor Msg State) : IO Unit

Wait for the actor to finish

Equations

• actor.join = do let __discr ← liftM (IO.wait actor.task) have x : Except IO.Error Unit := __discr pure PUnit.unit

def Lapis.Concurrent.Actor.Actor.shutdown {Msg State : Type} (actor : Actor Msg State) : IO Unit

Stop and wait for the actor to finish

Equations

• actor.shutdown = do actor.stop actor.join

def Lapis.Concurrent.Actor.Actor.getRef {Msg State : Type} (actor : Actor Msg State) : ActorRef Msg

Get the actor's reference for sending messages

Equations

• actor.getRef = actor.ref

Actor Spawning

inductive Lapis.Concurrent.Actor.HandleResult (State : Type) : Type

Message handler result

• continue {State : Type} (state : State) : HandleResult State

  Continue with new state

• stop {State : Type} : HandleResult State

  Stop the actor

• error {State : Type} (msg : String) : HandleResult State

  Stop with error

def Lapis.Concurrent.Actor.spawn {State Msg : Type} (initialState : State) (handler : State → Msg → IO (HandleResult State)) (config : ActorConfig := { }) : IO (Actor Msg State)

Spawn a new actor with the given initial state and message handler. The handler is called sequentially for each message.

Equations

• One or more equations did not get rendered due to their size.

def Lapis.Concurrent.Actor.spawnWithErrorHandler {State Msg : Type} (initialState : State) (handler : State → Msg → IO (HandleResult State)) (onError : String → IO Unit) (config : ActorConfig := { }) : IO (Actor Msg State)

Spawn an actor that can handle IO errors gracefully

Equations

• One or more equations did not get rendered due to their size.

Typed Request-Response Pattern

structure Lapis.Concurrent.Actor.Request (Req Resp : Type) : Type

A request message that expects a response

• payload : Req
• replyTo : Channel.Oneshot Resp

def Lapis.Concurrent.Actor.Request.new {Resp Req : Type} [Nonempty Resp] (payload : Req) : IO (Request Req Resp × Channel.Oneshot Resp)

Create a request and get the response channel

Equations

• Lapis.Concurrent.Actor.Request.new payload = do let reply ← Lapis.Concurrent.Channel.Oneshot.new pure ({ payload := payload, replyTo := reply }, reply)

def Lapis.Concurrent.Actor.Request.reply {Req Resp : Type} (req : Request Req Resp) (response : Resp) : IO Unit

Reply to a request

Equations

• req.reply response = req.replyTo.send response

def Lapis.Concurrent.Actor.ask {Resp Msg Req : Type} [Nonempty Resp] (ref : ActorRef Msg) (mkRequest : Request Req Resp → Msg) (payload : Req) : IO Resp

Send a request to an actor and wait for response

Equations

• One or more equations did not get rendered due to their size.

Actor Supervision

structure Lapis.Concurrent.Actor.Supervisor : Type

A supervisor that manages multiple actors

• actors : IO.Ref (Array (Task (Except IO.Error Unit)))

  All managed actors (as tasks)

• status : IO.Ref ActorStatus

  Supervisor status

def Lapis.Concurrent.Actor.Supervisor.new : IO Supervisor

Create a new supervisor

Equations

• Lapis.Concurrent.Actor.Supervisor.new = do let actors ← liftM (IO.mkRef #[]) let status ← liftM (IO.mkRef Lapis.Concurrent.Actor.ActorStatus.running) pure { actors := actors, status := status }

def Lapis.Concurrent.Actor.Supervisor.supervise {Msg State : Type} (sup : Supervisor) (actor : Actor Msg State) : IO Unit

Add an actor to supervision

Equations

• sup.supervise actor = ST.Ref.modify sup.actors fun (x : Array (Task (Except IO.Error Unit))) => x.push actor.task

def Lapis.Concurrent.Actor.Supervisor.shutdownAll (sup : Supervisor) : IO Unit

Shutdown all supervised actors

Equations

• One or more equations did not get rendered due to their size.

Utility Functions

def Lapis.Concurrent.Actor.refFromMailbox {Msg : Type} (mailbox : Channel.Unbounded Msg) : ActorRef Msg

Create an actor ref from an existing mailbox

Equations

• Lapis.Concurrent.Actor.refFromMailbox mailbox = { mailbox := mailbox }

def Lapis.Concurrent.Actor.broadcast {Msg : Type} (refs : Array (ActorRef Msg)) (msg : Msg) : IO Unit

Broadcast a message to multiple actors

Equations

• One or more equations did not get rendered due to their size.