diff options
Diffstat (limited to '')
| -rw-r--r-- | stdlib/source/lux/control/concurrency/actor.lux | 84 | ||||
| -rw-r--r-- | stdlib/source/lux/control/concurrency/promise.lux | 100 | ||||
| -rw-r--r-- | stdlib/source/lux/control/concurrency/stm.lux | 86 | ||||
| -rw-r--r-- | stdlib/source/lux/control/concurrency/task.lux | 26 |
4 files changed, 165 insertions, 131 deletions
diff --git a/stdlib/source/lux/control/concurrency/actor.lux b/stdlib/source/lux/control/concurrency/actor.lux index 6f4ddf2ad..9b20dcfde 100644 --- a/stdlib/source/lux/control/concurrency/actor.lux +++ b/stdlib/source/lux/control/concurrency/actor.lux @@ -23,7 +23,7 @@ abstract]] [// ["." atom (#+ Atom atom)] - ["." promise (#+ Promise promise) ("promise/." Monad<Promise>)] + ["." promise (#+ Promise Resolver) ("promise/." Monad<Promise>)] ["." task (#+ Task)]]) (exception: #export poisoned) @@ -37,11 +37,17 @@ (with-expansions [<Message> (as-is (-> s (Actor s) (Task s))) <Obituary> (as-is [Text s (List <Message>)]) - <Mailbox> (as-is (Rec Mailbox (Promise [<Message> Mailbox])))] - - (def: (obituary mailbox) - (All [a] (-> (Rec Mailbox (Promise [a Mailbox])) (List a))) - (case (promise.poll mailbox) + <Mailbox> (as-is (Rec Mailbox + [(Promise [<Message> Mailbox]) + (Resolver [<Message> Mailbox])]))] + + (def: (obituary [read write]) + (All [a] + (-> (Rec Mailbox + [(Promise [a Mailbox]) + (Resolver [a Mailbox])]) + (List a))) + (case (promise.poll read) (#.Some [head tail]) (#.Cons head (obituary tail)) @@ -50,12 +56,18 @@ (abstract: #export (Actor s) {#.doc "An actor, defined as all the necessities it requires."} + {#mailbox (Atom <Mailbox>) - #obituary (Promise <Obituary>)} + #obituary [(Promise <Obituary>) + (Resolver <Obituary>)]} ## TODO: Delete after new-luxc becomes the new standard compiler. (def: (actor mailbox obituary) - (All [s] (-> (Atom <Mailbox>) (Promise <Obituary>) (Actor s))) + (All [s] + (-> (Atom <Mailbox>) + [(Promise <Obituary>) + (Resolver <Obituary>)] + (Actor s))) (:abstraction {#mailbox mailbox #obituary obituary})) @@ -74,10 +86,10 @@ {#.doc "Given a behavior and initial state, spawns an actor and returns it."} (All [s] (-> (Behavior s) s (IO (Actor s)))) (io (let [[handle end] behavior - self (actor (atom (promise #.None)) - (promise #.None)) + self (actor (atom (promise.promise [])) + (promise.promise [])) process (loop [state init - |mailbox| (io.run (atom.read (get@ #mailbox (:representation self))))] + [|mailbox| _] (io.run (atom.read (get@ #mailbox (:representation self))))] (do promise.Monad<Promise> [[head tail] |mailbox| ?state' (handle head state self)] @@ -85,9 +97,9 @@ (#e.Failure error) (do @ [_ (end error state)] - (exec (io.run (promise.resolve [error state (#.Cons head (obituary tail))] - (get@ #obituary (:representation self)))) - (wrap []))) + (let [[_ resolve] (get@ #obituary (:representation self))] + (exec (io.run (resolve [error state (#.Cons head (..obituary tail))])) + (wrap [])))) (#e.Success state') (recur state' tail))))] @@ -95,35 +107,37 @@ (def: #export (alive? actor) (All [s] (-> (Actor s) Bit)) - (case (promise.poll (get@ #obituary (:representation actor))) - #.None - #1 + (let [[obituary _] (get@ #obituary (:representation actor))] + (case (promise.poll obituary) + #.None + #1 - _ - #0)) + _ + #0))) (def: #export (send message actor) {#.doc "Communicate with an actor through message passing."} (All [s] (-> (Message s) (Actor s) (IO Bit))) (if (alive? actor) - (let [entry [message (promise #.None)]] + (let [entry [message (promise.promise [])]] (do Monad<IO> - [|mailbox| (atom.read (get@ #mailbox (:representation actor)))] - (loop [|mailbox| |mailbox|] + [|mailbox|&resolve (atom.read (get@ #mailbox (:representation actor)))] + (loop [[|mailbox| resolve] |mailbox|&resolve] (case (promise.poll |mailbox|) #.None (do @ - [resolved? (promise.resolve entry |mailbox|)] + [resolved? (resolve entry)] (if resolved? (do @ [_ (atom.write (product.right entry) (get@ #mailbox (:representation actor)))] (wrap #1)) - (recur |mailbox|))) + (recur |mailbox|&resolve))) (#.Some [_ |mailbox|']) (recur |mailbox|'))))) (io/wrap #0))) - )) + ) + ) ## [Values] (def: (default-handle message state self) @@ -153,7 +167,7 @@ (-> Name cs.Annotations cs.Annotations) (|>> (#.Cons [(name-of <tag>) (code.tag name)]))) - + (def: #export (<resolve> name) (-> Name (Meta Name)) (do Monad<Meta> @@ -211,16 +225,16 @@ {#.doc (doc "Defines an actor, with its behavior and internal state." (actor: #export Counter Nat - + ((stop cause state) (:: promise.Monad<Promise> wrap (log! (if (ex.match? ..poisoned cause) (format "Counter was poisoned: " (%n state)) cause))))) - + (actor: #export (Stack a) (List a) - + ((handle message state self) (do task.Monad<Task> [#let [_ (log! "BEFORE")] @@ -317,7 +331,7 @@ (push [value a] state self (List a)) (let [state' (#.Cons value state)] (task.return [state' state']))))} - (with-gensyms [g!_ g!return g!error g!task g!sent?] + (with-gensyms [g!_ g!return g!error g!task g!sent? g!resolve] (do @ [current-module macro.current-module-name actor-name (resolve-actor actor-name) @@ -350,7 +364,9 @@ (with-message actor-name) csw.annotations)) (All [(~+ g!all-vars)] (-> (~+ g!inputsT) (~ actorC) (Task (~ (get@ #output signature))))) - (let [(~ g!task) (task.task (~ g!outputT))] + (let [[(~ g!task) (~ g!resolve)] (: [(task.Task (~ g!outputT)) + (task.Resolver (~ g!outputT))] + (task.task []))] (io.run (do io.Monad<IO> [(~ g!sent?) (..send (function ((~ g!_) (~ g!state) (~ g!self)) (do promise.Monad<Promise> @@ -361,11 +377,11 @@ (~ body)))] (case (~ g!return) (#.Right [(~ g!state) (~ g!return)]) - (exec (io.run (promise.resolve (#.Right (~ g!return)) (~ g!task))) + (exec (io.run ((~ g!resolve) (#.Right (~ g!return)))) (task.return (~ g!state))) - + (#.Left (~ g!error)) - (exec (io.run (promise.resolve (#.Left (~ g!error)) (~ g!task))) + (exec (io.run ((~ g!resolve) (#.Left (~ g!error)))) (task.fail (~ g!error)))) )) (~ g!self))] diff --git a/stdlib/source/lux/control/concurrency/promise.lux b/stdlib/source/lux/control/concurrency/promise.lux index 1a471022f..2530d6080 100644 --- a/stdlib/source/lux/control/concurrency/promise.lux +++ b/stdlib/source/lux/control/concurrency/promise.lux @@ -18,9 +18,38 @@ {#.doc "Represents values produced by asynchronous computations (unlike IO, which is synchronous)."} (Atom [(Maybe a) (List (-> a (IO Any)))]) - (def: #export (promise ?value) - (All [a] (-> (Maybe a) (Promise a))) - (:abstraction (atom [?value (list)]))) + (type: #export (Resolver a) + (-> a (IO Bit))) + + (def: (resolver (^:representation promise)) + {#.doc "Sets an promise's value if it has not been done yet."} + (All [a] (-> (Promise a) (Resolver a))) + (function (resolve value) + (do io.Monad<IO> + [(^@ old [_value _observers]) (atom.read promise)] + (case _value + (#.Some _) + (wrap #0) + + #.None + (do @ + [#let [new [(#.Some value) #.None]] + succeeded? (atom.compare-and-swap old new promise)] + (if succeeded? + (do @ + [_ (monad.map @ (function (_ f) (f value)) + _observers)] + (wrap #1)) + (resolve value))))))) + + (def: #export (resolved value) + (All [a] (-> a (Promise a))) + (:abstraction (atom [(#.Some value) (list)]))) + + (def: #export (promise _) + (All [a] (-> Any [(Promise a) (Resolver a)])) + (let [promise (:abstraction (atom [#.None (list)]))] + [promise (..resolver promise)])) (def: #export (poll (^:representation promise)) {#.doc "Polls a promise's value."} @@ -29,37 +58,17 @@ io.run product.left)) - (def: #export (resolve value (^:representation promise)) - {#.doc "Sets an promise's value if it has not been done yet."} - (All [a] (-> a (Promise a) (IO Bit))) - (do io.Monad<IO> - [(^@ old [_value _observers]) (atom.read promise)] - (case _value - (#.Some _) - (wrap #0) - - #.None - (do @ - [#let [new [(#.Some value) #.None]] - succeeded? (atom.compare-and-swap old new promise)] - (if succeeded? - (do @ - [_ (monad.map @ (function (_ f) (f value)) - _observers)] - (wrap #1)) - (resolve value (:abstraction promise))))))) - (def: #export (await f (^:representation promise)) - (All [a] (-> (-> a (IO Any)) (Promise a) Any)) + (All [a] (-> (-> a (IO Any)) (Promise a) (IO Any))) (let [(^@ old [_value _observers]) (io.run (atom.read promise))] (case _value (#.Some value) - (io.run (f value)) + (f value) #.None (let [new [_value (#.Cons f _observers)]] (if (io.run (atom.compare-and-swap old new promise)) - [] + (io.io []) (await f (:abstraction promise))))))) ) @@ -75,34 +84,31 @@ (structure: #export _ (Functor Promise) (def: (map f fa) - (let [fb (promise #.None)] - (exec (await (function (_ a) (resolve (f a) fb)) - fa) + (let [[fb resolve] (..promise [])] + (exec (io.run (await (|>> f resolve) fa)) fb)))) (structure: #export _ (Apply Promise) (def: functor Functor<Promise>) (def: (apply ff fa) - (let [fb (promise #.None)] - (exec (await (function (_ f) - (io (await (function (_ a) (resolve (f a) fb)) - fa))) - ff) + (let [[fb resolve] (..promise [])] + (exec (io.run (await (function (_ f) + (await (|>> f resolve) fa)) + ff)) fb)))) (structure: #export _ (Monad Promise) (def: functor Functor<Promise>) (def: (wrap a) - (promise (#.Some a))) + (..resolved a)) (def: (join mma) - (let [ma (promise #.None)] - (exec (await (function (_ ma') - (io (await (function (_ a') (resolve a' ma)) - ma'))) - mma) + (let [[ma resolve] (promise [])] + (exec (io.run (await (function (_ ma') + (await resolve ma')) + mma)) ma)))) (def: #export (and left right) @@ -116,11 +122,10 @@ (def: #export (or left right) {#.doc "Heterogeneous alternative combinator."} (All [a b] (-> (Promise a) (Promise b) (Promise (| a b)))) - (let [a|b (promise #.None)] + (let [[a|b resolve] (..promise [])] (with-expansions [<sides> (do-template [<promise> <tag>] - [(await (function (_ value) (resolve (<tag> value) a|b)) - <promise>)] + [(io.run (await (|>> <tag> resolve) <promise>))] [left #.Left] [right #.Right] @@ -131,10 +136,9 @@ (def: #export (either left right) {#.doc "Homogeneous alternative combinator."} (All [a] (-> (Promise a) (Promise a) (Promise a))) - (let [left||right (promise #.None)] + (let [[left||right resolve] (..promise [])] (`` (exec (~~ (do-template [<promise>] - [(await (function (_ value) (resolve value left||right)) - <promise>)] + [(io.run (await resolve <promise>))] [left] [right])) @@ -144,10 +148,10 @@ {#.doc (doc "Runs an I/O computation on its own process (after a specified delay)." "Returns a Promise that will eventually host its result.")} (All [a] (-> Nat (IO a) (Promise a))) - (let [!out (promise #.None)] + (let [[!out resolve] (..promise [])] (exec (|> (do io.Monad<IO> [value computation] - (resolve value !out)) + (resolve value)) (process.schedule millis-delay) io.run) !out))) diff --git a/stdlib/source/lux/control/concurrency/stm.lux b/stdlib/source/lux/control/concurrency/stm.lux index 3203b2d52..f54e16baf 100644 --- a/stdlib/source/lux/control/concurrency/stm.lux +++ b/stdlib/source/lux/control/concurrency/stm.lux @@ -9,12 +9,12 @@ ["." product] ["." maybe] [collection - ["." list ("list/." Functor<List> Fold<List>)]]] + ["." list]]] [type abstract]] [// ["." atom (#+ Atom atom)] - ["." promise (#+ Promise promise)] + ["." promise (#+ Promise Resolver)] ["." frp ("frp/." Functor<Channel>)]]) (type: #export (Observer a) @@ -177,45 +177,54 @@ tx)) (def: (commit-var! [_var _original _current]) - (-> (Ex [a] (Tx-Frame a)) Any) + (-> (Ex [a] (Tx-Frame a)) (IO Any)) (if (is? _original _current) - [] - (io.run (write! _current _var)))) + (io []) + (write! _current _var))) (def: fresh-tx Tx (list)) -(type: Commit (Ex [a] [(STM a) (Promise a)])) +(type: (Commit a) + [(STM a) + (Promise a) + (Resolver a)]) (def: pending-commits - (Atom (Rec Commits (Promise [Commit Commits]))) - (atom (promise #.None))) + (Atom (Rec Commits + [(Promise [(Ex [a] (Commit a)) Commits]) + (Resolver [(Ex [a] (Commit a)) Commits])])) + (atom (promise.promise []))) (def: commit-processor-flag (Atom Bit) (atom #0)) (def: (issue-commit commit) - (-> Commit (IO Any)) - (let [entry [commit (promise #.None)]] - (loop [|commits| (io.run (atom.read pending-commits))] - (case (promise.poll |commits|) - #.None - (do io.Monad<IO> - [resolved? (promise.resolve entry |commits|)] - (if resolved? - (atom.write (product.right entry) pending-commits) - (recur |commits|))) - - (#.Some [head tail]) - (recur tail))))) - -(def: (process-commit [stm-proc output]) - (-> [(STM Any) (Promise Any)] Any) - (let [[finished-tx value] (stm-proc fresh-tx)] - (io.run (if (can-commit? finished-tx) - (exec (list/map commit-var! finished-tx) - (promise.resolve value output)) - (issue-commit [stm-proc output]))))) + (All [a] (-> (Commit a) (IO Any))) + (let [entry [commit (promise.promise [])]] + (do io.Monad<IO> + [|commits|&resolve (atom.read pending-commits)] + (loop [[|commits| resolve] |commits|&resolve] + (case (promise.poll |commits|) + #.None + (do io.Monad<IO> + [resolved? (resolve entry)] + (if resolved? + (atom.write (product.right entry) pending-commits) + (recur |commits|&resolve))) + + (#.Some [head tail]) + (recur 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<IO> + [_ (monad.map @ commit-var! finished-tx)] + (resolve value)) + (issue-commit commit)))) (def: init-processor! (IO Any) @@ -226,11 +235,13 @@ (do @ [was-first? (atom.compare-and-swap flag #1 commit-processor-flag)] (if was-first? - (exec (|> (io.run (atom.read pending-commits)) - (promise.await (function (recur [head tail]) - (io (exec (process-commit (:coerce [(STM Any) (Promise Any)] head)) - (promise.await recur tail)))))) - (wrap [])) + (do @ + [[promise resolve] (atom.read pending-commits)] + (promise.await (function (recur [head [tail _resolve]]) + (do @ + [_ (process-commit head)] + (promise.await recur tail))) + promise)) (wrap []))) ))) @@ -239,7 +250,8 @@ "Note that a transaction may be re-run an indeterminate number of times if other transactions involving the same variables successfully commit first." "For this reason, it's important to note that transactions must be free from side-effects, such as I/O.")} (All [a] (-> (STM a) (Promise a))) - (let [output (promise #.None)] - (exec (io.run init-processor!) - (io.run (issue-commit [stm-proc output])) + (let [[output resolver] (promise.promise [])] + (exec (io.run (do io.Monad<IO> + [_ init-processor!] + (issue-commit [stm-proc output resolver]))) output))) diff --git a/stdlib/source/lux/control/concurrency/task.lux b/stdlib/source/lux/control/concurrency/task.lux index 96bc40f0a..a5bf17819 100644 --- a/stdlib/source/lux/control/concurrency/task.lux +++ b/stdlib/source/lux/control/concurrency/task.lux @@ -15,19 +15,23 @@ (type: #export (Task a) (Promise (Error a))) -(def: #export (fail error) - (All [a] (-> Text (Task a))) - (:: promise.Monad<Promise> wrap (#error.Failure error))) +(type: #export (Resolver a) + (promise.Resolver (Error a))) + +(do-template [<name> <input> <tag>] + [(def: #export <name> + (All [a] (-> <input> (Task a))) + (|>> <tag> promise.resolved))] + + [return a #error.Success] + [fail Text #error.Failure] + ) (def: #export (throw exception message) (All [e a] (-> (Exception e) e (Task a))) (:: promise.Monad<Promise> wrap (ex.throw exception message))) -(def: #export (return value) - (All [a] (-> a (Task a))) - (:: promise.Monad<Promise> wrap (#error.Success value))) - (def: #export (try computation) (All [a] (-> (Task a) (Task (Error a)))) (:: promise.Functor<Promise> map (|>> #error.Success) computation)) @@ -71,11 +75,9 @@ (#error.Success ma) ma)))) -(syntax: #export (task {type s.any}) - {#.doc (doc "Makes an uninitialized Task (in this example, of Any)." - (task Any))} - (wrap (list (` (: (..Task (~ type)) - (promise.promise #.None)))))) +(def: #export task + (All [a] (-> Any [(Task a) (Resolver a)])) + promise.promise) (def: #export (from-promise promise) (All [a] (-> (Promise a) (Task a))) |