path: root/stdlib/source/lux/control/concurrency/actor.lux

(.module: {#.doc "The actor model of concurrency."}
  [lux #*
   [abstract
    monad]
   [control
    [pipe (#+ case>)]
    ["." function]
    ["." try (#+ Try)]
    ["." exception (#+ exception:)]
    ["." io (#+ IO io)]
    ["<>" parser
     ["<c>" code (#+ Parser)]]]
   [data
    ["." product]
    [number
     ["n" nat]]
    [text
     ["%" format (#+ format)]]
    [collection
     ["." list ("#\." monoid monad fold)]]]
   [macro
    ["." code]
    [syntax (#+ syntax:)
     ["cs" common
      ["csr" reader]
      ["csw" writer]]]]
   ["." meta (#+ with-gensyms monad)
    ["." annotation]]
   [type (#+ :share)
    ["." abstract (#+ abstract: :representation :abstraction)]]]
  [//
   ["." atom (#+ Atom atom)]
   ["." promise (#+ Promise Resolver) ("#\." monad)]
   ["." frp (#+ Channel)]])

(exception: #export poisoned)
(exception: #export dead)

(with-expansions
  [<Mail> (as-is (-> s (Actor s) (Promise (Try s))))
   <Obituary> (as-is [Text s (List <Mail>)])
   <Mailbox> (as-is (Rec Mailbox
                      [(Promise [<Mail> Mailbox])
                       (Resolver [<Mail> Mailbox])]))]

  (def: (pending [read write])
    (All [a]
      (-> (Rec Mailbox
            [(Promise [a Mailbox])
             (Resolver [a Mailbox])])
          (IO (List a))))
    (do {! io.monad}
      [current (promise.poll read)]
      (case current
        (#.Some [head tail])
        (\ ! map (|>> (#.Cons head))
           (pending tail))
        
        #.None
        (wrap #.Nil))))
  
  (abstract: #export (Actor s)
    {#obituary [(Promise <Obituary>)
                (Resolver <Obituary>)]
     #mailbox (Atom <Mailbox>)}

    (type: #export (Mail s)
      <Mail>)

    (type: #export (Obituary s)
      <Obituary>)

    (type: #export (Behavior o s)
      {#.doc "An actor's behavior when mail is received and when a fatal error occurs."}
      {#on-init (-> o s)
       #on-mail (-> (Mail s) s (Actor s) (Promise (Try s)))
       #on-stop (-> Text s (Promise Any))})

    (def: #export (spawn! behavior init)
      {#.doc "Given a behavior and initial state, spawns an actor and returns it."}
      (All [o s] (-> (Behavior o s) o (IO (Actor s))))
      (io (let [[on-init on-mail on-stop] behavior
                self (:share [o s]
                             {(Behavior o s)
                              behavior}
                             {(Actor s)
                              (:abstraction {#obituary (promise.promise [])
                                             #mailbox (atom (promise.promise []))})})
                process (loop [state (on-init init)
                               [|mailbox| _] (io.run (atom.read (get@ #mailbox (:representation self))))]
                          (do {! promise.monad}
                            [[head tail] |mailbox|
                             ?state' (on-mail head state self)]
                            (case ?state'
                              (#try.Failure error)
                              (do !
                                [_ (on-stop error state)]
                                (let [[_ resolve] (get@ #obituary (:representation self))]
                                  (exec (io.run
                                         (do io.monad
                                           [pending (..pending tail)]
                                           (resolve [error state (#.Cons head pending)])))
                                    (wrap []))))

                              (#try.Success state')
                              (recur state' tail))))]
            self)))

    (def: #export (alive? actor)
      (All [s] (-> (Actor s) (IO Bit)))
      (let [[obituary _] (get@ #obituary (:representation actor))]
        (|> obituary
            promise.poll
            (\ io.functor map
               (|>> (case> #.None
                           yes

                           _
                           no))))))

    (def: #export (obituary actor)
      (All [s] (-> (Actor s) (IO (Maybe (Obituary s)))))
      (let [[obituary _] (get@ #obituary (:representation actor))]
        (promise.poll obituary)))

    (def: #export await
      (All [s] (-> (Actor s) (Promise (Obituary s))))
      (|>> :representation
           (get@ #obituary)
           product.left))

    (def: #export (mail! mail actor)
      {#.doc "Send mail to an actor.."}
      (All [s] (-> (Mail s) (Actor s) (IO (Try Any))))
      (do {! io.monad}
        [alive? (..alive? actor)]
        (if alive?
          (let [entry [mail (promise.promise [])]]
            (do !
              [|mailbox|&resolve (atom.read (get@ #mailbox (:representation actor)))]
              (loop [[|mailbox| resolve] |mailbox|&resolve]
                (do !
                  [|mailbox| (promise.poll |mailbox|)]
                  (case |mailbox|
                    #.None
                    (do !
                      [resolved? (resolve entry)]
                      (if resolved?
                        (do !
                          [_ (atom.write (product.right entry) (get@ #mailbox (:representation actor)))]
                          (wrap (exception.return [])))
                        (recur |mailbox|&resolve)))
                    
                    (#.Some [_ |mailbox|'])
                    (recur |mailbox|'))))))
          (wrap (exception.throw ..dead [])))))

    (type: #export (Message s o)
      (-> s (Actor s) (Promise (Try [s o]))))

    (def: (mail message)
      (All [s o] (-> (Message s o) [(Promise (Try o)) (Mail s)]))
      (let [[promise resolve] (:share [s o]
                                      {(Message s o)
                                       message}
                                      {[(Promise (Try o))
                                        (Resolver (Try o))]
                                       (promise.promise [])})]
        [promise
         (function (_ state self)
           (do {! promise.monad}
             [outcome (message state self)]
             (case outcome
               (#try.Success [state' return])
               (exec (io.run (resolve (#try.Success return)))
                 (promise.resolved (#try.Success state')))
               
               (#try.Failure error)
               (exec (io.run (resolve (#try.Failure error)))
                 (promise.resolved (#try.Failure error))))))]))

    (def: #export (tell! message actor)
      {#.doc "Communicate with an actor through message passing."}
      (All [s o] (-> (Message s o) (Actor s) (Promise (Try o))))
      (let [[promise mail] (..mail message)]
        (do promise.monad
          [outcome (promise.future (..mail! mail actor))]
          (case outcome
            (#try.Success)
            promise
            
            (#try.Failure error)
            (wrap (#try.Failure error))))))
    )
  )

(def: (default-on-mail mail state self)
  (All [s] (-> (Mail s) s (Actor s) (Promise (Try s))))
  (mail state self))

(def: (default-on-stop cause state)
  (All [s] (-> Text s (Promise Any)))
  (promise\wrap []))

(def: #export default
  (All [s] (Behavior s s))
  {#on-init function.identity
   #on-mail ..default-on-mail
   #on-stop ..default-on-stop})

(def: #export (poison! actor)
  {#.doc (doc "Kills the actor by sending mail that will kill it upon processing,"
              "but allows the actor to handle previous mail.")}
  (All [s] (-> (Actor s) (IO (Try Any))))
  (..mail! (function (_ state self)
             (promise.resolved (exception.throw ..poisoned [])))
           actor))

(def: actor-decl^
  (Parser [Text (List Text)])
  (<>.either (<c>.form (<>.and <c>.local-identifier (<>.some <c>.local-identifier)))
             (<>.and <c>.local-identifier (\ <>.monad wrap (list)))))

(type: On-MailC
  [[Text Text Text] Code])

(type: On-StopC
  [[Text Text] Code])

(type: BehaviorC
  [(Maybe On-MailC) (Maybe On-StopC) (List Code)])

(def: argument
  (Parser Text)
  <c>.local-identifier)

(def: behavior^
  (Parser BehaviorC)
  (let [on-mail-args ($_ <>.and ..argument ..argument ..argument)
        on-stop-args ($_ <>.and ..argument ..argument)]
    ($_ <>.and
        (<>.maybe (<c>.form (<>.and (<c>.form (<>.after (<c>.this! (' on-mail)) on-mail-args))
                                    <c>.any)))
        (<>.maybe (<c>.form (<>.and (<c>.form (<>.after (<c>.this! (' on-stop)) on-stop-args))
                                    <c>.any)))
        (<>.some <c>.any))))

(def: (on-mail g!_ ?on-mail)
  (-> Code (Maybe On-MailC) Code)
  (case ?on-mail
    #.None
    (` (~! ..default-on-mail))

    (#.Some [[mailN stateN selfN] bodyC])
    (` (function ((~ g!_)
                  (~ (code.local-identifier mailN))
                  (~ (code.local-identifier stateN))
                  (~ (code.local-identifier selfN)))
         (~ bodyC)))))

(def: (on-stop g!_ ?on-stop)
  (-> Code (Maybe On-StopC) Code)
  (case ?on-stop
    #.None
    (` (~! ..default-on-stop))

    (#.Some [[causeN stateN] bodyC])
    (` (function ((~ g!_)
                  (~ (code.local-identifier causeN))
                  (~ (code.local-identifier stateN)))
         (~ bodyC)))))

(with-expansions [<examples> (as-is (actor: #export (Stack a)
                                      (List a)

                                      ((on-mail mail state self)
                                       (do (try.with promise.monad)
                                         [#let [_ (log! "BEFORE")]
                                          output (mail state self)
                                          #let [_ (log! "AFTER")]]
                                         (wrap output)))

                                      (message: #export (push {value a} state self (List a))
                                        (let [state' (#.Cons value state)]
                                          (promise.resolved (#try.Success [state' state'])))))

                                    (actor: #export Counter
                                      Nat

                                      ((on-stop cause state)
                                       (\ promise.monad wrap
                                          (log! (if (exception.match? ..poisoned cause)
                                                  (format "Counter was poisoned: " (%.nat state))
                                                  cause))))

                                      (message: #export (count! {increment Nat} state self Any)
                                        (let [state' (n.+ increment state)]
                                          (promise.resolved (#try.Success [state' state']))))

                                      (message: #export (read! state self Nat)
                                        (promise.resolved (#try.Success [state state])))))]
  (syntax: #export (actor:
                     {export csr.export}
                     {[name vars] actor-decl^}
                     {annotations (<>.default cs.empty-annotations csr.annotations)}
                     state-type
                     {[?on-mail ?on-stop messages] behavior^})
    {#.doc (doc "Defines an actor, with its behavior and internal state."
                "Messages for the actor must be defined after the on-mail and on-stop handlers."
                <examples>)}
    (with-gensyms [g!_]
      (do meta.monad
        [g!type (meta.gensym (format name "-abstract-type"))
         #let [g!actor (code.local-identifier name)
               g!vars (list\map code.local-identifier vars)]]
        (wrap (list (` ((~! abstract:) (~+ (csw.export export)) ((~ g!type) (~+ g!vars))
                        (~ state-type)

                        (def: (~+ (csw.export export)) (~ g!actor)
                          (All [(~+ g!vars)]
                            (..Behavior (~ state-type) ((~ g!type) (~+ g!vars))))
                          {#..on-init (|>> ((~! abstract.:abstraction) (~ g!type)))
                           #..on-mail (~ (..on-mail g!_ ?on-mail))
                           #..on-stop (~ (..on-stop g!_ ?on-stop))})

                        (~+ messages))))))))

  (syntax: #export (actor {[state-type init] (<c>.record (<>.and <c>.any <c>.any))}
                          {[?on-mail ?on-stop messages] behavior^})
    (with-gensyms [g!_]
      (wrap (list (` (: ((~! io.IO) (..Actor (~ state-type)))
                        (..spawn! (: (..Behavior (~ state-type) (~ state-type))
                                     {#..on-init (|>>)
                                      #..on-mail (~ (..on-mail g!_ ?on-mail))
                                      #..on-stop (~ (..on-stop g!_ ?on-stop))})
                                  (: (~ state-type)
                                     (~ init)))))))))

  (type: Signature
    {#vars (List Text)
     #name Text
     #inputs (List cs.Typed-Input)
     #state Text
     #self Text
     #output Code})

  (def: signature^
    (Parser Signature)
    (<c>.form ($_ <>.and
                  (<>.default (list) (<c>.tuple (<>.some <c>.local-identifier)))
                  <c>.local-identifier
                  (<>.some csr.typed-input)
                  <c>.local-identifier
                  <c>.local-identifier
                  <c>.any)))

  (def: reference^
    (Parser [Name (List Text)])
    (<>.either (<c>.form (<>.and <c>.identifier (<>.some <c>.local-identifier)))
               (<>.and <c>.identifier (\ <>.monad wrap (list)))))

  (syntax: #export (message:
                     {export csr.export}
                     {signature signature^}
                     {annotations (<>.default cs.empty-annotations csr.annotations)}
                     body)
    {#.doc (doc "A message can access the actor's state through the state parameter."
                "A message can also access the actor itself through the self parameter."
                "A message's output must be a promise containing a 2-tuple with the updated state and a return value."
                "A message may succeed or fail (in case of failure, the actor dies)."

                <examples>)}
    (with-gensyms [g!_ g!return]
      (do meta.monad
        [actor-scope abstract.current
         #let [g!type (code.local-identifier (get@ #abstract.name actor-scope))
               g!message (code.local-identifier (get@ #name signature))
               g!actor-vars (get@ #abstract.type-vars actor-scope)
               g!all-vars (|> (get@ #vars signature) (list\map code.local-identifier) (list\compose g!actor-vars))
               g!inputsC (|> (get@ #inputs signature) (list\map product.left))
               g!inputsT (|> (get@ #inputs signature) (list\map product.right))
               g!state (|> signature (get@ #state) code.local-identifier)
               g!self (|> signature (get@ #self) code.local-identifier)]]
        (wrap (list (` (def: (~+ (csw.export export)) ((~ g!message) (~+ g!inputsC))
                         (~ (csw.annotations annotations))
                         (All [(~+ g!all-vars)]
                           (-> (~+ g!inputsT)
                               (..Message (~ (get@ #abstract.abstraction actor-scope))
                                          (~ (get@ #output signature)))))
                         (function ((~ g!_) (~ g!state) (~ g!self))
                           (let [(~ g!state) (:coerce (~ (get@ #abstract.representation actor-scope))
                                                      (~ g!state))]
                             (|> (~ body)
                                 (: ((~! promise.Promise) ((~! try.Try) [(~ (get@ #abstract.representation actor-scope))
                                                                         (~ (get@ #output signature))])))
                                 (:coerce ((~! promise.Promise) ((~! try.Try) [(~ (get@ #abstract.abstraction actor-scope))
                                                                               (~ (get@ #output signature))]))))))))
                    ))))))

(type: #export Stop
  (IO Any))

(def: continue! true)
(def: stop! false)

(def: #export (observe action channel actor)
  (All [e s] (-> (-> e Stop (Mail s)) (Channel e) (Actor s) (IO Any)))
  (let [signal (: (Atom Bit)
                  (atom.atom ..continue!))
        stop (: Stop
                (atom.write ..stop! signal))]
    (frp.subscribe (function (_ event)
                     (do {! io.monad}
                       [continue? (atom.read signal)]
                       (if continue?
                         (do !
                           [outcome (..mail! (action event stop) actor)]
                           (wrap (try.to-maybe outcome)))
                         (wrap #.None))))
                   channel)))