-! Copyright (C) 2006, 2007 Slava Pestov.
+! Copyright (C) 2006, 2008 Slava Pestov.
! See http://factorcode.org/license.txt for BSD license.
USING: arrays generator generator.registers generator.fixup
hashtables kernel math namespaces sequences words
inference.state inference.backend inference.dataflow system
math.parser classes alien.arrays alien.c-types alien.structs
alien.syntax cpu.architecture alien inspector quotations assocs
-kernel.private threads continuations.private libc combinators
-compiler.errors continuations ;
+kernel.private concurrency.threads continuations.private libc
+combinators compiler.errors continuations ;
IN: alien.compiler
! Common protocol for alien-invoke/alien-callback/alien-indirect
! Copyright (C) 2007 Slava Pestov.
! See http://factorcode.org/license.txt for BSD license.
-USING: alien alien.c-types parser threads words kernel.private
-kernel ;
+USING: alien alien.c-types parser concurrency.threads words
+kernel.private kernel ;
IN: alien.remote-control
: eval-callback
: data-base 1024 ; inline
-: userenv-size 40 ; inline
+: userenv-size 64 ; inline
: header-size 10 ; inline
"libc" require
"io.streams.c" require
+ "io.thread" require
"vocabs.loader" require
"syntax" require
USING: kernel namespaces arrays sequences io inference.backend
inference.state generator debugger math.parser prettyprint words
compiler.units continuations vocabs assocs alien.compiler dlists
-optimizer definitions math compiler.errors threads graphs
-generic ;
+optimizer definitions math compiler.errors concurrency.threads
+graphs generic ;
IN: compiler
: compiled-usages ( words -- seq )
--- /dev/null
+Slava Pestov
--- /dev/null
+Co-operative threading
--- /dev/null
+USING: help.markup help.syntax kernel kernel.private io
+concurrency.threads.private continuations dlists init
+quotations strings assocs heaps ;
+IN: concurrency.threads
+
+ARTICLE: "threads-start/stop" "Starting and stopping threads"
+"Spawning new threads:"
+{ $subsection spawn }
+"Creating and spawning a thread can be factored out into two separate steps:"
+{ $subsection <thread> }
+{ $subsection (spawn) }
+"Threads stop either when the quotation given to " { $link spawn } " returns, or when the following word is called:"
+{ $subsection stop }
+"If the image is saved and started again, all runnable threads are stopped. Vocabularies wishing to have a background thread always running should use " { $link add-init-hook } "." ;
+
+ARTICLE: "threads-yield" "Yielding and suspending threads"
+"Yielding to other threads:"
+{ $subsection yield }
+{ $subsection sleep }
+"Threads can be suspended and woken up at some point in the future when a condition is satisfied:"
+{ $subsection suspend }
+{ $subsection resume }
+{ $subsection resume-with } ;
+
+ARTICLE: "thread-state" "Thread-local state"
+"Threads form a class of objects:"
+{ $subsection thread }
+"The current thread:"
+{ $subsection self }
+"Thread-local variables:"
+{ $subsection tnamespace }
+{ $subsection tget }
+{ $subsection tset }
+{ $subsection tchange }
+"Global hashtable of all threads, keyed by " { $link thread-id } ":"
+{ $subsection threads }
+"Threads have an identity independent of continuations. If a continuation is refied in one thread and then resumed in another thread, the code running in that continuation will observe a change in the value output by " { $link self } "." ;
+
+ARTICLE: "thread-impl" "Thread implementation"
+"Thread implementation:"
+{ $subsection run-queue }
+{ $subsection sleep-queue } ;
+
+ARTICLE: "threads" "Lightweight co-operative threads"
+"Factor supports lightweight co-operative threads implemented on top of continuations. A thread will yield while waiting for I/O operations to complete, or when a yield has been explicitly requested."
+$nl
+"Words for working with threads are in the " { $vocab-link "concurrency.threads" } " vocabulary."
+{ $subsection "threads-start/stop" }
+{ $subsection "threads-yield" }
+{ $subsection "thread-state" }
+{ $subsection "thread-impl" } ;
+
+ABOUT: "threads"
+
+HELP: thread
+{ $class-description "A thread. The slots are as follows:"
+ { $list
+ { { $link thread-id } " - a unique identifier assigned to each thread." }
+ { { $link thread-name } " - the name passed to " { $link spawn } "." }
+ { { $link thread-quot } " - the initial quotation passed to " { $link spawn } "." }
+ { { $link thread-continuation } " - if the thread is waiting to run, the saved thread context. If the thread is currently running, will be " { $link f } "." }
+ { { $link thread-registered? } " - a boolean indicating whether the thread is eligible to run or not. Spawning a thread with " { $link (spawn) } " sets this flag and " { $link stop } " clears it." }
+ }
+} ;
+
+HELP: self
+{ $values { "thread" thread } }
+{ $description "Pushes the currently-running thread." } ;
+
+HELP: <thread>
+{ $values { "quot" quotation } { "name" string } { "error-handler" quotation } }
+{ $description "Low-level thread constructor. The thread runs the quotation when spawned; the name is simply used to identify the thread for debugging purposes. The error handler is called if the thread's quotation throws an unhandled error; it should either print the error or notify another thread." }
+{ $notes "In most cases, user code should call " { $link spawn } " instead, however for control over the error handler quotation, threads can be created with " { $link <thread> } " then passed to " { $link (spawn) } "." } ;
+
+HELP: run-queue
+{ $values { "queue" dlist } }
+{ $var-description "Global variable holding the queue of runnable threads. Calls to " { $link yield } " switch to the thread which has been in the queue for the longest period of time."
+$nl
+"By convention, threads are queued with " { $link push-front }
+" and dequeued with " { $link pop-back } "." } ;
+
+HELP: resume
+{ $values { "thread" thread } }
+{ $description "Adds a thread to the end of the run queue. The thread must have previously been suspended by a call to " { $link suspend } "." } ;
+
+HELP: resume-with
+{ $values { "obj" object } { "thread" thread } }
+{ $description "Adds a thread to the end of the run queue together with an object to pass to the thread. The thread must have previously been suspended by a call to " { $link suspend } "; the object is returned from the " { $link suspend } " call." } ;
+
+HELP: sleep-queue
+{ $var-description "A " { $link min-heap } " storing the queue of sleeping threads." } ;
+
+HELP: sleep-time
+{ $values { "ms" "a non-negative integer or " { $link f } } }
+{ $description "Outputs the time until the next sleeping thread is scheduled to wake up, which could be zero if there are threads in the run queue, or threads which need to wake up right now. If there are no runnable or sleeping threads, outputs " { $link f } "." } ;
+
+HELP: stop
+{ $description "Stops the current thread. The thread may be started again from another thread using " { $link (spawn) } "." } ;
+
+HELP: yield
+{ $description "Adds the current thread to the end of the run queue, and switches to the next runnable thread." } ;
+
+HELP: sleep
+{ $values { "ms" "a non-negative integer" } }
+{ $description "Suspends the current thread for " { $snippet "ms" } " milliseconds. It will not get woken up before this time period elapses, but since the multitasker is co-operative, the precise wakeup time is dependent on when other threads yield." } ;
+
+HELP: suspend
+{ $values { "quot" "a quotation with stack effect " { $snippet "( thread -- )" } } { "obj" object } }
+{ $description "Suspends the current thread and passes it to the quotation. After the quotation returns, control yields to the next runnable thread and the current thread does not execute again until it is resumed, and so the quotation must arrange for another thread to later resume the suspended thread with a call to " { $link resume } " or " { $link resume-with } "." } ;
+
+HELP: spawn
+{ $values { "quot" quotation } { "name" string } }
+{ $description "Spawns a new thread. The thread begins executing the given quotation; the name is for debugging purposes. The new thread begins running immediately and the current thread is added to the end of the run queue."
+$nl
+"The new thread begins with an empty data stack, an empty catch stack and a name stack containing the global namespace only. This means that the only way to pass data to the new thread is to explicitly construct a quotation containing the data, for example using " { $link curry } " or " { $link compose } "." }
+{ $examples
+ { $code "1 2 [ + . ] 2curry \"Addition thread\" spawn" }
+} ;
+
+HELP: init-threads
+{ $description "Called during startup to initialize the threading system. This word should never be called directly." } ;
+
+HELP: tnamespace
+{ $values { "assoc" assoc } }
+{ $description "Outputs the current thread's set of thread-local variables." } ;
+
+HELP: tget
+{ $values { "key" object } { "value" object } }
+{ $description "Outputs the value of a thread-local variable." } ;
+
+HELP: tset
+{ $values { "value" object } { "key" object } }
+{ $description "Sets the value of a thread-local variable." } ;
+
+HELP: tchange
+{ $values { "key" object } { "quot" "a quotation with stack effect " { $snippet "( value -- newvalue )" } } }
+{ $description "Applies the quotation to the current value of a thread-local variable, storing the result back to the same variable." } ;
--- /dev/null
+USING: namespaces io tools.test concurrency.threads kernel ;
+IN: temporary
+
+3 "x" set
+namespace [ [ yield 2 "x" set ] bind ] curry "Test" spawn drop
+[ 2 ] [ yield "x" get ] unit-test
+[ ] [ [ flush ] "flush test" spawn drop flush ] unit-test
+[ ] [ [ "Errors, errors" throw ] "error test" spawn drop ] unit-test
+yield
+
+[ ] [ 0.3 sleep ] unit-test
+[ "hey" sleep ] must-fail
+
+[ 3 ] [
+ [ 3 swap resume-with ] suspend
+] unit-test
--- /dev/null
+! Copyright (C) 2004, 2008 Slava Pestov.
+! Copyright (C) 2005 Mackenzie Straight.
+! See http://factorcode.org/license.txt for BSD license.
+IN: concurrency.threads
+USING: arrays hashtables heaps kernel kernel.private math
+namespaces sequences vectors continuations continuations.private
+dlists assocs system combinators debugger prettyprint io init ;
+
+SYMBOL: initial-thread
+
+TUPLE: thread
+name quot error-handler
+id registered?
+continuation
+mailbox variables ;
+
+: self ( -- thread ) 40 getenv ; inline
+
+! Thread-local storage
+: tnamespace ( -- assoc ) self thread-variables ;
+
+: tget ( key -- value ) tnamespace at ;
+
+: tset ( value key -- ) tnamespace set-at ;
+
+: tchange ( key quot -- ) tnamespace change-at ; inline
+
+SYMBOL: threads
+
+threads global [ H{ } assoc-like ] change-at
+
+: thread ( id -- thread ) threads get-global at ;
+
+<PRIVATE
+
+: check-unregistered
+ dup thread-registered?
+ [ "Registering a thread twice" throw ] when ;
+
+: check-registered
+ dup thread-registered?
+ [ "Unregistering a thread twice" throw ] unless ;
+
+: register-thread ( thread -- )
+ check-unregistered
+ t over set-thread-registered?
+ dup thread-id threads get-global set-at ;
+
+: unregister-thread ( thread -- )
+ check-registered
+ f over set-thread-registered?
+ thread-id threads get-global delete-at ;
+
+: set-self ( thread -- ) 40 setenv ; inline
+
+: <thread> ( quot name error-handler -- thread )
+ \ thread counter H{ } clone {
+ set-thread-quot
+ set-thread-name
+ set-thread-error-handler
+ set-thread-id
+ set-thread-variables
+ } \ thread construct ;
+
+PRIVATE>
+
+SYMBOL: run-queue
+SYMBOL: sleep-queue
+
+: resume ( thread -- )
+ check-registered run-queue get-global push-front ;
+
+: resume-with ( obj thread -- )
+ check-registered 2array run-queue get-global push-front ;
+
+<PRIVATE
+
+: schedule-sleep ( thread ms -- )
+ >r check-registered r> sleep-queue get-global heap-push ;
+
+: wake-up? ( heap -- ? )
+ dup heap-empty?
+ [ drop f ] [ heap-peek nip millis <= ] if ;
+
+: wake-up ( -- )
+ sleep-queue get-global
+ [ dup wake-up? ] [ dup heap-pop drop resume ] [ ] while
+ drop ;
+
+: next ( -- )
+ walker-hook [
+ continue
+ ] [
+ wake-up
+ run-queue get-global pop-back
+ dup array? [ first2 ] [ f swap ] if dup set-self
+ dup thread-continuation
+ f rot set-thread-continuation
+ continue-with
+ ] if* ;
+
+PRIVATE>
+
+: sleep-time ( -- ms )
+ {
+ { [ run-queue get-global dlist-empty? not ] [ 0 ] }
+ { [ sleep-queue get-global heap-empty? ] [ f ] }
+ { [ t ] [ sleep-queue get-global heap-peek nip millis [-] ] }
+ } cond ;
+
+: stop ( -- )
+ self unregister-thread next ;
+
+: suspend ( quot -- obj )
+ [
+ >r self [ set-thread-continuation ] keep r> call next
+ ] curry callcc1 ; inline
+
+: yield ( -- ) [ resume ] suspend drop ;
+
+: sleep ( ms -- )
+ >fixnum millis + [ schedule-sleep ] curry suspend drop ;
+
+: (spawn) ( thread -- )
+ [
+ resume [
+ dup set-self
+ dup register-thread
+ init-namespaces
+ V{ } set-catchstack
+ { } set-retainstack
+ >r { } set-datastack r>
+ thread-quot [ call stop ] call-clear
+ ] 1 (throw)
+ ] suspend 2drop ;
+
+: spawn ( quot name -- thread )
+ [
+ global [
+ "Error in thread " write
+ dup thread-id pprint
+ " (" write
+ dup thread-name pprint ")" print
+ "spawned to call " write
+ thread-quot short.
+ nl
+ print-error flush
+ ] bind
+ ] <thread>
+ [ (spawn) ] keep ;
+
+: in-thread ( quot -- ) "Thread" spawn drop ;
+
+<PRIVATE
+
+: init-threads ( -- )
+ <dlist> run-queue set-global
+ <min-heap> sleep-queue set-global
+ H{ } clone threads set-global
+ initial-thread global
+ [ drop f "Initial" [ die ] <thread> ] cache
+ f over set-thread-continuation
+ f over set-thread-registered?
+ dup register-thread
+ set-self ;
+
+[ self dup thread-error-handler call stop ]
+thread-error-hook set-global
+
+PRIVATE>
+
+[ init-threads ] "concurrency.threads" add-init-hook
PRIVATE>
+SYMBOL: thread-error-hook
+
: rethrow ( error -- * )
- catchstack* empty? [ die ] when
+ catchstack* empty? [
+ thread-error-hook get-global
+ [ 1 (throw) ] [ die ] if*
+ ] when
dup save-error c> continue-with ;
: recover ( try recovery -- )
inference inference.backend inference.dataflow io kernel
kernel.private layouts math namespaces optimizer
optimizer.specializers prettyprint quotations sequences system
-threads words vectors ;
+concurrency.threads words vectors ;
IN: generator
SYMBOL: compile-queue
M: object normalize-pathname ;
-[ init-io embedded? [ init-stdio ] unless ]
-"io.backend" add-init-hook
-
: set-io-backend ( backend -- )
io-backend set-global init-io init-stdio ;
+
+[ init-io embedded? [ init-stdio ] unless ]
+"io.backend" add-init-hook
[ f ] [ "test-blah" resource-path exists? ] unit-test
-[ ] [ "test-quux.txt" resource-path [ [ yield "Hi" write ] in-thread ] with-file-writer ] unit-test
+[ ] [ "test-quux.txt" resource-path [ [ yield "Hi" write ] "Test" spawn drop ] with-file-writer ] unit-test
[ ] [ "test-quux.txt" resource-path delete-file ] unit-test
-[ ] [ "test-quux.txt" resource-path [ [ yield "Hi" write ] in-thread ] with-file-writer ] unit-test
+[ ] [ "test-quux.txt" resource-path [ [ yield "Hi" write ] "Test" spawn drop ] with-file-writer ] unit-test
[ ] [ "test-quux.txt" "quux-test.txt" [ resource-path ] 2apply rename-file ] unit-test
[ t ] [ "quux-test.txt" resource-path exists? ] unit-test
--- /dev/null
+! Copyright (C) 2008 Slava Pestov.\r
+! See http://factorcode.org/license.txt for BSD license.\r
+IN: io.thread\r
+USING: concurrency.threads io.backend namespaces init ;\r
+\r
+: io-thread ( -- )\r
+ sleep-time io-multiplex yield io-thread ;\r
+\r
+: start-io-thread ( -- )\r
+ [ io-thread ]\r
+ "I/O wait" spawn\r
+ \ io-thread set-global ;\r
+\r
+[ start-io-thread ] "io.thread" add-init-hook\r
{ $description "Appends a sequence to the end of the sequence being constructed by " { $link make } "." } ;
HELP: init-namespaces
-{ $description "Resets the name stack to its initial state, holding a single copy of the global namespace. This word is called during startup and is rarely useful, except in certain situations such as the example below." }
-{ $examples
- "You can use this word to spawn a new thread which does not inherit the parent thread's dynamic variable bindings:"
- { $code "[ init-namestack do-some-work ] in-thread" }
-} ;
+{ $description "Resets the name stack to its initial state, holding a single copy of the global namespace." }
+$low-level-note ;
+++ /dev/null
-Slava Pestov
+++ /dev/null
-Co-operative threading
+++ /dev/null
-USING: help.markup help.syntax kernel kernel.private io
-threads.private continuations dlists ;
-IN: threads
-
-ARTICLE: "threads" "Threads"
-"A limited form of multiprocessing is supported in the form of cooperative threads, which are implemented on top of continuations. A thread will yield while waiting for I/O operations to complete, or when a yield has been explicitly requested."
-$nl
-"Words for working with threads are in the " { $vocab-link "threads" } " vocabulary."
-{ $subsection in-thread }
-{ $subsection yield }
-{ $subsection sleep }
-"Threads stop either when the quotation given to " { $link in-thread } " returns, or when the following word is called:"
-{ $subsection stop }
-"Continuations can be added to the run queue directly:"
-{ $subsection schedule-thread }
-{ $subsection schedule-thread-with }
-"Thread implementation:"
-{ $subsection run-queue }
-{ $subsection sleep-queue } ;
-
-ABOUT: "threads"
-
-HELP: run-queue
-{ $values { "queue" dlist } }
-{ $description "Outputs the runnable thread queue. By convention, continuations are queued with " { $link push-front }
-" and dequeued with " { $link pop-back } "." } ;
-
-HELP: schedule-thread
-{ $values { "continuation" "a continuation reified by " { $link callcc0 } } }
-{ $description "Adds a runnable thread to the end of the run queue." } ;
-
-HELP: schedule-thread-with
-{ $values { "obj" "an object" } { "continuation" "a continuation reified by " { $link callcc1 } } }
-{ $description "Adds a runnable thread to the end of the run queue. When the thread runs the object is passed to the continuation using " { $link continue-with } "." } ;
-
-HELP: sleep-queue
-{ $var-description "Sleeping thread queue. This is not actually a queue, but an array of pairs of the shape " { $snippet "{ time continuation }" } "." } ;
-
-HELP: sleep-time
-{ $values { "ms" "a non-negative integer" } }
-{ $description "Outputs the time until the next sleeping thread is scheduled to wake up, or a default sleep time if there are no sleeping threads." } ;
-
-HELP: stop
-{ $description "Stops the current thread." } ;
-
-HELP: yield
-{ $description "Adds the current thread to the end of the run queue, and switches to the next runnable thread." } ;
-
-HELP: sleep
-{ $values { "ms" "a non-negative integer" } }
-{ $description "Suspends the current thread for " { $snippet "ms" } " milliseconds. It will not get woken up before this time period elapses, but since the multitasker is co-operative, the precise wakeup time is dependent on when other threads yield." } ;
-
-HELP: in-thread
-{ $values { "quot" "a quotation" } }
-{ $description "Spawns a new thread. The new thread begins running immediately."
-$nl
-"The new thread inherits the current data stack and name stack. The call stack initially contains the new quotation only, so when the quotation returns the thread stops. The catch stack contains a default handler which logs errors to the " { $link stdio } " stream." }
-{ $examples
- { $code "1 2 [ + . ] in-thread" }
-} ;
-
-HELP: idle-thread
-{ $description "Runs the idle thread, which services I/O requests and relinquishes control to the operating system until the next Factor thread has to wake up again."
-$nl
-"If the run queue is empty, the idle thread will sleep until the next sleeping thread is scheduled to wake up, otherwise it yields immediately after checking for any completed I/O requests." }
-{ $notes "This word should never be called directly. The idle thread is always running." } ;
-
-HELP: init-threads
-{ $description "Called during startup to initialize the threading system. This word should never be called directly." } ;
+++ /dev/null
-USING: namespaces io tools.test threads kernel ;
-IN: temporary
-
-3 "x" set
-[ yield 2 "x" set ] in-thread
-[ 2 ] [ yield "x" get ] unit-test
-[ ] [ [ flush ] in-thread flush ] unit-test
-[ ] [ [ "Errors, errors" throw ] in-thread ] unit-test
-yield
-
-[ ] [ 0.3 sleep ] unit-test
-[ "hey" sleep ] must-fail
+++ /dev/null
-! Copyright (C) 2004, 2007 Slava Pestov.
-! Copyright (C) 2005 Mackenzie Straight.
-! See http://factorcode.org/license.txt for BSD license.
-IN: threads
-USING: arrays init hashtables heaps io.backend kernel
-kernel.private math namespaces sequences vectors io system
-continuations debugger dlists ;
-
-<PRIVATE
-
-SYMBOL: sleep-queue
-
-: sleep-time ( -- ms )
- sleep-queue get-global dup heap-empty?
- [ drop 1000 ] [ heap-peek nip millis [-] ] if ;
-
-: run-queue ( -- queue ) \ run-queue get-global ;
-
-: schedule-sleep ( continuation ms -- )
- sleep-queue get-global heap-push ;
-
-: wake-up ( -- continuation )
- sleep-queue get-global heap-pop drop ;
-
-PRIVATE>
-
-: schedule-thread ( continuation -- )
- run-queue push-front ;
-
-: schedule-thread-with ( obj continuation -- )
- 2array schedule-thread ;
-
-: stop ( -- )
- walker-hook [
- continue
- ] [
- run-queue pop-back dup array?
- [ first2 continue-with ] [ continue ] if
- ] if* ;
-
-: yield ( -- ) [ schedule-thread stop ] callcc0 ;
-
-: sleep ( ms -- )
- >fixnum millis + [ schedule-sleep stop ] curry callcc0 ;
-
-: in-thread ( quot -- )
- [
- >r schedule-thread r> [
- V{ } set-catchstack
- { } set-retainstack
- [ [ print-error ] recover stop ] call-clear
- ] 1 (throw)
- ] curry callcc0 ;
-
-<PRIVATE
-
-: (idle-thread) ( slow? -- )
- sleep-time dup zero?
- [ wake-up schedule-thread 2drop ]
- [ 0 ? io-multiplex ] if ;
-
-: idle-thread ( -- )
- run-queue dlist-empty? (idle-thread) yield idle-thread ;
-
-: init-threads ( -- )
- <dlist> \ run-queue set-global
- <min-heap> sleep-queue set-global
- [ idle-thread ] in-thread ;
-
-[ init-threads ] "threads" add-init-hook
-PRIVATE>
"tools.annotations"
"tools.crossref"
"tools.deploy"
+ "tools.disassembler"
"tools.memory"
"tools.profiler"
"tools.test"
"tools.time"
- "tools.disassembler"
+ "tools.threads"
"editors"
} [ require ] each
! Copyright (C) 2008 Slava Pestov\r
! See http://factorcode.org/license.txt for BSD license.\r
-USING: calendar namespaces models threads init ;\r
+USING: calendar namespaces models concurrency.threads init ;\r
IN: calendar.model\r
\r
SYMBOL: time\r
now time get set-model\r
1000 sleep (time-thread) ;\r
\r
-: time-thread ( -- ) [ (time-thread) ] in-thread ;\r
+: time-thread ( -- )\r
+ [ (time-thread) ] "Time model update" spawn drop ;\r
\r
f <model> time set-global\r
[ time-thread ] "calendar.model" add-init-hook\r
+++ /dev/null
-Chris Double
+++ /dev/null
-! Copyright (C) 2006 Chris Double.
-! See http://factorcode.org/license.txt for BSD license.
-USING: help.syntax help.markup concurrency.private match ;
-IN: concurrency
-
-HELP: make-mailbox
-{ $values { "mailbox" "a mailbox object" }
-}
-{ $description "A mailbox is an object that can be used for safe thread communication. Items can be put in the mailbox and retrieved in a FIFO order. If the mailbox is empty when a get operation is performed then the thread will block until another thread places something in the mailbox. If multiple threads are waiting on the same mailbox, only one of the waiting threads will be unblocked to process the get operation." }
-{ $see-also mailbox-empty? mailbox-put mailbox-get mailbox-get-all while-mailbox-empty mailbox-get? } ;
-
-HELP: mailbox-empty?
-{ $values { "mailbox" "a mailbox object" }
- { "bool" "a boolean value" }
-}
-{ $description "Return true if the mailbox is empty." }
-{ $see-also make-mailbox mailbox-put mailbox-get mailbox-get-all while-mailbox-empty mailbox-get? } ;
-
-HELP: mailbox-put
-{ $values { "obj" "an object" }
- { "mailbox" "a mailbox object" }
-}
-{ $description "Put the object into the mailbox. Any threads that have a blocking get on the mailbox are resumed. Only one of those threads will successfully get the object, the rest will immediately block waiting for the next item in the mailbox." }
-{ $see-also make-mailbox mailbox-empty? mailbox-get mailbox-get-all while-mailbox-empty mailbox-get? } ;
-
-HELP: (mailbox-block-unless-pred)
-{ $values { "pred" "a quotation with stack effect " { $snippet "( X -- bool )" } }
- { "mailbox" "a mailbox object" }
- { "timeout" "a timeout in milliseconds" }
-}
-{ $description "Block the thread if there are no items in the mailbox that return true when the predicate is called with the item on the stack. The predicate must have stack effect " { $snippet "( X -- bool )" } "." }
-{ $see-also make-mailbox mailbox-empty? mailbox-put mailbox-get mailbox-get-all while-mailbox-empty mailbox-get? } ;
-
-HELP: (mailbox-block-if-empty)
-{ $values { "mailbox" "a mailbox object" }
- { "mailbox2" "same object as 'mailbox'" }
- { "timeout" "a timeout in milliseconds" }
-}
-{ $description "Block the thread if the mailbox is empty." }
-{ $see-also make-mailbox mailbox-empty? mailbox-put mailbox-get mailbox-get-all while-mailbox-empty mailbox-get? } ;
-
-HELP: mailbox-get
-{ $values { "mailbox" "a mailbox object" }
- { "obj" "an object" }
-}
-{ $description "Get the first item put into the mailbox. If it is empty the thread blocks until an item is put into it. The thread then resumes, leaving the item on the stack." }
-{ $see-also make-mailbox mailbox-empty? mailbox-put while-mailbox-empty mailbox-get-all mailbox-get? } ;
-
-HELP: mailbox-get-all
-{ $values { "mailbox" "a mailbox object" }
- { "array" "an array" }
-}
-{ $description "Blocks the thread if the mailbox is empty, otherwise removes all objects in the mailbox and returns an array containing the objects." }
-{ $see-also make-mailbox mailbox-empty? mailbox-put while-mailbox-empty mailbox-get-all mailbox-get? } ;
-
-HELP: while-mailbox-empty
-{ $values { "mailbox" "a mailbox object" }
- { "quot" "a quotation with stack effect " { $snippet "( -- )" } }
-}
-{ $description "Repeatedly call the quotation while there are no items in the mailbox. Quotation should have stack effect " { $snippet "( -- )" } "." }
-{ $see-also make-mailbox mailbox-empty? mailbox-put mailbox-get mailbox-get-all mailbox-get? } ;
-
-HELP: mailbox-get?
-{ $values { "pred" "a quotation with stack effect " { $snippet "( X -- bool )" } }
- { "mailbox" "a mailbox object" }
- { "obj" "an object" }
-}
-{ $description "Get the first item in the mailbox which satisfies the predicate. 'pred' will be called repeatedly for each item in the mailbox. When 'pred' returns true that item will be returned. If nothing in the mailbox satisfies the predicate then the thread will block until something does. 'pred' must have stack effect " { $snippet "( X -- bool }" } "." }
-{ $see-also make-mailbox mailbox-empty? mailbox-put mailbox-get mailbox-get-all while-mailbox-empty } ;
-
-HELP: <process>
-{ $values { "links" "an array of processes" }
- { "pid" "the process id" }
- { "mailbox" "a mailbox object" }
-}
-{ $description "Constructs a process object. A process is a lightweight thread with a mailbox that can be used to communicate with other processes. Each process has a unique process id." }
-{ $see-also spawn send receive } ;
-
-HELP: self
-{ $values { "process" "a process object" }
-}
-{ $description "Returns the currently running process object." }
-{ $see-also <process> send receive receive-if } ;
-
-HELP: send
-{ $values { "message" "an object" }
- { "process" "a process object" }
-}
-{ $description "Send the message to the process by placing it in the processes mailbox. This is an asynchronous operation and will return immediately. The receving process will act on the message the next time it retrieves that item from its mailbox (usually using the " { $link receive } " word. The message can be any Factor object. For destinations that are instances of remote-process the message must be a serializable Factor type." }
-{ $see-also <process> receive receive-if } ;
-
-HELP: receive
-{ $values { "message" "an object" }
-}
-{ $description "Return a message from the current processes mailbox. If the box is empty, suspend the process until another process places an item in the mailbox (usually via the " { $link send } " word." }
-{ $see-also send receive-if } ;
-
-HELP: receive-if
-{ $values { "pred" "a predicate with stack effect " { $snippet "( X -- bool )" } }
- { "message" "an object" }
-}
-{ $description "Return the first message from the current processes mailbox that satisfies the predicate. To satisfy the predicate, 'pred' is called with the item on the stack and the predicate should leave a boolean indicating whether it was satisfied or not. The predicate must have stack effect " { $snippet "( X -- bool )" } ". If nothing in the mailbox satisfies the predicate then the process will block until something does." }
-{ $see-also send receive } ;
-
-HELP: spawn
-{ $values { "quot" "a predicate with stack effect " { $snippet "( -- )" } }
- { "process" "a process object" }
-}
-{ $description "Start a process which runs the given quotation." }
-{ $see-also send receive receive-if self spawn-link } ;
-
-HELP: spawn-link
-{ $values { "quot" "a predicate with stack effect " { $snippet "( -- )" } }
- { "process" "a process object" }
-}
-{ $description "Start a process which runs the given quotation. If that quotation throws an error which is not caught then the error will get propagated to the process that spawned it. This can be used to set up 'supervisor' processes that restart child processes that crash due to uncaught errors.\n" }
-{ $see-also spawn } ;
-
-ARTICLE: { "concurrency" "loading" } "Loading"
-"The Factor module system can be used to load the Concurrency library:"
-{ $code "USING: concurrency ;" } ;
-
-ARTICLE: { "concurrency" "processes" } "Processes"
-"A process is basically a thread with a message queue. Other processes can place items on this queue by sending the process a message. A process can check its queue for messages, blocking if none are pending, and process them as they are queued.\n\nFactor processes are very lightweight. Each process can take as little as 900 bytes of memory. This library has been tested running hundreds of thousands of simple processes.\n\nThe messages that are sent from process to process are any Factor value. Factor tuples are ideal for this sort of thing as you can send a tuple to a process and the predicate dispatch mechanism can be used to perform actions depending on what the type of the tuple is.\n\nProcesses are usually created using " { $link spawn } ". This word takes a quotation on the stack and starts a process that will execute that quotation asynchronously. When the quotation completes the process will die. 'spawn' leaves on the stack the process object that was started. This object can be used to send messages to the process using " { $link send } ".\n\n'send' will return immediately after placing the message in the target processes message queue.\n\nA process can get a message from its queue using " { $link receive } ". This will get the most recent message and leave it on the stack. If there are no messages in the queue the process will 'block' until a message is available. When a process is blocked it takes no CPU time at all."
-{ $code "[ receive print ] spawn\n\"Hello Process!\" swap send" }
-"This example spawns a process that first blocks, waiting to receive a message. When a message is received, the 'receive' call returns leaving it on the stack. It then prints the message and exits. 'spawn' left the process on the stack so it's available to send the 'Hello Process!' message to it. Immediately after the 'send' you should see 'Hello Process!' printed on the console.\n\nIt is also possible to selectively retrieve messages from the message queue. " { $link receive-if } " takes a predicate quotation on the stack and returns the first message in the queue that satisfies the predicate. If no items satisfy the predicate then the process is blocked until a message is received that does."
-{ $code ": odd? ( n -- ? ) 2 mod 1 = ;\n1 self send 2 self send 3 self send\n\nreceive .\n => 1\n\n[ odd? ] receive-if .\n => 3\n\nreceive .\n => 2" } ;
-
-ARTICLE: { "concurrency" "self" } "Self"
-"A process can get access to its own process object using " { $link self } " so it can pass it to other processes. This allows the other processes to send messages back. A simple example of using this gets the current process' 'self' and spawns a process which sends a message to it. We then receive the message from the original process:"
-{ $code "self [ \"Hello!\" swap send ] spawn 2drop receive .\n => \"Hello!\"" } ;
-
-ARTICLE: { "concurrency" "servers" } "Servers"
-"A common idiom is to create 'server' processes that act on messages that are sent to it. These follow a basic pattern of blocking until a message is received, processing that message then looping back to blocking for a message.\n\nThe following example shows a very simple server that expects an array as its message. The first item of the array should be the senders process object. If the second item is 'ping' then the server sends 'pong' back to the caller. If the second item is anything else then the server exits:"
-{ $code ": pong-server ( -- )\n receive {\n { { ?from \"ping\" } [ \"pong\" ?from send pong-server ] }\n { { ?from _ } [ \"server shutdown\" ?from send ] }\n } match-cond ;\n\n[ pong-server ] spawn" }
-"Handling the deconstructing of messages and dispatching based on the message can be a bit of a chore. Especially in servers that take a number of different messages. The approach taken above is to use the 'match' library which allows easy deconstructing of messages using " { $link match-cond } "." ;
-
-ARTICLE: { "concurrency" "synchronous-sends" } "Synchronous Sends"
-{ $link send } " sends a message asynchronously, and the sending process continues immediately. The 'pong server' example shown previously all sent messages to the server and waited for a reply back from the server. This pattern of synchronous sending is made easier with " { $link send-synchronous } ".\n\nThis word will send a message to the given process and immediately block until a reply is received for this particular message send. It leaves the reply on the stack. Note that it doesn't wait for just any reply, it waits for a reply specifically to this send.\n\nTo do this it wraps the requested message inside a tagged message format using " { $link tag-message } ":"
-{ $code "\"My Message\" tag-message .\n => { ...from... ...tag... \"My Message\" }" }
-"The message is wrapped in array where the first item is the sending process object, the second is a unique tag, and the third is the original message. Server processes can use the 'from' to reply to the process that originally sent the message. The tag is used in the receiving server to include the value in the reply. After the send-synchronous call the current process will block waiting for a reply that has the exact same tag. In this way you can be sure that the reply you got was for the specific message sent. Here is the pong-server recoded to use 'send-synchronous':"
-{ $code ": pong-server ( -- )\n receive {\n { { ?from ?tag \"ping\" } [ ?tag \"pong\" 2array ?from send pong-server ] }\n { { ?from _ } [ ?tag \"server shutdown\" 2array ?from send ] }\n } match-cond ;\n\n[ pong-server ] spawn \"ping\" swap send-synchronous .\n => \"pong\"" }
-"Notice that the code to send the reply back to the original caller wraps the reply in an array where the first item is the tag originally sent. 'send-synchronous' only returns if it receives a reply containing that specific tag." ;
-
-ARTICLE: { "concurrency" "exceptions" } "Exceptions"
-"A process can handle exceptions using the standard Factor exception handling mechanism. If an exception is uncaught the process will terminate. For example:"
-{ $code "[ 1 0 / \"This will not print\" print ] spawn" }
-"Processes can be linked so that a parent process can receive the exception that caused the child process to terminate. In this way 'supervisor' processes can be created that are notified when child processes terminate and possibly restart them.\n\nThe easiest way to form this link is using " { $link spawn-link } ". This will create a unidirectional link, such that if an uncaught exception causes the child to terminate, the parent process can catch it:"
-{ $code "[\n [ 1 0 / \"This will not print\" print ] spawn-link drop\n receive\n] [ \"Exception caught.\" print ] recover" }
-"Exceptions are only raised in the parent when the parent does a " { $link receive } " or " { $link receive-if } ". This is because the exception is sent from the child to the parent as a message." ;
-
-ARTICLE: { "concurrency" "futures" } "Futures"
-"A future is a placeholder for the result of a computation that is being calculated in a process. When the process has completed the computation the future can be queried to find out the result. If the computation has not completed when the future is queried them the process will block until the result is completed. A future is created using " { $link future } ".\n\nThe quotation will be run in a spawned process, and a future object is immediately returned. This future object can be resolved using " { $link ?future } ".\n\nFutures are useful for starting calculations that take a long time to run but aren't needed until later in the process. When the process needs the value it can use '?future' to get the result or block until the result is available. For example:"
-{ $code "[ 30 fib ] future\n...do stuff...\n?future" } ;
-
-ARTICLE: { "concurrency" "promises" } "Promises"
-"A promise is similar to a future but it is not produced by calculating something in the background. It represents a promise to provide a value sometime later. A process can request the value of a promise and will block if the promise is not fulfilled. Later, another process can fulfill the promise, providing a value. All threads waiting on the promise will then resume with that value on the stack. Use " { $link <promise> } " to create a promise, " { $link fulfill } " to set it to a value, and " { $link ?promise } " to retrieve the value, or block until the promise is fulfilled:"
-{ $code "<promise>\n[ ?promise \"Promise fulfilled: \" write print ] spawn drop\n[ ?promise \"Promise fulfilled: \" write print ] spawn drop\n[ ?promise \"Promise fulfilled: \" write print ] spawn drop\n\"hello\" swap fulfill\n => Promise fulfilled: hello\n Promise fulfilled: hello\n Promise fulfilled: hello" } ;
-
-ARTICLE: { "concurrency" "concurrency" } "Concurrency"
-"The concurrency library is based upon the style of concurrency used in systems like Erlang and Termite. It is built on top of the standard Factor lightweight thread system.\nA concurrency oriented program is one in which multiple processes run simultaneously in a single Factor image or across multiple running Factor instances. The processes can communicate with each other by asynchronous message sends. Although processes can share data via Factor's mutable data structures it is not recommended as the use of shared state concurrency is often a cause of problems."
-{ $subsection { "concurrency" "loading" } }
-{ $subsection { "concurrency" "processes" } }
-{ $subsection { "concurrency" "self" } }
-{ $subsection { "concurrency" "servers" } }
-{ $subsection { "concurrency" "synchronous-sends" } }
-{ $subsection { "concurrency" "exceptions" } }
-{ $subsection { "concurrency" "futures" } }
-{ $subsection { "concurrency" "promises" } } ;
-
-ABOUT: { "concurrency" "concurrency" }
+++ /dev/null
-! Copyright (C) 2005 Chris Double. All Rights Reserved.
-! See http://factorcode.org/license.txt for BSD license.
-!
-USING: kernel concurrency threads vectors arrays sequences
-namespaces tools.test continuations dlists strings math words
-match quotations concurrency.private ;
-IN: temporary
-
-[ ] [ self process-mailbox mailbox-data dlist-delete-all ] unit-test
-
-[ V{ 1 2 3 } ] [
- 0 <vector>
- make-mailbox
- 2dup [ mailbox-get swap push ] 2curry in-thread
- 2dup [ mailbox-get swap push ] 2curry in-thread
- 2dup [ mailbox-get swap push ] 2curry in-thread
- 1 over mailbox-put
- 2 over mailbox-put
- 3 swap mailbox-put
-] unit-test
-
-[ V{ 1 2 3 } ] [
- 0 <vector>
- make-mailbox
- 2dup [ [ integer? ] swap mailbox-get? swap push ] 2curry in-thread
- 2dup [ [ integer? ] swap mailbox-get? swap push ] 2curry in-thread
- 2dup [ [ integer? ] swap mailbox-get? swap push ] 2curry in-thread
- 1 over mailbox-put
- 2 over mailbox-put
- 3 swap mailbox-put
-] unit-test
-
-[ V{ 1 "junk" 3 "junk2" } [ 456 ] ] [
- 0 <vector>
- make-mailbox
- 2dup [ [ integer? ] swap mailbox-get? swap push ] 2curry in-thread
- 2dup [ [ integer? ] swap mailbox-get? swap push ] 2curry in-thread
- 2dup [ [ string? ] swap mailbox-get? swap push ] 2curry in-thread
- 2dup [ [ string? ] swap mailbox-get? swap push ] 2curry in-thread
- 1 over mailbox-put
- "junk" over mailbox-put
- [ 456 ] over mailbox-put
- 3 over mailbox-put
- "junk2" over mailbox-put
- mailbox-get
-] unit-test
-
-[ "test" ] [
- [ self ] "test" with-process
-] unit-test
-
-
-[ "received" ] [
- [
- receive {
- { { ?from ?tag _ } [ ?tag "received" 2array ?from send ] }
- } match-cond
- ] spawn
- "sent" swap send-synchronous
-] unit-test
-
-[ 1 3 2 ] [
- 1 self send
- 2 self send
- 3 self send
- receive
- [ 2 mod 0 = not ] receive-if
- receive
-] unit-test
-
-
-[
- [
- "crash" throw
- ] spawn-link drop
- receive
-] [ "crash" = ] must-fail-with
-
-[ 50 ] [
- [ 50 ] future ?future
-] unit-test
-
-[ V{ 50 50 50 } ] [
- 0 <vector>
- <promise>
- 2dup [ ?promise swap push ] 2curry spawn drop
- 2dup [ ?promise swap push ] 2curry spawn drop
- 2dup [ ?promise swap push ] 2curry spawn drop
- 50 swap fulfill
-] unit-test
-
-MATCH-VARS: ?value ;
-SYMBOL: increment
-SYMBOL: decrement
-SYMBOL: value
-
-: counter ( value -- )
- receive {
- { { increment ?value } [ ?value + counter ] }
- { { decrement ?value } [ ?value - counter ] }
- { { value ?from } [ dup ?from send counter ] }
- } match-cond ;
-
-[ -5 ] [
- [ 0 counter ] spawn
- { increment 10 } over send
- { decrement 15 } over send
- [ value , self , ] { } make swap send
- receive
-] unit-test
-
-! The following unit test blocks forever if the
-! exception does not propogate. Uncomment when
-! this is fixed (via a timeout).
-[
- [ "this should propogate" throw ] future ?future
-] must-fail
-
-[ ] [
- [ "this should not propogate" throw ] future drop
-] unit-test
-
-[ f ] [
- [ 1 drop ] spawn 100 sleep process-pid get-process
-] unit-test
-
-[ f ] [
- [ "testing unregistering on error" throw ] spawn
- 100 sleep process-pid get-process
-] unit-test
-
-! Race condition with futures
-[ 3 3 ] [
- [ 3 ] future
- dup ?future swap ?future
-] unit-test
-
-! Another race
-[ 3 ] [
- [ 3 yield ] future ?future
-] unit-test
\ No newline at end of file
+++ /dev/null
-! Copyright (C) 2005 Chris Double. All Rights Reserved.
-! See http://factorcode.org/license.txt for BSD license.
-!
-! Concurrency library for Factor based on Erlang/Termite style
-! concurrency.
-USING: vectors dlists threads sequences continuations
- namespaces random math quotations words kernel match
- arrays io assocs init shuffle system ;
-IN: concurrency
-
-TUPLE: mailbox threads data ;
-
-TUPLE: thread timeout continuation continued? ;
-
-: <thread> ( timeout continuation -- obj )
- >r dup [ millis + ] when r>
- {
- set-thread-timeout
- set-thread-continuation
- } thread construct ;
-
-: make-mailbox ( -- mailbox )
- V{ } clone <dlist> mailbox construct-boa ;
-
-: mailbox-empty? ( mailbox -- bool )
- mailbox-data dlist-empty? ;
-
-: mailbox-put ( obj mailbox -- )
- [ mailbox-data push-back ] keep
- [ mailbox-threads ] keep
- V{ } clone swap set-mailbox-threads
- [ thread-continuation schedule-thread ] each yield ;
-
-<PRIVATE
-: (mailbox-block-unless-pred) ( pred mailbox timeout -- )
- 2over mailbox-data dlist-contains? [
- 3drop
- ] [
- [ <thread> swap mailbox-threads push stop ] callcc0
- (mailbox-block-unless-pred)
- ] if ; inline
-
-: (mailbox-block-if-empty) ( mailbox timeout -- mailbox2 )
- over mailbox-empty? [
- [ <thread> swap mailbox-threads push stop ] callcc0
- (mailbox-block-if-empty)
- ] [
- drop
- ] if ;
-PRIVATE>
-: mailbox-get* ( mailbox timeout -- obj )
- (mailbox-block-if-empty)
- mailbox-data pop-front ;
-
-: mailbox-get ( mailbox -- obj )
- f mailbox-get* ;
-
-: mailbox-get-all* ( mailbox timeout -- array )
- (mailbox-block-if-empty)
- [ dup mailbox-empty? ]
- [ dup mailbox-data pop-front ]
- [ ] unfold nip ;
-
-: mailbox-get-all ( mailbox -- array )
- f mailbox-get-all* ;
-
-: while-mailbox-empty ( mailbox quot -- )
- over mailbox-empty? [
- dup >r swap slip r> while-mailbox-empty
- ] [
- 2drop
- ] if ; inline
-
-: mailbox-get?* ( pred mailbox timeout -- obj )
- 2over >r >r (mailbox-block-unless-pred) r> r>
- mailbox-data delete-node-if ; inline
-
-: mailbox-get? ( pred mailbox -- obj )
- f mailbox-get?* ;
-
-TUPLE: process links pid mailbox ;
-
-C: <process> process
-
-GENERIC: send ( message process -- )
-
-<PRIVATE
-: make-process ( -- process )
- #! Return a process set to run on the local node. A process is
- #! similar to a thread but can send and receive messages to and
- #! from other processes. It may also be linked to other processes so
- #! that it receives a message if that process terminates.
- [ ] random-256 make-mailbox <process> ;
-
-: make-linked-process ( process -- process )
- #! Return a process set to run on the local node. That process is
- #! linked to the process on the stack. It will receive a message if
- #! that process terminates.
- 1quotation random-256 make-mailbox <process> ;
-PRIVATE>
-
-: self ( -- process )
- \ self get ;
-
-<PRIVATE
-: init-main-process ( -- )
- #! Setup the main process.
- make-process \ self set-global ;
-
-: with-process ( quot process -- )
- #! Calls the quotation with 'self' set
- #! to the given process.
- \ self rot with-variable ; inline
-
-PRIVATE>
-
-DEFER: register-process
-DEFER: unregister-process
-
-<PRIVATE
-: ((spawn)) ( quot -- )
- self dup process-pid swap register-process
- [ self process-pid unregister-process ] [ ] cleanup ; inline
-
-: (spawn) ( quot -- process )
- [ in-thread ] make-process [ with-process ] keep ; inline
-
-PRIVATE>
-
-: spawn ( quot -- process )
- [ ((spawn)) ] curry (spawn) ; inline
-
-TUPLE: linked-exception error ;
-
-C: <linked-exception> linked-exception
-
-: while-no-messages ( quot -- )
- #! Run the quotation in a loop while no messages are in
- #! the processes mailbox. The quot should have stack effect
- #! ( -- ).
- >r self process-mailbox r> while-mailbox-empty ; inline
-
-M: process send ( message process -- )
- process-mailbox mailbox-put ;
-
-: receive ( -- message )
- self process-mailbox mailbox-get dup linked-exception? [
- linked-exception-error rethrow
- ] when ;
-
-: receive-if ( pred -- message )
- self process-mailbox mailbox-get? dup linked-exception? [
- linked-exception-error rethrow
- ] when ; inline
-
-: rethrow-linked ( error -- )
- #! Rethrow the error to the linked process
- self process-links [
- over <linked-exception> swap send
- ] each drop ;
-
-<PRIVATE
-: (spawn-link) ( quot -- process )
- [ in-thread ] self make-linked-process
- [ with-process ] keep ; inline
-PRIVATE>
-
-: spawn-link ( quot -- process )
- [ [ rethrow-linked ] recover ] curry
- [ ((spawn)) ] curry (spawn-link) ; inline
-
-<PRIVATE
-: (recv) ( msg form -- )
- #! Process a form with the following format:
- #! [ pred match-quot ]
- #! 'pred' is a word that has stack effect ( msg -- bool ). It is
- #! executed with the message on the stack. It should return a
- #! boolean if it is a message this form should process.
- #! 'match-quot' is a quotation with stack effect ( msg -- ). It
- #! will be called with the message on the top of the stack if
- #! the 'pred' word returned true.
- [ first execute ] 2keep rot [ second call ] [ 2drop ] if ;
-PRIVATE>
-
-: recv ( forms -- )
- #! Get a message from the processes mailbox. Compare it against the
- #! forms to run a quotation if it matches the given message. 'forms'
- #! is a list of quotations in the following format:
- #! [ pred match-quot ]
- #! 'pred' is a word that has stack effect ( msg -- bool ). It is
- #! executed with the message on the stack. It should return a
- #! boolean if it is a message this form should process.
- #! 'match-quot' is a quotation with stack effect ( msg -- ). It
- #! will be called with the message on the top of the stack if
- #! the 'pred' word returned true.
- #! Each form in the list will be matched against the message,
- #! even if a prior match succeeded. This means multiple quotations
- #! may be run against the message.
- receive swap [ dupd (recv) ] each drop ;
-
-MATCH-VARS: ?from ?tag ;
-
-<PRIVATE
-: tag-message ( message -- tagged-message )
- #! Given a message, wrap it with the sending process and a unique tag.
- >r self random-256 r> 3array ;
-PRIVATE>
-
-: send-synchronous ( message process -- reply )
- #! Sends a message to the process synchronously. The
- #! message will be wrapped to include the process of the sender
- #! and a unique tag. After being sent the sending process will
- #! block for a reply tagged with the same unique tag.
- >r tag-message dup r> send second _ 2array [ match ] curry
- receive-if second ;
-
-<PRIVATE
-: forever ( quot -- )
- #! Loops forever executing the quotation.
- dup slip forever ;
-
-SYMBOL: quit-cc
-
-: (spawn-server) ( quot -- )
- #! Receive a message, and run 'quot' on it. If 'quot'
- #! returns true, start again, otherwise exit loop.
- #! The quotation should have stack effect ( message -- bool ).
- "Waiting for message in server: " write
- self process-pid print
- receive over call [ (spawn-server) ] when ;
-PRIVATE>
-
-: spawn-server ( quot -- process )
- #! Spawn a server that receives messages, calling the
- #! quotation on the message. If the quotation returns false
- #! the spawned process exits. If it returns true, the process
- #! starts from the beginning again. The quotation should have
- #! stack effect ( message -- bool ).
- [
- (spawn-server)
- "Exiting process: " write self process-pid print
- ] curry spawn ; inline
-
-: spawn-linked-server ( quot -- process )
- #! Similar to 'spawn-server' but the parent process will be linked
- #! to the child.
- [
- (spawn-server)
- "Exiting process: " write self process-pid print
- ] curry spawn-link ; inline
-
-: server-cc ( -- cc|process )
- #! Captures the current continuation and returns the value.
- #! If that CC is called with a process on the stack it will
- #! set 'self' for the current process to it. Otherwise it will
- #! return the value. This allows capturing a continuation in a server,
- #! and jumping back into it from a spawn and keeping the 'self'
- #! variable correct. It's a workaround until I can find out how to
- #! stop 'self' from being clobbered back to its old value.
- [ ] callcc1 dup process? [ \ self set-global f ] when ;
-
-: call-server-cc ( server-cc -- )
- #! Calls the server continuation passing the current 'self'
- #! so the server continuation gets its new self updated.
- self swap call ;
-
-TUPLE: future value processes ;
-
-: notify-future ( value future -- )
- tuck set-future-value
- dup future-processes [ schedule-thread ] each
- f swap set-future-processes ;
-
-: future ( quot -- future )
- #! Spawn a process to call the quotation and immediately return.
- f V{ } clone \ future construct-boa [
- [
- >r [ t 2array ] compose [ f 2array ] recover r>
- notify-future
- ] 2curry spawn drop
- ] keep ;
-
-: ?future ( future -- result )
- #! Block the process until the future has completed and then
- #! place the result on the stack. Return the result
- #! immediately if the future has completed.
- dup future-value [
- first2 [ rethrow ] unless
- ] [
- dup [ future-processes push stop ] curry callcc0 ?future
- ] ?if ;
-
-: parallel-map ( seq quot -- newseq )
- #! Spawn a process to apply quot to each element of seq,
- #! joining the results into a sequence at the end.
- [ curry future ] curry map [ ?future ] map ;
-
-: parallel-each ( seq quot -- )
- #! Spawn a process to apply quot to each element of seq,
- #! and waits for all processes to complete.
- [ f ] compose parallel-map drop ;
-
-TUPLE: promise fulfilled? value processes ;
-
-: <promise> ( -- <promise> )
- f f V{ } clone promise construct-boa ;
-
-: fulfill ( value promise -- )
- #! Set the future of the promise to the given value. Threads
- #! blocking on the promise will then be released.
- dup promise-fulfilled? [
- 2drop
- ] [
- [ set-promise-value ] keep
- [ t swap set-promise-fulfilled? ] keep
- [ promise-processes ] keep
- V{ } clone swap set-promise-processes
- [ thread-continuation schedule-thread ] each yield
- ] if ;
-
-<PRIVATE
- : (maybe-block-promise) ( promise timeout -- promise )
- #! Block the process if the promise is unfulfilled. This is different from
- #! (mailbox-block-if-empty) in that when a promise is fulfilled, all threads
- #! need to be resumed, rather than just one.
- over promise-fulfilled? [
- drop
- ] [
- [ <thread> swap promise-processes push stop ] callcc0
- drop
- ] if ;
-PRIVATE>
-
-: ?promise* ( promise timeout -- result )
- (maybe-block-promise) promise-value ;
-
-: ?promise ( promise -- result )
- f ?promise* ;
-
-! ******************************
-! Experimental code below
-! ******************************
-<PRIVATE
-: (lazy) ( v -- )
- receive {
- { { ?from ?tag _ }
- [ ?tag over 2array ?from send (lazy) ] }
- } match-cond ;
-PRIVATE>
-
-: lazy ( quot -- lazy )
- #! Spawn a process that immediately blocks and return it.
- #! When '?lazy' is called on the returned process, call the quotation
- #! and return the result. The quotation must have stack effect ( -- X ).
- [
- receive {
- { { ?from ?tag _ }
- [ call ?tag over 2array ?from send (lazy) ] }
- } match-cond
- ] spawn nip ;
-
-: ?lazy ( lazy -- result )
- #! Given a process spawned using 'lazy', evaluate it and return the result.
- f swap send-synchronous ;
-
-<PRIVATE
-: remote-processes ( -- hash )
- \ remote-processes get-global ;
-PRIVATE>
-
-: register-process ( name process -- )
- swap remote-processes set-at ;
-
-: unregister-process ( name -- )
- remote-processes delete-at ;
-
-: get-process ( name -- process )
- remote-processes at ;
-
-[
- H{ } clone \ remote-processes set-global
- init-main-process
- self [ process-pid ] keep register-process
-] "process-registry" add-init-hook
! Copyright (C) 2005 Chris Double. All Rights Reserved.
! See http://factorcode.org/license.txt for BSD license.
-USING: serialize sequences concurrency io io.server qualified
-threads arrays namespaces kernel ;
+USING: serialize sequences concurrency.messaging
+concurrency.threads io io.server qualified arrays
+namespaces kernel ;
QUALIFIED: io.sockets
IN: concurrency.distributed
-TUPLE: node hostname port ;
-
-C: <node> node
-
: handle-node-client ( -- )
- deserialize first2 get-process send ;
-
-: node-server ( port -- )
- internet-server
- "concurrency.distributed"
- [ handle-node-client ] with-server ;
-
-: send-to-node ( msg pid host port -- )
- io.sockets:<inet> io.sockets:<client> [
- 2array serialize
- ] with-stream ;
+ deserialize first2 thread send ;
-: localnode ( -- node )
- \ localnode get ;
+: (start-node) ( addrspecs addrspec -- )
+ [
+ local-node set-global
+ "concurrency.distributed"
+ [ handle-node-client ] with-server
+ ] 2curry f spawn drop ;
-: start-node ( hostname port -- )
- [ node-server ] in-thread
- <node> \ localnode set-global ;
+SYMBOL: local-node ( -- addrspec )
-TUPLE: remote-process node pid ;
+: start-node ( port -- )
+ dup internet-server host-name rot <inet> (start-node) ;
-C: <remote-process> remote-process
+TUPLE: remote-thread pid node ;
-M: remote-process send ( message process -- )
- #! Send the message via the inter-node protocol
- { remote-process-pid remote-process-node } get-slots
- { node-hostname node-port } get-slots
- send-to-node ;
+M: remote-thread send ( message thread -- )
+ { remote-thread-pid remote-thread-node } get-slots
+ io.sockets:<client> [ 2array serialize ] with-stream ;
-M: process (serialize) ( obj -- )
- localnode swap process-pid <remote-process> (serialize) ;
+M: thread (serialize) ( obj -- )
+ thread-id local-node get-global
+ remote-thread construct-boa
+ (serialize) ;
--- /dev/null
+Slava Pestov
--- /dev/null
+! Copyright (C) 2008 Slava Pestov.\r
+! See http://factorcode.org/license.txt for BSD license.\r
+USING: kernel concurrency.threads ;\r
+IN: concurrency.exchangers\r
+\r
+! Motivated by\r
+! http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/Exchanger.html\r
+\r
+TUPLE: exchanger thread ;\r
+\r
+: <exchanger> ( -- exchanger )\r
+ f exchanger construct-boa ;\r
+\r
+: exchange ( obj exchanger -- newobj )\r
+ dup exchanger-thread [\r
+ dup exchanger-thread\r
+ f rot set-exchanger-thread\r
+ resume-with\r
+ ] [\r
+ [ set-exchanger-thread ] curry suspend\r
+ ] if ;\r
--- /dev/null
+Thread rendezvous points
--- /dev/null
+Chris Double
+Slava Pestov
--- /dev/null
+! Copyright (C) 2005, 2008 Chris Double, Slava Pestov.\r
+! See http://factorcode.org/license.txt for BSD license.\r
+IN: concurrency.futures\r
+\r
+: future ( quot -- future )\r
+ <promise> [\r
+ [\r
+ >r\r
+ [ t 2array ] compose\r
+ [ <linked> f 2array ] recover\r
+ r> fulfill\r
+ ] 2curry "Future" spawn drop\r
+ ] keep ; inline\r
+\r
+: ?future-timeout ( future timeout -- value )\r
+ ?promise-timeout first2 [ rethrow ] unless ;\r
+\r
+: ?future ( future -- value )\r
+ f ?future-timeout ;\r
+\r
+: parallel-map ( seq quot -- newseq )\r
+ [ curry future ] curry map [ ?future ] map ;\r
+\r
+: parallel-each ( seq quot -- )\r
+ [ f ] compose parallel-map drop ;\r
--- /dev/null
+Deferred computations
--- /dev/null
+Slava Pestov
--- /dev/null
+! Copyright (C) 2008 Slava Pestov.\r
+! See http://factorcode.org/license.txt for BSD license.\r
+USING: dlists kernel concurrency.threads continuations math ;\r
+IN: concurrency.locks\r
+\r
+! Simple critical sections\r
+TUPLE: lock threads owner ;\r
+\r
+: <lock> <dlist> lock construct-boa ;\r
+\r
+: notify-1 ( dlist -- )\r
+ dup dlist-empty? [ pop-back resume ] [ drop ] if ;\r
+\r
+<PRIVATE\r
+\r
+: wait-for-lock ( lock -- )\r
+ [ swap lock-threads push-front ] suspend drop ;\r
+\r
+: acquire-lock ( lock -- )\r
+ dup lock-owner [ wait-for-lock ] when\r
+ self swap set-lock-owner ;\r
+\r
+: release-lock ( lock -- )\r
+ f over set-lock-owner\r
+ lock-threads notify-1 ;\r
+\r
+: do-lock ( lock quot acquire release -- )\r
+ >r >r over r> call over r> curry [ ] cleanup ; inline\r
+\r
+PRIVATE>\r
+\r
+: with-lock ( lock quot -- )\r
+ [ acquire-lock ] [ release-lock ] do-lock ; inline\r
+\r
+: with-reentrant-lock ( lock quot -- )\r
+ over lock-owner self eq?\r
+ [ nip call ] [ with-lock ] if ; inline\r
+\r
+! Many-reader/single-writer locks\r
+TUPLE: rw-lock readers writers reader# writer ;\r
+\r
+: <rw-lock> ( -- lock )\r
+ <dlist> <dlist> 0 f rw-lock construct-boa ;\r
+\r
+<PRIVATE\r
+\r
+: wait-for-read-lock ( lock -- )\r
+ [ swap lock-readers push-front ] suspend drop ;\r
+\r
+: acquire-read-lock ( lock -- )\r
+ dup rw-lock-writer [ dup wait-for-read-lock ] when\r
+ dup rw-lock-reader# 1+ swap set-rw-lock-reader# ;\r
+\r
+: notify-writer ( lock -- )\r
+ lock-writers notify-1 ;\r
+\r
+: release-read-lock ( lock -- )\r
+ dup rw-lock-reader# 1- dup pick set-rw-lock-reader#\r
+ zero? [ notify-writers ] [ drop ] if ;\r
+\r
+: wait-for-write-lock ( lock -- )\r
+ [ swap lock-writers push-front ] suspend drop ;\r
+\r
+: acquire-write-lock ( lock -- )\r
+ dup rw-lock-writer over rw-lock-reader# 0 > or\r
+ [ dup wait-for-write-lock ] when\r
+ self over set-rw-lock-writer ;\r
+\r
+: release-write-lock ( lock -- )\r
+ f over set-rw-lock-writer\r
+ dup rw-lock-readers dlist-empty?\r
+ [ notify-writer ] [ rw-lock-readers notify-all ] if ;\r
+\r
+: do-recursive-rw-lock ( lock quot quot' -- )\r
+ >r over rw-lock-writer self eq? [ nip call ] r> if ; inline\r
+\r
+PRIVATE>\r
+\r
+: with-read-lock ( lock quot -- )\r
+ [\r
+ [ acquire-read-lock ] [ release-read-lock ] do-lock\r
+ ] do-recursive-rw-lock ; inline\r
+\r
+: with-write-lock ( lock quot -- )\r
+ [\r
+ [ acquire-write-lock ] [ release-write-lock ] do-lock\r
+ ] do-recursive-rw-lock ; inline\r
--- /dev/null
+Traditional locks and many reader/single writer locks
--- /dev/null
+Chris Double
+Slava Pestov
--- /dev/null
+! Copyright (C) 2006 Chris Double.
+! See http://factorcode.org/license.txt for BSD license.
+USING: help.syntax help.markup concurrency.private match ;
+IN: concurrency
+
+HELP: make-mailbox
+{ $values { "mailbox" "a mailbox object" }
+}
+{ $description "A mailbox is an object that can be used for safe thread communication. Items can be put in the mailbox and retrieved in a FIFO order. If the mailbox is empty when a get operation is performed then the thread will block until another thread places something in the mailbox. If multiple threads are waiting on the same mailbox, only one of the waiting threads will be unblocked to process the get operation." }
+{ $see-also mailbox-empty? mailbox-put mailbox-get mailbox-get-all while-mailbox-empty mailbox-get? } ;
+
+HELP: mailbox-empty?
+{ $values { "mailbox" "a mailbox object" }
+ { "bool" "a boolean value" }
+}
+{ $description "Return true if the mailbox is empty." }
+{ $see-also make-mailbox mailbox-put mailbox-get mailbox-get-all while-mailbox-empty mailbox-get? } ;
+
+HELP: mailbox-put
+{ $values { "obj" "an object" }
+ { "mailbox" "a mailbox object" }
+}
+{ $description "Put the object into the mailbox. Any threads that have a blocking get on the mailbox are resumed. Only one of those threads will successfully get the object, the rest will immediately block waiting for the next item in the mailbox." }
+{ $see-also make-mailbox mailbox-empty? mailbox-get mailbox-get-all while-mailbox-empty mailbox-get? } ;
+
+HELP: (mailbox-block-unless-pred)
+{ $values { "pred" "a quotation with stack effect " { $snippet "( X -- bool )" } }
+ { "mailbox" "a mailbox object" }
+ { "timeout" "a timeout in milliseconds" }
+}
+{ $description "Block the thread if there are no items in the mailbox that return true when the predicate is called with the item on the stack. The predicate must have stack effect " { $snippet "( X -- bool )" } "." }
+{ $see-also make-mailbox mailbox-empty? mailbox-put mailbox-get mailbox-get-all while-mailbox-empty mailbox-get? } ;
+
+HELP: (mailbox-block-if-empty)
+{ $values { "mailbox" "a mailbox object" }
+ { "mailbox2" "same object as 'mailbox'" }
+ { "timeout" "a timeout in milliseconds" }
+}
+{ $description "Block the thread if the mailbox is empty." }
+{ $see-also make-mailbox mailbox-empty? mailbox-put mailbox-get mailbox-get-all while-mailbox-empty mailbox-get? } ;
+
+HELP: mailbox-get
+{ $values { "mailbox" "a mailbox object" }
+ { "obj" "an object" }
+}
+{ $description "Get the first item put into the mailbox. If it is empty the thread blocks until an item is put into it. The thread then resumes, leaving the item on the stack." }
+{ $see-also make-mailbox mailbox-empty? mailbox-put while-mailbox-empty mailbox-get-all mailbox-get? } ;
+
+HELP: mailbox-get-all
+{ $values { "mailbox" "a mailbox object" }
+ { "array" "an array" }
+}
+{ $description "Blocks the thread if the mailbox is empty, otherwise removes all objects in the mailbox and returns an array containing the objects." }
+{ $see-also make-mailbox mailbox-empty? mailbox-put while-mailbox-empty mailbox-get-all mailbox-get? } ;
+
+HELP: while-mailbox-empty
+{ $values { "mailbox" "a mailbox object" }
+ { "quot" "a quotation with stack effect " { $snippet "( -- )" } }
+}
+{ $description "Repeatedly call the quotation while there are no items in the mailbox. Quotation should have stack effect " { $snippet "( -- )" } "." }
+{ $see-also make-mailbox mailbox-empty? mailbox-put mailbox-get mailbox-get-all mailbox-get? } ;
+
+HELP: mailbox-get?
+{ $values { "pred" "a quotation with stack effect " { $snippet "( X -- bool )" } }
+ { "mailbox" "a mailbox object" }
+ { "obj" "an object" }
+}
+{ $description "Get the first item in the mailbox which satisfies the predicate. 'pred' will be called repeatedly for each item in the mailbox. When 'pred' returns true that item will be returned. If nothing in the mailbox satisfies the predicate then the thread will block until something does. 'pred' must have stack effect " { $snippet "( X -- bool }" } "." }
+{ $see-also make-mailbox mailbox-empty? mailbox-put mailbox-get mailbox-get-all while-mailbox-empty } ;
+
+HELP: <process>
+{ $values { "links" "an array of processes" }
+ { "pid" "the process id" }
+ { "mailbox" "a mailbox object" }
+}
+{ $description "Constructs a process object. A process is a lightweight thread with a mailbox that can be used to communicate with other processes. Each process has a unique process id." }
+{ $see-also spawn send receive } ;
+
+HELP: self
+{ $values { "process" "a process object" }
+}
+{ $description "Returns the currently running process object." }
+{ $see-also <process> send receive receive-if } ;
+
+HELP: send
+{ $values { "message" "an object" }
+ { "process" "a process object" }
+}
+{ $description "Send the message to the process by placing it in the processes mailbox. This is an asynchronous operation and will return immediately. The receving process will act on the message the next time it retrieves that item from its mailbox (usually using the " { $link receive } " word. The message can be any Factor object. For destinations that are instances of remote-process the message must be a serializable Factor type." }
+{ $see-also <process> receive receive-if } ;
+
+HELP: receive
+{ $values { "message" "an object" }
+}
+{ $description "Return a message from the current processes mailbox. If the box is empty, suspend the process until another process places an item in the mailbox (usually via the " { $link send } " word." }
+{ $see-also send receive-if } ;
+
+HELP: receive-if
+{ $values { "pred" "a predicate with stack effect " { $snippet "( X -- bool )" } }
+ { "message" "an object" }
+}
+{ $description "Return the first message from the current processes mailbox that satisfies the predicate. To satisfy the predicate, 'pred' is called with the item on the stack and the predicate should leave a boolean indicating whether it was satisfied or not. The predicate must have stack effect " { $snippet "( X -- bool )" } ". If nothing in the mailbox satisfies the predicate then the process will block until something does." }
+{ $see-also send receive } ;
+
+HELP: spawn
+{ $values { "quot" "a predicate with stack effect " { $snippet "( -- )" } }
+ { "process" "a process object" }
+}
+{ $description "Start a process which runs the given quotation." }
+{ $see-also send receive receive-if self spawn-link } ;
+
+HELP: spawn-link
+{ $values { "quot" "a predicate with stack effect " { $snippet "( -- )" } }
+ { "process" "a process object" }
+}
+{ $description "Start a process which runs the given quotation. If that quotation throws an error which is not caught then the error will get propagated to the process that spawned it. This can be used to set up 'supervisor' processes that restart child processes that crash due to uncaught errors.\n" }
+{ $see-also spawn } ;
+
+ARTICLE: { "concurrency" "loading" } "Loading"
+"The Factor module system can be used to load the Concurrency library:"
+{ $code "USING: concurrency ;" } ;
+
+ARTICLE: { "concurrency" "processes" } "Processes"
+"A process is basically a thread with a message queue. Other processes can place items on this queue by sending the process a message. A process can check its queue for messages, blocking if none are pending, and process them as they are queued.\n\nFactor processes are very lightweight. Each process can take as little as 900 bytes of memory. This library has been tested running hundreds of thousands of simple processes.\n\nThe messages that are sent from process to process are any Factor value. Factor tuples are ideal for this sort of thing as you can send a tuple to a process and the predicate dispatch mechanism can be used to perform actions depending on what the type of the tuple is.\n\nProcesses are usually created using " { $link spawn } ". This word takes a quotation on the stack and starts a process that will execute that quotation asynchronously. When the quotation completes the process will die. 'spawn' leaves on the stack the process object that was started. This object can be used to send messages to the process using " { $link send } ".\n\n'send' will return immediately after placing the message in the target processes message queue.\n\nA process can get a message from its queue using " { $link receive } ". This will get the most recent message and leave it on the stack. If there are no messages in the queue the process will 'block' until a message is available. When a process is blocked it takes no CPU time at all."
+{ $code "[ receive print ] spawn\n\"Hello Process!\" swap send" }
+"This example spawns a process that first blocks, waiting to receive a message. When a message is received, the 'receive' call returns leaving it on the stack. It then prints the message and exits. 'spawn' left the process on the stack so it's available to send the 'Hello Process!' message to it. Immediately after the 'send' you should see 'Hello Process!' printed on the console.\n\nIt is also possible to selectively retrieve messages from the message queue. " { $link receive-if } " takes a predicate quotation on the stack and returns the first message in the queue that satisfies the predicate. If no items satisfy the predicate then the process is blocked until a message is received that does."
+{ $code ": odd? ( n -- ? ) 2 mod 1 = ;\n1 self send 2 self send 3 self send\n\nreceive .\n => 1\n\n[ odd? ] receive-if .\n => 3\n\nreceive .\n => 2" } ;
+
+ARTICLE: { "concurrency" "self" } "Self"
+"A process can get access to its own process object using " { $link self } " so it can pass it to other processes. This allows the other processes to send messages back. A simple example of using this gets the current process' 'self' and spawns a process which sends a message to it. We then receive the message from the original process:"
+{ $code "self [ \"Hello!\" swap send ] spawn 2drop receive .\n => \"Hello!\"" } ;
+
+ARTICLE: { "concurrency" "servers" } "Servers"
+"A common idiom is to create 'server' processes that act on messages that are sent to it. These follow a basic pattern of blocking until a message is received, processing that message then looping back to blocking for a message.\n\nThe following example shows a very simple server that expects an array as its message. The first item of the array should be the senders process object. If the second item is 'ping' then the server sends 'pong' back to the caller. If the second item is anything else then the server exits:"
+{ $code ": pong-server ( -- )\n receive {\n { { ?from \"ping\" } [ \"pong\" ?from send pong-server ] }\n { { ?from _ } [ \"server shutdown\" ?from send ] }\n } match-cond ;\n\n[ pong-server ] spawn" }
+"Handling the deconstructing of messages and dispatching based on the message can be a bit of a chore. Especially in servers that take a number of different messages. The approach taken above is to use the 'match' library which allows easy deconstructing of messages using " { $link match-cond } "." ;
+
+ARTICLE: { "concurrency" "synchronous-sends" } "Synchronous Sends"
+{ $link send } " sends a message asynchronously, and the sending process continues immediately. The 'pong server' example shown previously all sent messages to the server and waited for a reply back from the server. This pattern of synchronous sending is made easier with " { $link send-synchronous } ".\n\nThis word will send a message to the given process and immediately block until a reply is received for this particular message send. It leaves the reply on the stack. Note that it doesn't wait for just any reply, it waits for a reply specifically to this send.\n\nTo do this it wraps the requested message inside a tagged message format using " { $link tag-message } ":"
+{ $code "\"My Message\" tag-message .\n => { ...from... ...tag... \"My Message\" }" }
+"The message is wrapped in array where the first item is the sending process object, the second is a unique tag, and the third is the original message. Server processes can use the 'from' to reply to the process that originally sent the message. The tag is used in the receiving server to include the value in the reply. After the send-synchronous call the current process will block waiting for a reply that has the exact same tag. In this way you can be sure that the reply you got was for the specific message sent. Here is the pong-server recoded to use 'send-synchronous':"
+{ $code ": pong-server ( -- )\n receive {\n { { ?from ?tag \"ping\" } [ ?tag \"pong\" 2array ?from send pong-server ] }\n { { ?from _ } [ ?tag \"server shutdown\" 2array ?from send ] }\n } match-cond ;\n\n[ pong-server ] spawn \"ping\" swap send-synchronous .\n => \"pong\"" }
+"Notice that the code to send the reply back to the original caller wraps the reply in an array where the first item is the tag originally sent. 'send-synchronous' only returns if it receives a reply containing that specific tag." ;
+
+ARTICLE: { "concurrency" "exceptions" } "Exceptions"
+"A process can handle exceptions using the standard Factor exception handling mechanism. If an exception is uncaught the process will terminate. For example:"
+{ $code "[ 1 0 / \"This will not print\" print ] spawn" }
+"Processes can be linked so that a parent process can receive the exception that caused the child process to terminate. In this way 'supervisor' processes can be created that are notified when child processes terminate and possibly restart them.\n\nThe easiest way to form this link is using " { $link spawn-link } ". This will create a unidirectional link, such that if an uncaught exception causes the child to terminate, the parent process can catch it:"
+{ $code "[\n [ 1 0 / \"This will not print\" print ] spawn-link drop\n receive\n] [ \"Exception caught.\" print ] recover" }
+"Exceptions are only raised in the parent when the parent does a " { $link receive } " or " { $link receive-if } ". This is because the exception is sent from the child to the parent as a message." ;
+
+ARTICLE: { "concurrency" "futures" } "Futures"
+"A future is a placeholder for the result of a computation that is being calculated in a process. When the process has completed the computation the future can be queried to find out the result. If the computation has not completed when the future is queried them the process will block until the result is completed. A future is created using " { $link future } ".\n\nThe quotation will be run in a spawned process, and a future object is immediately returned. This future object can be resolved using " { $link ?future } ".\n\nFutures are useful for starting calculations that take a long time to run but aren't needed until later in the process. When the process needs the value it can use '?future' to get the result or block until the result is available. For example:"
+{ $code "[ 30 fib ] future\n...do stuff...\n?future" } ;
+
+ARTICLE: { "concurrency" "promises" } "Promises"
+"A promise is similar to a future but it is not produced by calculating something in the background. It represents a promise to provide a value sometime later. A process can request the value of a promise and will block if the promise is not fulfilled. Later, another process can fulfill the promise, providing a value. All threads waiting on the promise will then resume with that value on the stack. Use " { $link <promise> } " to create a promise, " { $link fulfill } " to set it to a value, and " { $link ?promise } " to retrieve the value, or block until the promise is fulfilled:"
+{ $code "<promise>\n[ ?promise \"Promise fulfilled: \" write print ] spawn drop\n[ ?promise \"Promise fulfilled: \" write print ] spawn drop\n[ ?promise \"Promise fulfilled: \" write print ] spawn drop\n\"hello\" swap fulfill\n => Promise fulfilled: hello\n Promise fulfilled: hello\n Promise fulfilled: hello" } ;
+
+ARTICLE: { "concurrency" "concurrency" } "Concurrency"
+"The concurrency library is based upon the style of concurrency used in systems like Erlang and Termite. It is built on top of the standard Factor lightweight thread system.\nA concurrency oriented program is one in which multiple processes run simultaneously in a single Factor image or across multiple running Factor instances. The processes can communicate with each other by asynchronous message sends. Although processes can share data via Factor's mutable data structures it is not recommended as the use of shared state concurrency is often a cause of problems."
+{ $subsection { "concurrency" "loading" } }
+{ $subsection { "concurrency" "processes" } }
+{ $subsection { "concurrency" "self" } }
+{ $subsection { "concurrency" "servers" } }
+{ $subsection { "concurrency" "synchronous-sends" } }
+{ $subsection { "concurrency" "exceptions" } }
+{ $subsection { "concurrency" "futures" } }
+{ $subsection { "concurrency" "promises" } } ;
+
+ABOUT: { "concurrency" "concurrency" }
--- /dev/null
+! Copyright (C) 2005 Chris Double. All Rights Reserved.
+! See http://factorcode.org/license.txt for BSD license.
+!
+USING: kernel concurrency threads vectors arrays sequences
+namespaces tools.test continuations dlists strings math words
+match quotations concurrency.private ;
+IN: temporary
+
+[ ] [ self process-mailbox mailbox-data dlist-delete-all ] unit-test
+
+[ V{ 1 2 3 } ] [
+ 0 <vector>
+ make-mailbox
+ 2dup [ mailbox-get swap push ] 2curry in-thread
+ 2dup [ mailbox-get swap push ] 2curry in-thread
+ 2dup [ mailbox-get swap push ] 2curry in-thread
+ 1 over mailbox-put
+ 2 over mailbox-put
+ 3 swap mailbox-put
+] unit-test
+
+[ V{ 1 2 3 } ] [
+ 0 <vector>
+ make-mailbox
+ 2dup [ [ integer? ] swap mailbox-get? swap push ] 2curry in-thread
+ 2dup [ [ integer? ] swap mailbox-get? swap push ] 2curry in-thread
+ 2dup [ [ integer? ] swap mailbox-get? swap push ] 2curry in-thread
+ 1 over mailbox-put
+ 2 over mailbox-put
+ 3 swap mailbox-put
+] unit-test
+
+[ V{ 1 "junk" 3 "junk2" } [ 456 ] ] [
+ 0 <vector>
+ make-mailbox
+ 2dup [ [ integer? ] swap mailbox-get? swap push ] 2curry in-thread
+ 2dup [ [ integer? ] swap mailbox-get? swap push ] 2curry in-thread
+ 2dup [ [ string? ] swap mailbox-get? swap push ] 2curry in-thread
+ 2dup [ [ string? ] swap mailbox-get? swap push ] 2curry in-thread
+ 1 over mailbox-put
+ "junk" over mailbox-put
+ [ 456 ] over mailbox-put
+ 3 over mailbox-put
+ "junk2" over mailbox-put
+ mailbox-get
+] unit-test
+
+[ "test" ] [
+ [ self ] "test" with-process
+] unit-test
+
+
+[ "received" ] [
+ [
+ receive {
+ { { ?from ?tag _ } [ ?tag "received" 2array ?from send ] }
+ } match-cond
+ ] spawn
+ "sent" swap send-synchronous
+] unit-test
+
+[ 1 3 2 ] [
+ 1 self send
+ 2 self send
+ 3 self send
+ receive
+ [ 2 mod 0 = not ] receive-if
+ receive
+] unit-test
+
+
+[
+ [
+ "crash" throw
+ ] spawn-link drop
+ receive
+] [ "crash" = ] must-fail-with
+
+[ 50 ] [
+ [ 50 ] future ?future
+] unit-test
+
+[ V{ 50 50 50 } ] [
+ 0 <vector>
+ <promise>
+ 2dup [ ?promise swap push ] 2curry spawn drop
+ 2dup [ ?promise swap push ] 2curry spawn drop
+ 2dup [ ?promise swap push ] 2curry spawn drop
+ 50 swap fulfill
+] unit-test
+
+MATCH-VARS: ?value ;
+SYMBOL: increment
+SYMBOL: decrement
+SYMBOL: value
+
+: counter ( value -- )
+ receive {
+ { { increment ?value } [ ?value + counter ] }
+ { { decrement ?value } [ ?value - counter ] }
+ { { value ?from } [ dup ?from send counter ] }
+ } match-cond ;
+
+[ -5 ] [
+ [ 0 counter ] spawn
+ { increment 10 } over send
+ { decrement 15 } over send
+ [ value , self , ] { } make swap send
+ receive
+] unit-test
+
+! The following unit test blocks forever if the
+! exception does not propogate. Uncomment when
+! this is fixed (via a timeout).
+[
+ [ "this should propogate" throw ] future ?future
+] must-fail
+
+[ ] [
+ [ "this should not propogate" throw ] future drop
+] unit-test
+
+[ f ] [
+ [ 1 drop ] spawn 100 sleep process-pid get-process
+] unit-test
+
+[ f ] [
+ [ "testing unregistering on error" throw ] spawn
+ 100 sleep process-pid get-process
+] unit-test
+
+! Race condition with futures
+[ 3 3 ] [
+ [ 3 ] future
+ dup ?future swap ?future
+] unit-test
+
+! Another race
+[ 3 ] [
+ [ 3 yield ] future ?future
+] unit-test
\ No newline at end of file
--- /dev/null
+! Copyright (C) 2005, 2008 Chris Double, Slava Pestov.\r
+! See http://factorcode.org/license.txt for BSD license.\r
+!\r
+! Concurrency library for Factor based on Erlang/Termite style\r
+! concurrency.\r
+IN: concurrency.messaging\r
+USING: dlists concurrency.threads sequences continuations\r
+namespaces random math quotations words kernel arrays assocs\r
+init system ;\r
+\r
+TUPLE: mailbox threads data ;\r
+\r
+: <mailbox> ( -- mailbox )\r
+ <dlist> <dlist> mailbox construct-boa ;\r
+\r
+: mailbox-empty? ( mailbox -- bool )\r
+ mailbox-data dlist-empty? ;\r
+\r
+: notify-all ( dlist -- )\r
+ [ second resume ] dlist-slurp yield ;\r
+\r
+: mailbox-put ( obj mailbox -- )\r
+ [ mailbox-data push-front ] keep\r
+ mailbox-threads notify-all ;\r
+\r
+<PRIVATE\r
+\r
+: mailbox-wait ( mailbox timeout -- mailbox timeout )\r
+ [ 2array swap mailbox-threads push-front ] suspend drop ;\r
+ inline\r
+\r
+: block-unless-pred ( pred mailbox timeout -- )\r
+ 2over mailbox-data dlist-contains? [\r
+ 3drop\r
+ ] [\r
+ mailbox-wait block-unless-pred\r
+ ] if ; inline\r
+\r
+: block-if-empty ( mailbox timeout -- mailbox )\r
+ over mailbox-empty? [\r
+ mailbox-wait block-if-empty\r
+ ] [\r
+ drop\r
+ ] if ;\r
+\r
+PRIVATE>\r
+\r
+: mailbox-peek ( mailbox -- obj )\r
+ mailbox-data peek-front ;\r
+\r
+: mailbox-get-timeout ( mailbox timeout -- obj )\r
+ block-if-empty mailbox-data pop-front ;\r
+\r
+: mailbox-get ( mailbox -- obj )\r
+ f mailbox-timeout-get ;\r
+\r
+: mailbox-get-all-timeout ( mailbox timeout -- array )\r
+ (mailbox-block-if-empty)\r
+ [ dup mailbox-empty? ]\r
+ [ dup mailbox-data pop-back ]\r
+ [ ] unfold nip ;\r
+\r
+: mailbox-get-all ( mailbox -- array )\r
+ f mailbox-timeout-get-all ;\r
+\r
+: while-mailbox-empty ( mailbox quot -- )\r
+ over mailbox-empty? [\r
+ dup >r swap slip r> while-mailbox-empty\r
+ ] [\r
+ 2drop\r
+ ] if ; inline\r
+\r
+: mailbox-timeout-get? ( pred mailbox timeout -- obj )\r
+ [ (mailbox-block-unless-pred) ] 3keep drop\r
+ mailbox-data delete-node-if ; inline\r
+\r
+: mailbox-get? ( pred mailbox -- obj )\r
+ f mailbox-timeout-get? ;\r
+\r
+TUPLE: linked error thread ;\r
+\r
+: <linked> self linked construct-boa ;\r
+\r
+GENERIC: send ( message thread -- )\r
+\r
+M: thread send ( message thread -- )\r
+ thread-mailbox mailbox-put ;\r
+\r
+: ?linked dup linked? [ rethrow ] when ;\r
+\r
+: mailbox self thread-mailbox ;\r
+\r
+: receive ( -- message )\r
+ mailbox mailbox-get ?linked ;\r
+\r
+: receive-if ( pred -- message )\r
+ mailbox mailbox-get? ?linked ; inline\r
+\r
+: rethrow-linked ( error supervisor -- )\r
+ >r <linked> r> send ;\r
+\r
+: spawn-linked ( quot name -- thread )\r
+ self [ rethrow-linked ] curry <thread> [ (spawn) ] keep ;\r
+\r
+TUPLE: synchronous data sender tag ;\r
+\r
+: <synchronous> ( data -- sync )\r
+ self random-256 synchronous construct-boa ;\r
+\r
+TUPLE: reply data tag ;\r
+\r
+: <reply> ( data synchronous -- reply )\r
+ synchronous-tag \ reply construct-boa ;\r
+\r
+: send-synchronous ( message thread -- reply )\r
+ >r <synchronous> dup r> send\r
+ [ over reply? [ reply-tag = ] [ 2drop f ] if ] curry\r
+ receive-if reply-data ;\r
+\r
+: reply-synchronous ( message synchronous -- )\r
+ [ <reply> ] keep synchronous-sender reply ;\r
--- /dev/null
+Erlang/Termite-style message-passing concurrency
--- /dev/null
+Chris Double
+Slava Pestov
--- /dev/null
+! Copyright (C) 2005, 2008 Chris Double, Slava Pestov.\r
+! See http://factorcode.org/license.txt for BSD license.\r
+IN: concurrency.promises\r
+\r
+TUPLE: promise mailbox ;\r
+\r
+: <promise> ( -- promise )\r
+ <mailbox> promise construct-boa ;\r
+\r
+: promise-fulfilled? ( promise -- ? )\r
+ promise-mailbox mailbox-empty? not ;\r
+\r
+: fulfill ( value promise -- )\r
+ dup promise-fulfilled? [ \r
+ "Promise already fulfilled" throw\r
+ ] [\r
+ promise-mailbox mailbox-put\r
+ ] if ;\r
+\r
+: ?promise-timeout ( promise timeout -- result )\r
+ >r promise-mailbox r> block-if-empty mailbox-peek ;\r
+\r
+: ?promise ( promise -- result )\r
+ f ?promise-timeout ;\r
--- /dev/null
+Thread-safe write-once variables
--- /dev/null
+Slava Pestov
--- /dev/null
+IN: concurrency.semaphores\r
+\r
+TUPLE: semaphore count threads ;\r
+\r
+: <semaphore> ( -- semaphore )\r
+ 0 <dlist> semaphore construct-boa ;\r
+\r
+: wait-to-acquire ( semaphore -- )\r
+ [ semaphore-threads push-front ] suspend drop ;\r
+\r
+: acquire ( semaphore -- )\r
+ dup semaphore-count zero? [\r
+ wait-to-acquire\r
+ ] [\r
+ dup semaphore-count 1- swap set-semaphore-count\r
+ ] if ;\r
+\r
+: release ( semaphore -- )\r
+ dup semaphore-count 1+ over set-semaphore-count\r
+ semaphore-threads notify-1 ;\r
--- /dev/null
+Counting semaphores
+++ /dev/null
-Erlang-style concurrency
+++ /dev/null
-extensions
! See http://factorcode.org/license.txt for BSD license.
USING: io io.backend io.timeouts system kernel namespaces
strings hashtables sequences assocs combinators vocabs.loader
-init threads continuations math ;
+init concurrency.threads continuations math ;
IN: io.launcher
! Non-blocking process exit notification facility
: wait-for-process ( process -- status )
[
dup process-handle
- [ dup [ processes get at push stop ] curry callcc0 ] when
+ [ dup [ processes get at push ] curry suspend drop ] when
dup process-killed?
[ "Process was killed" throw ] [ process-status ] if
] with-timeout ;
: notify-exit ( status process -- )
[ set-process-status ] keep
- [ processes get delete-at* drop [ schedule-thread ] each ] keep
+ [ processes get delete-at* drop [ resume ] each ] keep
f swap set-process-handle ;
! Copyright (C) 2008 Slava Pestov.\r
! See http://factorcode.org/license.txt for BSD license.\r
USING: io.backend kernel continuations namespaces sequences\r
-assocs hashtables sorting arrays threads ;\r
+assocs hashtables sorting arrays concurrency.threads ;\r
IN: io.monitors\r
\r
<PRIVATE\r
: notify-callback ( simple-monitor -- )\r
dup simple-monitor-callback\r
f rot set-simple-monitor-callback\r
- [ schedule-thread ] when* ;\r
+ [ resume ] when* ;\r
\r
M: simple-monitor fill-queue ( monitor -- )\r
dup simple-monitor-callback [\r
"Cannot wait for changes on the same file from multiple threads" throw\r
] when\r
- [ swap set-simple-monitor-callback stop ] callcc0\r
+ [ swap set-simple-monitor-callback ] suspend drop\r
check-monitor ;\r
\r
M: simple-monitor dispose ( monitor -- )\r
-USING: help.markup help.syntax io io.backend threads
+USING: help.markup help.syntax io io.backend concurrency.threads
strings byte-arrays continuations ;
IN: io.sockets
! Copyright (C) 2008 Slava Pestov, Doug Coleman\r
! See http://factorcode.org/license.txt for BSD license.\r
-USING: kernel math system dlists namespaces assocs init threads\r
-io.streams.duplex ;\r
+USING: kernel math system dlists namespaces assocs init\r
+concurrency.threads io.streams.duplex ;\r
IN: io.timeouts\r
\r
TUPLE: lapse entry timeout cutoff ;\r
: expiry-thread ( -- )\r
expire-timeouts 5000 sleep expiry-thread ;\r
\r
-[ [ expiry-thread ] in-thread ] "io.timeouts" add-init-hook\r
+: start-expiry-thread ( -- )\r
+ [ expiry-thread ] "I/O expiry" spawn drop ;\r
+\r
+[ start-expiry-thread ] "io.timeouts" add-init-hook\r
2dup check-io-task fd/container set-at ;
: add-io-task ( task -- )
- mx get-global register-io-task stop ;
+ mx get-global register-io-task ;
: with-port-continuation ( port quot -- port )
- [ callcc0 ] curry with-timeout ; inline
+ [ suspend drop ] curry with-timeout ; inline
M: mx unregister-io-task ( task mx -- )
fd/container delete-at drop ;
: pop-callbacks ( mx task -- )
dup rot unregister-io-task
- io-task-callbacks [ schedule-thread ] each ;
+ io-task-callbacks [ resume ] each ;
: handle-io-task ( mx task -- )
dup do-io-task [ pop-callbacks ] [ 2drop ] if ;
: add-write-io-task ( port continuation -- )
over port-handle mx get-global mx-writes at*
- [ io-task-callbacks push stop ]
+ [ io-task-callbacks push ]
[ drop <write-task> add-io-task ] if ;
: (wait-to-write) ( port -- )
wait-for-processes [ 250 sleep ] when wait-loop ;
: start-wait-thread ( -- )
- [ wait-loop ] in-thread ;
+ [ wait-loop ] "Process reaper" spawn drop ;
] with-stream
"unix-domain-socket-test" resource-path delete-file
-] in-thread
+] "Test" spawn drop
yield
"unix-domain-datagram-test" resource-path delete-file
] with-scope
-] in-thread
+] "Test" spawn drop
yield
io.windows io.windows.nt.pipes libc io.nonblocking
io.streams.duplex windows.types math windows.kernel32 windows
namespaces io.launcher kernel sequences windows.errors assocs
-splitting system threads init strings combinators io.backend ;
+splitting system concurrency.threads init strings combinators
+io.backend ;
IN: io.windows.launcher
TUPLE: CreateProcess-args
wait-loop ;
: start-wait-thread ( -- )
- [ wait-loop ] in-thread ;
+ [ wait-loop ] "Process wait" spawn drop ;
[ start-wait-thread ] "io.windows.launcher" add-init-hook
USING: alien alien.c-types arrays assocs combinators
continuations destructors io io.backend io.nonblocking
io.windows libc kernel math namespaces sequences
-threads tuples.lib windows windows.errors windows.kernel32
-strings splitting io.files qualified ascii combinators.lib ;
+concurrency.threads tuples.lib windows windows.errors
+windows.kernel32 strings splitting io.files qualified ascii
+combinators.lib ;
QUALIFIED: windows.winsock
IN: io.windows.nt.backend
SYMBOL: io-hash
-TUPLE: io-callback port continuation ;
+TUPLE: io-callback port thread ;
C: <io-callback> io-callback
[
<io-callback> swap
dup alien? [ "bad overlapped in save-callback" throw ] unless
- io-hash get-global set-at stop
- ] callcc0 2drop ;
+ io-hash get-global set-at
+ ] suspend 3drop ;
: wait-for-overlapped ( ms -- overlapped ? )
>r master-completion-port get-global r> ! port ms
] [
(win32-error-string) swap lookup-callback
[ io-callback-port set-port-error ] keep
- ] if io-callback-continuation schedule-thread f
+ ] if io-callback-thread resume f
] if
] [
lookup-callback
- io-callback-continuation schedule-thread f
+ io-callback-thread resume f
] if ;
: drain-overlapped ( timeout -- )
USING: continuations destructors io.buffers io.files io.backend
io.timeouts io.nonblocking io.windows io.windows.nt.backend
-kernel libc math threads windows windows.kernel32 alien.c-types
-alien.arrays sequences combinators combinators.lib sequences.lib
-ascii splitting alien strings assocs ;
+kernel libc math concurrency.threads windows windows.kernel32
+alien.c-types alien.arrays sequences combinators combinators.lib
+sequences.lib ascii splitting alien strings assocs ;
IN: io.windows.nt.files
M: windows-nt-io cwd
USE: io.backend
T{ windows-nt-io } set-io-backend
+
+"vocabs.monitor" require
continuations destructors io.nonblocking io.timeouts io.sockets
io.sockets.impl io namespaces io.streams.duplex io.windows
io.windows.nt.backend windows.winsock kernel libc math sequences
-threads tuples.lib ;
+concurrency.threads tuples.lib ;
IN: io.windows.nt.sockets
: malloc-int ( object -- object )
USING: arrays assocs classes combinators sequences.private
continuations continuations.private generic hashtables io kernel
kernel.private math namespaces namespaces.private prettyprint
-quotations sequences splitting strings threads vectors words ;
+quotations sequences splitting strings concurrency.threads
+vectors words ;
IN: tools.interpreter
: walk ( quot -- ) \ break add* call ;
--- /dev/null
+! Copyright (C) 2008 Slava Pestov.\r
+! See http://factorcode.org/license.txt for BSD license.\r
+IN: tools.threads\r
+USING: concurrency.threads kernel prettyprint prettyprint.config\r
+io io.styles sequences assocs namespaces sorting ;\r
+\r
+: thread. ( thread -- )\r
+ dup thread-id pprint-cell\r
+ dup thread-name pprint-cell\r
+ thread-continuation "Waiting" "Running" ? [ write ] with-cell ;\r
+\r
+: threads. ( -- )\r
+ standard-table-style [\r
+ [\r
+ { "ID" "Name" "State" }\r
+ [ [ write ] with-cell ] each\r
+ ] with-row\r
+\r
+ threads get-global >alist sort-keys values [\r
+ [ thread. ] with-row\r
+ ] each\r
+ ] tabular-output ;\r
USING: arrays assocs combinators continuations documents
ui.tools.workspace hashtables io io.styles kernel math
math.vectors models namespaces parser prettyprint quotations
-sequences sequences.lib strings threads listener tuples
-ui.commands ui.gadgets ui.gadgets.editors
+sequences sequences.lib strings concurrency.threads listener
+tuples ui.commands ui.gadgets ui.gadgets.editors
ui.gadgets.presentations ui.gadgets.worlds ui.gestures
definitions ;
IN: ui.tools.interactor
TUPLE: interactor
history output
-continuation quot busy?
+thread quot busy?
help ;
: interactor-use ( interactor -- seq )
use swap
- interactor-continuation continuation-name
+ interactor-thread thread-continuation continuation-name
assoc-stack ;
: init-caret-help ( interactor -- )
dup dup interactor-help add-connection
f swap set-interactor-busy? ;
-M: interactor ungraft*
- dup dup interactor-help remove-connection
- delegate ungraft* ;
-
: word-at-loc ( loc interactor -- word )
over [
[ gadget-model T{ one-word-elt } elt-string ] keep
: interactor-continue ( obj interactor -- )
t over set-interactor-busy?
- interactor-continuation schedule-thread-with ;
+ interactor-thread resume-with ;
: clear-input ( interactor -- ) gadget-model clear-doc ;
: evaluate-input ( interactor -- )
dup interactor-busy? [
- [
- [ control-value ] keep interactor-continue
- ] in-thread
+ dup control-value over interactor-continue
] unless drop ;
: interactor-yield ( interactor -- obj )
- f over set-interactor-busy?
- [ set-interactor-continuation stop ] curry callcc1 ;
+ dup gadget-graft-state first [
+ f over set-interactor-busy?
+ [ set-interactor-thread ] curry suspend
+ ] [
+ drop f
+ ] if ;
M: interactor stream-readln
[ interactor-yield ] keep interactor-finish ?first ;
ui.gadgets ui.gadgets.editors ui.gadgets.labelled
ui.gadgets.panes ui.gadgets.buttons ui.gadgets.scrollers
ui.gadgets.tracks ui.gestures ui.operations vocabs words
-prettyprint listener debugger threads ;
+prettyprint listener debugger concurrency.threads ;
IN: ui.tools.listener
TUPLE: listener-gadget input output stack ;
] with-stream* ;
: restart-listener ( listener -- )
- [ >r clear r> init-namespaces listener-thread ] in-thread
- drop ;
+ [ listener-thread ] curry "Listener" spawn drop ;
: init-listener ( listener -- )
f <model> swap set-listener-gadget-stack ;
! See http://factorcode.org/license.txt for BSD license.
USING: arrays assocs ui.tools.listener ui.tools.traceback
ui.tools.workspace inspector kernel models namespaces
-prettyprint quotations sequences threads tools.interpreter
-ui.commands ui.gadgets ui.gadgets.labelled ui.gadgets.tracks
-ui.gestures ui.gadgets.buttons ui.gadgets.panes
-prettyprint.config prettyprint.backend continuations ;
+prettyprint quotations sequences concurrency.threads
+tools.interpreter ui.commands ui.gadgets ui.gadgets.labelled
+ui.gadgets.tracks ui.gestures ui.gadgets.buttons
+ui.gadgets.panes prettyprint.config prettyprint.backend
+continuations ;
IN: ui.tools.walker
TUPLE: walker model interpreter history ;
! Copyright (C) 2006, 2007 Slava Pestov.
! See http://factorcode.org/license.txt for BSD license.
USING: arrays assocs io kernel math models namespaces
-prettyprint dlists sequences threads sequences words timers
-debugger ui.gadgets ui.gadgets.worlds ui.gadgets.tracks
-ui.gestures ui.backend ui.render continuations init
-combinators hashtables ;
+prettyprint dlists sequences concurrency.threads sequences words
+timers debugger ui.gadgets ui.gadgets.worlds ui.gadgets.tracks
+ui.gestures ui.backend ui.render continuations init combinators
+hashtables ;
IN: ui
! Assoc mapping aliens to gadgets
ui.backend ui.clipboards ui.gadgets.worlds ui.gestures io kernel
math math.vectors namespaces prettyprint sequences strings
vectors words windows.kernel32 windows.gdi32 windows.user32
-windows.opengl32 windows.messages windows.types
-windows.nt windows threads timers libc combinators continuations
-command-line shuffle opengl ui.render unicode.case ascii
-math.bitfields ;
+windows.opengl32 windows.messages windows.types windows.nt
+windows concurrency.threads timers libc combinators
+continuations command-line shuffle opengl ui.render unicode.case
+ascii math.bitfields ;
IN: ui.windows
TUPLE: windows-ui-backend ;
-USING: threads io.files io.monitors init kernel tools.browser\r
-continuations ;\r
+USING: concurrency.threads io.files io.monitors init kernel\r
+tools.browser ;\r
IN: vocabs.monitor\r
\r
! Use file system change monitoring to flush the tags/authors\r
! cache\r
-: update-thread ( monitor -- )\r
- dup next-change 2drop reset-cache update-thread ;\r
+: (monitor-thread) ( monitor -- )\r
+ dup next-change 2drop reset-cache (monitor-thread) ;\r
\r
-: start-update-thread\r
+: monitor-thread ( -- )\r
+ "" resource-path t <monitor> (monitor-thread) ;\r
+\r
+: start-monitor-thread\r
#! Silently ignore errors during monitor creation since\r
#! monitors are not supported on all platforms.\r
- [\r
- [ "" resource-path t <monitor> ] [ drop f ] recover\r
- [ update-thread ] when*\r
- ] in-thread ;\r
+ [ monitor-thread ] "Vocabulary monitor" spawn drop ;\r
\r
-[ start-update-thread ] "tools.browser" add-init-hook\r
+[ start-monitor-thread ] "vocabs.monitor" add-init-hook\r
: update-thread ( -- )
millis last-update set-global
- [ update-cached-postings ] in-thread
+ [ update-cached-postings ] "RSS feed update slave" spawn drop
10 60 * 1000 * sleep
update-thread ;
"webapps.planet" [
update-thread
] with-logging
- ] in-thread ;
+ ] "RSS feed update master" spawn drop ;
"planet" "planet-factor" "extra/webapps/planet" web-app
-#define USER_ENV 40
+#define USER_ENV 64
typedef enum {
NAMESTACK_ENV, /* used by library only */
STDERR_ENV = 38, /* stderr FILE* handle */
- STAGE2_ENV = 39 /* have we bootstrapped? */
+ STAGE2_ENV = 39, /* have we bootstrapped? */
+
+ CURRENT_THREAD_ENV = 40
} F_ENVTYPE;
#define FIRST_SAVE_ENV BOOT_ENV
projects / factor.git / commitdiff