factor/basis/concurrency/mailboxes/mailboxes.factor

96 lines
2.7 KiB
Factor
Executable File

! Copyright (C) 2005, 2010 Chris Double, Slava Pestov.
! See http://factorcode.org/license.txt for BSD license.
USING: dlists deques threads sequences continuations namespaces
math quotations words kernel arrays assocs init system
concurrency.conditions accessors locals fry vocabs.loader ;
IN: concurrency.mailboxes
TUPLE: mailbox { threads dlist } { data dlist } ;
: <mailbox> ( -- mailbox )
mailbox new
<dlist> >>threads
<dlist> >>data ; inline
: mailbox-empty? ( mailbox -- bool )
data>> deque-empty? ; inline
GENERIC: mailbox-put ( obj mailbox -- )
M: mailbox mailbox-put
[ data>> push-front ]
[ threads>> notify-all ] bi yield ;
: wait-for-mailbox ( mailbox timeout -- )
[ threads>> ] dip "mailbox" wait ; inline
:: block-unless-pred ( ... mailbox timeout pred: ( ... message -- ... ? ) -- ... )
mailbox data>> pred dlist-any? [
mailbox timeout wait-for-mailbox
mailbox timeout pred block-unless-pred
] unless ; inline recursive
: block-if-empty ( mailbox timeout -- mailbox )
over mailbox-empty? [
2dup wait-for-mailbox block-if-empty
] [
drop
] if ; inline recursive
: mailbox-peek ( mailbox -- obj )
data>> peek-back ;
GENERIC# mailbox-get-timeout 1 ( mailbox timeout -- obj )
M: mailbox mailbox-get-timeout block-if-empty data>> pop-back ;
: mailbox-get ( mailbox -- obj )
f mailbox-get-timeout ; inline
: mailbox-get-all-timeout ( mailbox timeout -- array )
block-if-empty
[ dup mailbox-empty? not ]
[ dup data>> pop-back ]
produce nip ;
: mailbox-get-all ( mailbox -- array )
f mailbox-get-all-timeout ;
: while-mailbox-empty ( mailbox quot -- )
[ '[ _ mailbox-empty? ] ] dip while ; inline
: mailbox-get-timeout? ( mailbox timeout pred -- obj )
[ block-unless-pred ]
[ [ drop data>> ] dip delete-node-if ]
3bi ; inline
: mailbox-get? ( mailbox pred -- obj )
f swap mailbox-get-timeout? ; inline
: wait-for-close-timeout ( mailbox timeout -- )
over disposed>>
[ 2drop ] [ 2dup wait-for-mailbox wait-for-close-timeout ] if ;
: wait-for-close ( mailbox -- )
f wait-for-close-timeout ;
TUPLE: linked-error error thread ;
C: <linked-error> linked-error
: ?linked ( message -- message )
dup linked-error? [ rethrow ] when ;
TUPLE: linked-thread < thread supervisor ;
M: linked-thread error-in-thread
[ <linked-error> ] [ supervisor>> ] bi mailbox-put stop ;
: <linked-thread> ( quot name mailbox -- thread' )
[ linked-thread new-thread ] dip >>supervisor ;
: spawn-linked-to ( quot name mailbox -- thread )
<linked-thread> [ (spawn) ] keep ;
{ "concurrency.mailboxes" "debugger" } "concurrency.mailboxes.debugger" require-when