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

(.using
 [library
  [lux (.except)
   [abstract
    [functor (.only Functor)]
    [apply (.only Apply)]
    ["[0]" monad (.only Monad do)]]
   [control
    ["[0]" io (.only IO io)]
    ["[0]" maybe]
    ["[0]" try]]
   [data
    ["[0]" product]
    [collection
     ["[0]" list]]]
   [macro
    ["^" pattern]]
   [type (.only sharing)
    [primitive (.except)]
    ["[0]" variance (.only Mutable)]]]]
 [//
  ["[0]" atom (.only Atom atom)]
  ["[0]" async (.only Async Resolver)]
  ["[0]" frp (.only Channel Sink)]])

(type: (Observer a)
  (-> a (IO Any)))

(primitive .public (Var'' a)
  (Atom [a (List (Sink a))])

  (type: .public (Var' r w)
    (Var'' (Mutable r w)))
  
  (type: .public (Var a)
    (Var'' (Mutable a a)))

  (def .public (var value)
    (All (_ a) (-> a (Var a)))
    (abstraction (atom.atom [(variance.write value) (list)])))

  (def read!
    (All (_ r w) (-> (Var' r w) r))
    (|>> representation atom.read! io.run! product.left variance.read))

  (def (write! new_value var)
    (All (_ r w) (-> w (Var' r w) (IO Any)))
    (do [! io.monad]
      [.let [var' (representation var)]
       (^.let old [_ observers]) (atom.read! var')
       succeeded? (atom.compare_and_swap! old [(variance.write new_value) observers] var')]
      (if succeeded?
        (do !
          [banned (monad.only ! (function (_ sink)
                                  (do !
                                    [result (at sink feed (variance.write new_value))]
                                    (in (case result
                                          {try.#Success _}
                                          false
                                          
                                          {try.#Failure _}
                                          true))))
                              observers)
           _ (atom.update! (function (_ [value audience])
                             (|> audience
                                 (list.only (function (_ it)
                                              (not (list.any? (same? it) banned))))
                                 [value]))
                           var')]
          (in []))
        (write! new_value var))))

  (def .public (changes target)
    (All (_ a) (-> (Var a) (IO [(Channel a) (Sink a)])))
    (do io.monad
      [.let [[channel sink] (sharing [a]
                              (is (Var a)
                                  target)
                              (is [(Channel a) (Sink a)]
                                  (frp.channel [])))]
       _ (atom.update! (function (_ [value observers])
                         [value {.#Item (implementation
                                         (def close (at sink close))
                                         (def feed (|>> variance.read (at sink feed))))
                                        observers}])
                       (representation target))]
      (in [channel sink])))
  )

(type: (Tx_Frame r w)
  (Record
   [#var (Var' r w)
    #original r
    #current w]))

(type: Tx
  (List (Ex (_ r w) (Tx_Frame r w))))

(type: .public (STM a)
  (-> Tx [Tx a]))

(def (var_value var tx)
  (All (_ r w) (-> (Var' r w) Tx (Maybe r)))
  (|> tx
      (list.example (function (_ [_var _original _current])
                      (same? (as (Var Any) var)
                             (as (Var Any) _var))))
      (at maybe.monad each (function (_ [_var _original _current])
                             _current))
      as_expected))

(def .public (read var)
  (All (_ r w) (-> (Var' r w) (STM r)))
  (function (_ tx)
    (case (var_value var tx)
      {.#Some value}
      [tx value]

      {.#None}
      (let [value (..read! var)]
        [{.#Item [#var var
                  #original value
                  #current (as_expected value)]
                 tx}
         value]))))

(def (with_updated_var var value tx)
  (All (_ r w) (-> (Var' r w) w Tx Tx))
  (case tx
    {.#End}
    {.#End}
    
    {.#Item [_var _original _current] tx'}
    (if (same? (as (Var Any) var)
               (as (Var Any) _var))
      {.#Item [#var _var
               #original _original
               #current (as_expected value)]
              tx'}
      {.#Item [#var _var
               #original _original
               #current _current]
              (with_updated_var var value tx')})))

(def .public (write value var)
  (All (_ r w) (-> w (Var' r w) (STM Any)))
  (function (_ tx)
    (case (var_value var tx)
      {.#Some _}
      [(with_updated_var var value tx)
       []]

      {.#None}
      [{.#Item [#var var
                #original (..read! var)
                #current value]
               tx}
       []])))

(def .public functor
  (Functor STM)
  (implementation
   (def (each f fa)
     (function (_ tx)
       (let [[tx' a] (fa tx)]
         [tx' (f a)])))))

(def .public apply
  (Apply STM)
  (implementation
   (def functor ..functor)

   (def (on fa ff)
     (function (_ tx)
       (let [[tx' f] (ff tx)
             [tx'' a] (fa tx')]
         [tx'' (f a)])))))

(def .public monad
  (Monad STM)
  (implementation
   (def functor ..functor)

   (def (in a)
     (function (_ tx)
       [tx a]))

   (def (conjoint mma)
     (function (_ tx)
       (let [[tx' ma] (mma tx)]
         (ma tx'))))))

(def .public (update f var)
  (All (_ r w) (-> (-> r w) (Var' r w) (STM [r w])))
  (do ..monad
    [before (..read var)
     .let [after (f before)]
     _ (..write after var)]
    (in [before after])))

(def (can_commit? tx)
  (-> Tx Bit)
  (list.every? (function (_ [_var _original _current])
                 (same? _original (..read! _var)))
               tx))

(def (commit_var! [_var _original _current])
  (-> (Ex (_ r w) (Tx_Frame r w)) (IO Any))
  (if (same? (as Any _original) (as Any _current))
    (io [])
    (..write! _current _var)))

(def fresh_tx Tx (list))

(type: (Commit a)
  [(STM a)
   (Async a)
   (Resolver a)])

(def pending_commits
  (Atom (Rec Commits
          [(Async [(Ex (_ a) (Commit a)) Commits])
           (Resolver [(Ex (_ a) (Commit a)) Commits])]))
  (atom (async.async [])))

(def commit_processor_flag
  (Atom Bit)
  (atom #0))

(def (issue_commit! commit)
  (All (_ a) (-> (Commit a) (IO Any)))
  (let [entry [commit (async.async [])]]
    (do [! io.monad]
      [|commits|&resolve (atom.read! pending_commits)]
      (loop (again [[|commits| resolve] |commits|&resolve])
        (do !
          [|commits| (async.value |commits|)]
          (case |commits|
            {.#None}
            (do io.monad
              [resolved? (resolve entry)]
              (if resolved?
                (atom.write! (product.right entry) pending_commits)
                (again |commits|&resolve)))
            
            {.#Some [head tail]}
            (again tail)))))))

(def (process_commit! commit)
  (All (_ a) (-> (Commit a) (IO Any)))
  (let [[stm_proc output resolve] commit
        [finished_tx value] (stm_proc fresh_tx)]
    (if (can_commit? finished_tx)
      (do [! io.monad]
        [_ (monad.each ! ..commit_var! finished_tx)]
        (resolve value))
      (..issue_commit! commit))))

(def start_commit_processor!
  (IO Any)
  (do [! io.monad]
    [flag (atom.read! commit_processor_flag)]
    (if flag
      (in [])
      (do !
        [was_first? (atom.compare_and_swap! flag #1 commit_processor_flag)]
        (if was_first?
          (do !
            [[async resolve] (atom.read! pending_commits)]
            (async.upon! (function (again [head [tail _resolve]])
                           (do !
                             [_ (..process_commit! head)]
                             (async.upon! again tail)))
                         async))
          (in [])))
      )))

(def .public (commit! stm_proc)
  (All (_ a) (-> (STM a) (Async a)))
  (let [[output resolver] (async.async [])]
    (exec
      (io.run! (do io.monad
                 [_ ..start_commit_processor!]
                 (..issue_commit! [stm_proc output resolver])))
      output)))