Revert "reworked messaging, now always use an envelope around the message to handle expiration correctly in all use-cases"
This reverts commit 8231d10136.
db4
Doug Coleman
parent
0f4dce4e4b
commit
97b6509cfe
|
|
@ -10,24 +10,21 @@ IN: concurrency.messaging.tests
|
|||
|
||||
[ "received" ] [
|
||||
[
|
||||
[ drop "received" ] handle-synchronous
|
||||
receive "received" swap reply-synchronous
|
||||
] "Synchronous test" spawn
|
||||
"sent" swap send-synchronous
|
||||
] unit-test
|
||||
|
||||
[ "received" ] [
|
||||
[
|
||||
[ drop "received" ] handle-synchronous
|
||||
receive "received" swap reply-synchronous
|
||||
] "Synchronous test" spawn
|
||||
[ 100 milliseconds "sent" ] dip send-synchronous-timeout
|
||||
] unit-test
|
||||
|
||||
[
|
||||
[
|
||||
100 milliseconds sleep
|
||||
[ drop "received" ] handle-synchronous
|
||||
] "Synchronous test" spawn
|
||||
[ 5 milliseconds "sent" ] dip send-synchronous-timeout
|
||||
[ [ 100 milliseconds sleep
|
||||
receive "received" swap reply-synchronous ] "Synchronous test" spawn
|
||||
[ 50 milliseconds "sent" ] dip send-synchronous-timeout
|
||||
] [ wait-timeout? ] must-fail-with
|
||||
|
||||
|
||||
|
|
@ -80,4 +77,3 @@ SYMBOL: exit
|
|||
! ] "Bad synchronous send" spawn "t" set
|
||||
|
||||
! [ 3 "t" get send-synchronous ] must-fail
|
||||
|
||||
|
|
|
|||
|
|
@ -1,82 +1,32 @@
|
|||
! Copyright (C) 2005, 2008 Chris Double, Slava Pestov.
|
||||
! See http://factorcode.org/license.txt for BSD license.
|
||||
USING: kernel threads concurrency.mailboxes continuations
|
||||
namespaces assocs accessors summary fry calendar math sequences ;
|
||||
namespaces assocs accessors summary fry ;
|
||||
IN: concurrency.messaging
|
||||
|
||||
TUPLE: envelope data sender tag expiry ;
|
||||
|
||||
<PRIVATE
|
||||
|
||||
: new-envelope ( data class -- envelope )
|
||||
new swap >>data self >>sender ;
|
||||
|
||||
: <envelope> ( data -- envelope )
|
||||
dup envelope?
|
||||
[ envelope new-envelope ] unless ;
|
||||
|
||||
: expired? ( message -- ? )
|
||||
dup envelope?
|
||||
[ expiry>>
|
||||
[ now (time-) 0 < ]
|
||||
[ f ] if*
|
||||
] [ drop f ] if ; inline
|
||||
|
||||
: if-expired ( message quot -- message )
|
||||
[ dup expired? ] dip
|
||||
'[ drop _ call( -- message ) ] [ ] if ; inline
|
||||
|
||||
PRIVATE>
|
||||
|
||||
GENERIC: send ( message thread -- )
|
||||
|
||||
GENERIC: send-timeout ( timeout message thread -- )
|
||||
|
||||
: mailbox-of ( thread -- mailbox )
|
||||
dup mailbox>> [ ] [
|
||||
<mailbox> [ >>mailbox drop ] keep
|
||||
] ?if ;
|
||||
|
||||
M: thread send ( message thread -- )
|
||||
[ <envelope> ] dip
|
||||
check-registered mailbox-of mailbox-put ;
|
||||
|
||||
M: thread send-timeout ( timeout message thread -- )
|
||||
[ <envelope> swap hence >>expiry ] dip send ;
|
||||
|
||||
: my-mailbox ( -- mailbox ) self mailbox-of ;
|
||||
|
||||
: (receive) ( -- message )
|
||||
my-mailbox mailbox-get ?linked
|
||||
[ (receive) ] if-expired ;
|
||||
|
||||
: receive ( -- message )
|
||||
(receive) data>> ;
|
||||
|
||||
: (receive-timeout) ( timeout -- message )
|
||||
[ my-mailbox ] dip
|
||||
[ mailbox-get-timeout ?linked ] keep
|
||||
'[ _ (receive-timeout) ] if-expired ; inline
|
||||
my-mailbox mailbox-get ?linked ;
|
||||
|
||||
: receive-timeout ( timeout -- message )
|
||||
(receive-timeout) data>> ;
|
||||
|
||||
: (receive-if) ( pred -- message )
|
||||
[ my-mailbox ] dip
|
||||
[ mailbox-get? ?linked ] keep
|
||||
'[ _ (receive-if) ] if-expired ; inline
|
||||
[ my-mailbox ] dip mailbox-get-timeout ?linked ;
|
||||
|
||||
: receive-if ( pred -- message )
|
||||
[ data>> ] prepend (receive-if) data>> ; inline
|
||||
|
||||
: (receive-if-timeout) ( timeout pred -- message )
|
||||
[ my-mailbox ] 2dip
|
||||
[ mailbox-get-timeout? ?linked ] 2keep
|
||||
'[ _ _ (receive-if-timeout) ] if-expired ; inline
|
||||
[ my-mailbox ] dip mailbox-get? ?linked ; inline
|
||||
|
||||
: receive-if-timeout ( timeout pred -- message )
|
||||
[ data>> ] prepend
|
||||
(receive-if-timeout) data>> ; inline
|
||||
[ my-mailbox ] 2dip mailbox-get-timeout? ?linked ; inline
|
||||
|
||||
: rethrow-linked ( error process supervisor -- )
|
||||
[ <linked-error> ] dip send ;
|
||||
|
|
@ -84,17 +34,15 @@ M: thread send-timeout ( timeout message thread -- )
|
|||
: spawn-linked ( quot name -- thread )
|
||||
my-mailbox spawn-linked-to ;
|
||||
|
||||
TUPLE: synchronous < envelope ;
|
||||
TUPLE: synchronous data sender tag ;
|
||||
|
||||
: <synchronous> ( data -- sync )
|
||||
synchronous new-envelope
|
||||
synchronous counter >>tag ;
|
||||
self synchronous counter synchronous boa ;
|
||||
|
||||
TUPLE: reply < envelope ;
|
||||
TUPLE: reply data tag ;
|
||||
|
||||
: <reply> ( data synchronous -- reply )
|
||||
[ reply new-envelope ] dip
|
||||
tag>> >>tag ;
|
||||
tag>> \ reply boa ;
|
||||
|
||||
: synchronous-reply? ( response synchronous -- ? )
|
||||
over reply? [ [ tag>> ] bi@ = ] [ 2drop f ] if ;
|
||||
|
|
@ -109,28 +57,24 @@ M: cannot-send-synchronous-to-self summary
|
|||
cannot-send-synchronous-to-self
|
||||
] [
|
||||
[ <synchronous> dup ] dip send
|
||||
'[ _ synchronous-reply? ] (receive-if) data>>
|
||||
] if ;
|
||||
'[ _ synchronous-reply? ] receive-if
|
||||
data>>
|
||||
] if ;
|
||||
|
||||
: send-synchronous-timeout ( timeout message thread -- reply )
|
||||
dup self eq? [
|
||||
cannot-send-synchronous-to-self
|
||||
] [
|
||||
[ <synchronous> 2dup ] dip send-timeout
|
||||
'[ _ synchronous-reply? ] (receive-if-timeout) data>>
|
||||
[ <synchronous> dup ] dip send
|
||||
'[ _ synchronous-reply? ] receive-if-timeout
|
||||
data>>
|
||||
] if ;
|
||||
|
||||
<PRIVATE
|
||||
|
||||
|
||||
: reply-synchronous ( message synchronous -- )
|
||||
dup expired?
|
||||
[ 2drop ]
|
||||
[ [ <reply> ] keep sender>> send ] if ;
|
||||
|
||||
PRIVATE>
|
||||
[ <reply> ] keep sender>> send ;
|
||||
|
||||
: handle-synchronous ( quot -- )
|
||||
(receive) [
|
||||
receive [
|
||||
data>> swap call
|
||||
] keep reply-synchronous ; inline
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue