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: dlists deques threads sequences continuations namespaces
   4 math quotations words kernel arrays assocs init system
   5 concurrency.conditions accessors locals fry vocabs.loader ;
   6 IN: concurrency.mailboxes
   7
   8 TUPLE: mailbox { threads dlist } { data dlist } ;
   9
  10 : <mailbox> ( -- mailbox )
  11     mailbox new
  12         <dlist> >>threads
  13         <dlist> >>data ; inline
  14
  15 : mailbox-empty? ( mailbox -- bool )
  16     data>> deque-empty? ; inline
  17
  18 GENERIC: mailbox-put ( obj mailbox -- )
  19
  20 M: mailbox mailbox-put
  21     [ data>> push-front ]
  22     [ threads>> notify-all ] bi yield ;
  23
  24 : wait-for-mailbox ( mailbox timeout -- )
  25     [ threads>> ] dip "mailbox" wait ; inline
  26
  27 :: block-unless-pred ( ... mailbox timeout pred: ( ... message -- ... ? ) -- ... )
  28     mailbox data>> pred dlist-any? [
  29         mailbox timeout wait-for-mailbox
  30         mailbox timeout pred block-unless-pred
  31     ] unless ; inline recursive
  32
  33 : block-if-empty ( mailbox timeout -- mailbox )
  34     over mailbox-empty? [
  35         2dup wait-for-mailbox block-if-empty
  36     ] [
  37         drop
  38     ] if ; inline recursive
  39
  40 : mailbox-peek ( mailbox -- obj )
  41     data>> peek-back ;
  42
  43 GENERIC#: mailbox-get-timeout 1 ( mailbox timeout -- obj )
  44
  45 M: mailbox mailbox-get-timeout block-if-empty data>> pop-back ;
  46
  47 : mailbox-get ( mailbox -- obj )
  48     f mailbox-get-timeout ; inline
  49
  50 : mailbox-get-all-timeout ( mailbox timeout -- array )
  51     block-if-empty
  52     [ dup mailbox-empty? not ]
  53     [ dup data>> pop-back ]
  54     produce nip ;
  55
  56 : mailbox-get-all ( mailbox -- array )
  57     f mailbox-get-all-timeout ;
  58
  59 : while-mailbox-empty ( mailbox quot -- )
  60     [ '[ _ mailbox-empty? ] ] dip while ; inline
  61
  62 : mailbox-get-timeout? ( mailbox timeout pred -- obj )
  63     [ block-unless-pred ]
  64     [ [ drop data>> ] dip delete-node-if ]
  65     3bi ; inline
  66
  67 : mailbox-get? ( mailbox pred -- obj )
  68     f swap mailbox-get-timeout? ; inline
  69
  70 : wait-for-close-timeout ( mailbox timeout -- )
  71     over disposed>>
  72     [ 2drop ] [ 2dup wait-for-mailbox wait-for-close-timeout ] if ;
  73
  74 : wait-for-close ( mailbox -- )
  75     f wait-for-close-timeout ;
  76
  77 TUPLE: linked-error error thread ;
  78
  79 C: <linked-error> linked-error
  80
  81 : ?linked ( message -- message )
  82     dup linked-error? [ rethrow ] when ;
  83
  84 TUPLE: linked-thread < thread supervisor ;
  85
  86 M: linked-thread error-in-thread
  87     [ <linked-error> ] [ supervisor>> ] bi mailbox-put stop ;
  88
  89 : <linked-thread> ( quot name mailbox -- thread' )
  90     [ linked-thread new-thread ] dip >>supervisor ;
  91
  92 : spawn-linked-to ( quot name mailbox -- thread )
  93     <linked-thread> [ (spawn) ] keep ;
  94
  95 { "concurrency.mailboxes" "debugger" } "concurrency.mailboxes.debugger" require-when