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

(.module:
  {#.doc "The actor model of concurrency."}
  [library
   [lux #*
    ["." debug]
    [abstract
     monad]
    [control
     [pipe (#+ case>)]
     ["." function]
     ["." try (#+ Try)]
     ["." exception (#+ exception:)]
     ["." io (#+ IO io)]
     ["<>" parser ("#\." monad)
      ["<.>" code (#+ Parser)]]]
    [data
     ["." bit]
     ["." product]
     [text
      ["%" format (#+ format)]]
     [collection
      ["." list ("#\." monoid monad fold)]]]
    ["." macro (#+ with_gensyms)
     ["." code]
     [syntax (#+ syntax:)
      ["|.|" input]
      ["|.|" annotations]]]
    [math
     [number
      ["n" nat]]]
    ["." meta (#+ monad)
     ["." annotation]]
    [type (#+ :sharing)
     ["." abstract (#+ abstract: :representation :abstraction)]]]]
  [//
   ["." atom (#+ Atom atom)]
   ["." async (#+ Async Resolver) ("#\." monad)]
   ["." frp (#+ Channel)]])

(exception: .public poisoned)
(exception: .public dead)

(with_expansions
  [<Mail> (as_is (-> s (Actor s) (Async (Try s))))
   <Obituary> (as_is [Text s (List <Mail>)])
   <Mailbox> (as_is (Rec Mailbox
                      [(Async [<Mail> Mailbox])
                       (Resolver [<Mail> Mailbox])]))]

  (def: (pending [read write])
    (All [a]
      (-> (Rec Mailbox
            [(Async [a Mailbox])
             (Resolver [a Mailbox])])
          (IO (List a))))
    (do {! io.monad}
      [current (async.poll read)]
      (case current
        (#.Some [head tail])
        (\ ! map (|>> (#.Item head))
           (pending tail))
        
        #.None
        (in #.End))))
  
  (abstract: .public (Actor s)
    {#.doc (doc "An entity that can react to messages (mail) sent to it concurrently.")}

    {#obituary [(Async <Obituary>)
                (Resolver <Obituary>)]
     #mailbox (Atom <Mailbox>)}

    (type: .public (Mail s)
      {#.doc (doc "A one-way message sent to an actor, without expecting a reply.")}
      <Mail>)

    (type: .public (Obituary s)
      {#.doc (doc "Details on the death of an actor.")}
      <Obituary>)

    (type: .public (Behavior o s)
      {#.doc (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) (Async (Try s)))})

    (def: .public (spawn! behavior init)
      {#.doc (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] behavior
                self (:sharing [o s]
                               (Behavior o s)
                               behavior
                               
                               (Actor s)
                               (:abstraction {#obituary (async.async [])
                                              #mailbox (atom (async.async []))}))
                process (loop [state (on_init init)
                               [|mailbox| _] (io.run (atom.read (get@ #mailbox (:representation self))))]
                          (do {! async.monad}
                            [[head tail] |mailbox|
                             ?state' (on_mail head state self)]
                            (case ?state'
                              (#try.Failure error)
                              (let [[_ resolve] (get@ #obituary (:representation self))]
                                (exec (io.run
                                       (do io.monad
                                         [pending (..pending tail)]
                                         (resolve [error state (#.Item head pending)])))
                                  (in [])))

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

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

                           _
                           bit.no))))))

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

    (def: .public await
      {#.doc (doc "Await for an actor to end working.")}
      (All [s] (-> (Actor s) (Async (Obituary s))))
      (|>> :representation
           (get@ #obituary)
           product.left))

    (def: .public (mail! mail actor)
      {#.doc (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 (async.async [])]]
            (do !
              [|mailbox|&resolve (atom.read (get@ #mailbox (:representation actor)))]
              (loop [[|mailbox| resolve] |mailbox|&resolve]
                (do !
                  [|mailbox| (async.poll |mailbox|)]
                  (case |mailbox|
                    #.None
                    (do !
                      [resolved? (resolve entry)]
                      (if resolved?
                        (do !
                          [_ (atom.write (product.right entry) (get@ #mailbox (:representation actor)))]
                          (in (exception.return [])))
                        (recur |mailbox|&resolve)))
                    
                    (#.Some [_ |mailbox|'])
                    (recur |mailbox|'))))))
          (in (exception.except ..dead [])))))

    (type: .public (Message s o)
      {#.doc (doc "A two-way message sent to an actor, expecting a reply.")}
      (-> s (Actor s) (Async (Try [s o]))))

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

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

(def: (default_on_mail mail state self)
  (All [s] (-> (Mail s) s (Actor s) (Async (Try s))))
  (mail state self))

(def: .public default
  {#.doc (doc "Default actor behavior.")}
  (All [s] (Behavior s s))
  {#on_init function.identity
   #on_mail ..default_on_mail})

(def: .public (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)
             (async.resolved (exception.except ..poisoned [])))
           actor))

(def: actor_decl^
  (Parser [Text (List Text)])
  (<>.either (<code>.form (<>.and <code>.local_identifier (<>.some <code>.local_identifier)))
             (<>.and <code>.local_identifier (\ <>.monad in (list)))))

(type: On_MailC
  [[Text Text Text] Code])

(type: BehaviorC
  [(Maybe On_MailC) (List Code)])

(def: argument
  (Parser Text)
  <code>.local_identifier)

(def: behavior^
  (Parser BehaviorC)
  (let [on_mail_args ($_ <>.and ..argument ..argument ..argument)]
    ($_ <>.and
        (<>.maybe (<code>.form (<>.and (<code>.form (<>.after (<code>.this! (' on_mail)) on_mail_args))
                                       <code>.any)))
        (<>.some <code>.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: actorP
  (Parser [Code [Text (List Text)] |annotations|.Annotations Code BehaviorC])
  (let [private ($_ <>.and
                    ..actor_decl^
                    |annotations|.parser
                    <code>.any
                    behavior^)]
    ($_ <>.either
        (<>.and <code>.any private)
        (<>.and (<>\in (` .private)) private))))

(with_expansions [<examples> (as_is (actor: .public (stack a)
                                      {}

                                      (List a)

                                      ((on_mail mail state self)
                                       (do (try.with async.monad)
                                         [.let [_ (debug.log! "BEFORE")]
                                          output (mail state self)
                                          .let [_ (debug.log! "AFTER")]]
                                         (in output)))

                                      (message: .public (push {value a} state self)
                                        (List a)
                                        (let [state' (#.Item value state)]
                                          (async.resolved (#try.Success [state' state'])))))

                                    (actor: .public counter
                                      {}

                                      Nat

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

                                      (message: .public (read! state self)
                                        Nat
                                        (async.resolved (#try.Success [state state])))))]
  (syntax: .public (actor:
                     {[export_policy [name vars] annotations state_type [?on_mail messages]] ..actorP})
    {#.doc (doc "Defines a named actor, with its behavior and internal state."
                "Messages for the actor must be defined after the on_mail handler."
                <examples>)}
    (with_gensyms [g!_]
      (do meta.monad
        [g!type (macro.gensym (format name "_abstract_type"))
         .let [g!actor (code.local_identifier name)
               g!vars (list\map code.local_identifier vars)]]
        (in (list (` ((~! abstract:) (~ export_policy) ((~ g!type) (~+ g!vars))
                      {}

                      (~ state_type)

                      (def: (~ export_policy) (~ 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))})

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

  (syntax: .public (actor {[state_type init] (<code>.record (<>.and <code>.any <code>.any))}
                          {[?on_mail messages] behavior^})
    {#.doc (doc "Defines an anonymous actor, with its behavior and internal state."
                "Messages for the actor must be defined after the on_mail handler."
                (actor {Nat
                        123}
                       ((on_mail message state self)
                        (message (inc state) self))))}
    (with_gensyms [g!_]
      (in (list (` (: ((~! io.IO) (..Actor (~ state_type)))
                      (..spawn! (: (..Behavior (~ state_type) (~ state_type))
                                   {#..on_init (|>>)
                                    #..on_mail (~ (..on_mail g!_ ?on_mail))})
                                (: (~ state_type)
                                   (~ init)))))))))

  (type: Signature
    {#vars (List Text)
     #name Text
     #inputs (List |input|.Input)
     #state Text
     #self Text})

  (def: signature^
    (Parser Signature)
    (<code>.form ($_ <>.and
                     (<>.else (list) (<code>.tuple (<>.some <code>.local_identifier)))
                     <code>.local_identifier
                     (<>.some |input|.parser)
                     <code>.local_identifier
                     <code>.local_identifier)))

  (def: reference^
    (Parser [Name (List Text)])
    (<>.either (<code>.form (<>.and <code>.identifier (<>.some <code>.local_identifier)))
               (<>.and <code>.identifier (\ <>.monad in (list)))))

  (def: messageP
    (Parser [Code Signature |annotations|.Annotations Code Code])
    (let [private ($_ <>.and
                      ..signature^
                      (<>.else |annotations|.empty |annotations|.parser)
                      <code>.any
                      <code>.any)]
      ($_ <>.either
          (<>.and <code>.any private)
          (<>.and (<>\in (` .private)) private))))

  (syntax: .public (message:
                     {[export_policy signature annotations output_type body] ..messageP})
    {#.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 an async 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 (|> signature (get@ #vars) (list\map code.local_identifier) (list\compose g!actor_vars))
               g!inputsC (|> signature (get@ #inputs) (list\map product.left))
               g!inputsT (|> signature (get@ #inputs) (list\map product.right))
               g!state (|> signature (get@ #state) code.local_identifier)
               g!self (|> signature (get@ #self) code.local_identifier)]]
        (in (list (` (def: (~ export_policy) ((~ g!message) (~+ g!inputsC))
                       (~ (|annotations|.format annotations))
                       (All [(~+ g!all_vars)]
                         (-> (~+ g!inputsT)
                             (..Message (~ (get@ #abstract.abstraction actor_scope))
                                        (~ output_type))))
                       (function ((~ g!_) (~ g!state) (~ g!self))
                         (let [(~ g!state) (:as (~ (get@ #abstract.representation actor_scope))
                                                (~ g!state))]
                           (|> (~ body)
                               (: ((~! async.Async) ((~! try.Try) [(~ (get@ #abstract.representation actor_scope))
                                                                   (~ output_type)])))
                               (:as ((~! async.Async) ((~! try.Try) [(~ (get@ #abstract.abstraction actor_scope))
                                                                     (~ output_type)]))))))))
                  ))))))

(type: .public Stop
  {#.doc (doc "A signal to stop an actor from observing a channel.")}
  (IO Any))

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

(def: .public (observe action channel actor)
  {#.doc (doc "Use an actor to observe a channel by transforming each datum"
              "flowing through the channel into mail the actor can process."
              "Can stop observing the channel by executing the Stop value.")}
  (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?
                         (|> actor
                             (..mail! (action event stop))
                             (\ ! map try.maybe))
                         (in #.None))))
                   channel)))