basis/concurrency/mailboxes/mailboxes.factor

   1 ! Copyright (C) 2005, 2010 Chris Double, Slava Pestov.
   2 ! See http://factorcode.org/license.txt for BSD license.
   3 USING: accessors concurrency.conditions continuations deques
   4 destructors dlists kernel sequences threads vocabs.loader ;
   5 IN: concurrency.mailboxes
   6
   7 TUPLE: mailbox { threads dlist } { data dlist } ;
   8
   9 : <mailbox> ( -- mailbox )
  10     mailbox new
  11         <dlist> >>threads
  12         <dlist> >>data ; inline
  13
  14 : mailbox-empty? ( mailbox -- bool )
  15     data>> deque-empty? ; inline
  16
  17 GENERIC: mailbox-put ( obj mailbox -- )
  18
  19 M: mailbox mailbox-put
  20     [ data>> push-front ]
  21     [ threads>> notify-all ] bi yield ;
  22
  23 : wait-for-mailbox ( mailbox timeout -- )
  24     [ threads>> ] dip "mailbox" wait ; inline
  25
  26 :: block-unless-pred ( ... mailbox timeout pred: ( ... message -- ... ? ) -- ... )
  27     mailbox data>> pred dlist-any? [
  28         mailbox timeout wait-for-mailbox
  29         mailbox timeout pred block-unless-pred
  30     ] unless ; inline recursive
  31
  32 : block-if-empty ( mailbox timeout -- mailbox )
  33     over mailbox-empty? [
  34         2dup wait-for-mailbox block-if-empty
  35     ] [
  36         drop
  37     ] if ; inline recursive
  38
  39 : mailbox-peek ( mailbox -- obj )
  40     data>> peek-back ;
  41
  42 GENERIC#: mailbox-get-timeout 1 ( mailbox timeout -- obj )
  43
  44 M: mailbox mailbox-get-timeout block-if-empty data>> pop-back ;
  45
  46 : mailbox-get ( mailbox -- obj )
  47     f mailbox-get-timeout ; inline
  48
  49 : mailbox-get-all-timeout ( mailbox timeout -- seq )
  50     block-if-empty data>> [ ] collector [ slurp-deque ] dip ;
  51
  52 : mailbox-get-all ( mailbox -- seq )
  53     f mailbox-get-all-timeout ;
  54
  55 : while-mailbox-empty ( mailbox quot -- )
  56     [ '[ _ mailbox-empty? ] ] dip while ; inline
  57
  58 : mailbox-get-timeout? ( mailbox timeout pred -- obj )
  59     [ block-unless-pred ]
  60     [ [ drop data>> ] dip delete-node-if ]
  61     3bi ; inline
  62
  63 : mailbox-get? ( mailbox pred -- obj )
  64     f swap mailbox-get-timeout? ; inline
  65
  66 : wait-for-close-timeout ( mailbox timeout -- )
  67     '[
  68         _ 2dup wait-for-mailbox wait-for-close-timeout
  69     ] unless-disposed ;
  70
  71 : wait-for-close ( mailbox -- )
  72     f wait-for-close-timeout ;
  73
  74 TUPLE: linked-error error thread ;
  75
  76 C: <linked-error> linked-error
  77
  78 : ?linked ( message -- message )
  79     dup linked-error? [ rethrow ] when ;
  80
  81 TUPLE: linked-thread < thread supervisor ;
  82
  83 M: linked-thread error-in-thread
  84     [ <linked-error> ] [ supervisor>> ] bi mailbox-put stop ;
  85
  86 : <linked-thread> ( quot name mailbox -- thread' )
  87     [ linked-thread new-thread ] dip >>supervisor ;
  88
  89 : spawn-linked-to ( quot name mailbox -- thread )
  90     <linked-thread> [ (spawn) ] keep ;
  91
  92 USE-WHEN-LOADED: concurrency.mailboxes.debugger { "concurrency.mailboxes" "debugger" }