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 debugger debugger.threads
   6 locals fry ;
   7 IN: concurrency.mailboxes
   8
   9 TUPLE: mailbox threads data ;
  10
  11 : <mailbox> ( -- mailbox )
  12     mailbox new
  13         <dlist> >>threads
  14         <dlist> >>data ;
  15
  16 : mailbox-empty? ( mailbox -- bool )
  17     data>> deque-empty? ;
  18
  19 : mailbox-put ( obj mailbox -- )
  20     [ data>> push-front ]
  21     [ threads>> notify-all ] bi yield ;
  22
  23 : wait-for-mailbox ( mailbox timeout -- )
  24     [ threads>> ] dip "mailbox" wait ;
  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 ;
  38
  39 : mailbox-peek ( mailbox -- obj )
  40     data>> peek-back ;
  41
  42 : mailbox-get-timeout ( mailbox timeout -- obj )
  43     block-if-empty data>> pop-back ;
  44
  45 : mailbox-get ( mailbox -- obj )
  46     f mailbox-get-timeout ;
  47
  48 : mailbox-get-all-timeout ( mailbox timeout -- array )
  49     block-if-empty
  50     [ dup mailbox-empty? not ]
  51     [ dup data>> pop-back ]
  52     produce nip ;
  53
  54 : mailbox-get-all ( mailbox -- array )
  55     f mailbox-get-all-timeout ;
  56
  57 : while-mailbox-empty ( mailbox quot -- )
  58     [ '[ _ mailbox-empty? ] ] dip while ; inline
  59
  60 : mailbox-get-timeout? ( mailbox timeout pred -- obj )
  61     [ block-unless-pred ]
  62     [ [ drop data>> ] dip delete-node-if ]
  63     3bi ; inline
  64
  65 : mailbox-get? ( mailbox pred -- obj )
  66     f swap mailbox-get-timeout? ; inline
  67
  68 : wait-for-close-timeout ( mailbox timeout -- )
  69     over disposed>>
  70     [ 2drop ] [ 2dup wait-for-mailbox wait-for-close-timeout ] if ;
  71
  72 : wait-for-close ( mailbox -- )
  73     f wait-for-close-timeout ;
  74
  75 TUPLE: linked-error error thread ;
  76
  77 M: linked-error error.
  78     [ thread>> error-in-thread. ] [ error>> error. ] bi ;
  79
  80 C: <linked-error> linked-error
  81
  82 : ?linked ( message -- message )
  83     dup linked-error? [ rethrow ] when ;
  84
  85 TUPLE: linked-thread < thread supervisor ;
  86
  87 M: linked-thread error-in-thread
  88     [ <linked-error> ] [ supervisor>> ] bi mailbox-put ;
  89
  90 : <linked-thread> ( quot name mailbox -- thread' )
  91     [ linked-thread new-thread ] dip >>supervisor ;
  92
  93 : spawn-linked-to ( quot name mailbox -- thread )
  94     <linked-thread> [ (spawn) ] keep ;