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

(.module:
  [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_identifiers)
     ["." 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.value read)]
      (case current
        (#.Some [head tail])
        (\ ! map (|>> (#.Item head))
           (pending tail))
        
        #.None
        (in #.End))))
  
  (abstract: .public (Actor s)
    {}

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

    (type: .public (Mail s)
      <Mail>)

    (type: .public (Obituary s)
      <Obituary>)

    (type: .public (Behavior o s)
      {#on_init (-> o s)
       #on_mail (-> (Mail s) s (Actor s) (Async (Try s)))})

    (def: .public (spawn! behavior init)
      (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! (value@ #mailbox (:representation self))))]
                          (do {! async.monad}
                            [[head tail] |mailbox|
                             ?state' (on_mail head state self)]
                            (case ?state'
                              (#try.Failure error)
                              (let [[_ resolve] (value@ #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 _] (value@ #obituary (:representation actor))]
        (|> obituary
            async.value
            (\ io.functor map
               (|>> (case> #.None
                           bit.yes

                           _
                           bit.no))))))

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

    (def: .public obituary
      (All [s] (-> (Actor s) (Async (Obituary s))))
      (|>> :representation
           (value@ #obituary)
           product.left))

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

    (type: .public (Message s o)
      (-> 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)
      (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
  (All [s] (Behavior s s))
  {#on_init function.identity
   #on_mail ..default_on_mail})

(def: .public (poison! actor)
  (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))))

(syntax: .public (actor: [[export_policy [name vars] annotations state_type [?on_mail messages]] ..actorP])
  (with_identifiers [g!_]
    (do meta.monad
      [g!type (macro.identifier (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^])
  (with_identifiers [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])
  (with_identifiers [g!_ g!return]
    (do meta.monad
      [actor_scope abstract.current
       .let [g!type (code.local_identifier (value@ #abstract.name actor_scope))
             g!message (code.local_identifier (value@ #name signature))
             g!actor_vars (value@ #abstract.type_vars actor_scope)
             g!all_vars (|> signature (value@ #vars) (list\map code.local_identifier) (list\compose g!actor_vars))
             g!inputsC (|> signature (value@ #inputs) (list\map product.left))
             g!inputsT (|> signature (value@ #inputs) (list\map product.right))
             g!state (|> signature (value@ #state) code.local_identifier)
             g!self (|> signature (value@ #self) code.local_identifier)]]
      (in (list (` (def: (~ export_policy) ((~ g!message) (~+ g!inputsC))
                     (~ (|annotations|.format annotations))
                     (All [(~+ g!all_vars)]
                       (-> (~+ g!inputsT)
                           (..Message (~ (value@ #abstract.abstraction actor_scope))
                                      (~ output_type))))
                     (function ((~ g!_) (~ g!state) (~ g!self))
                       (let [(~ g!state) (:as (~ (value@ #abstract.representation actor_scope))
                                              (~ g!state))]
                         (|> (~ body)
                             (: ((~! async.Async) ((~! try.Try) [(~ (value@ #abstract.representation actor_scope))
                                                                 (~ output_type)])))
                             (:as ((~! async.Async) ((~! try.Try) [(~ (value@ #abstract.abstraction actor_scope))
                                                                   (~ output_type)]))))))))
                )))))

(type: .public Stop
  (IO Any))

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

(def: .public (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?
                          (|> actor
                              (..mail! (action event stop))
                              (\ ! map try.maybe))
                          (in #.None))))
                    channel)))