basis/threads/threads.factor

   1 ! Copyright (C) 2004, 2009 Slava Pestov.
   2 ! Copyright (C) 2005 Mackenzie Straight.
   3 ! See http://factorcode.org/license.txt for BSD license.
   4 USING: arrays hashtables heaps kernel kernel.private math
   5 namespaces sequences vectors continuations continuations.private
   6 dlists assocs system combinators combinators.private init boxes
   7 accessors math.order deques strings quotations fry ;
   8 IN: threads
   9
  10 <PRIVATE
  11
  12 ! (set-context) and (start-context) are sub-primitives, but
  13 ! we don't want them inlined into callers since their behavior
  14 ! depends on what frames are on the callstack
  15 : start-context ( obj quot: ( obj -- * ) -- ) (start-context) ;
  16 : set-context ( context -- ) (set-context) ;
  17
  18 PRIVATE>
  19
  20 SYMBOL: initial-thread
  21
  22 TUPLE: thread
  23 { name string }
  24 { quot callable initial: [ ] }
  25 { exit-handler callable initial: [ ] }
  26 { id integer }
  27 continuation
  28 state
  29 runnable
  30 mailbox
  31 variables
  32 sleep-entry ;
  33
  34 : self ( -- thread ) 63 special-object ; inline
  35
  36 ! Thread-local storage
  37 : tnamespace ( -- assoc )
  38     self variables>> [ H{ } clone dup self (>>variables) ] unless* ;
  39
  40 : tget ( key -- value )
  41     self variables>> at ;
  42
  43 : tset ( value key -- )
  44     tnamespace set-at ;
  45
  46 : tchange ( key quot -- )
  47     tnamespace swap change-at ; inline
  48
  49 : threads ( -- assoc ) 64 special-object ;
  50
  51 : thread ( id -- thread ) threads at ;
  52
  53 : thread-registered? ( thread -- ? )
  54     id>> threads key? ;
  55
  56 ERROR: already-stopped thread ;
  57
  58 : check-unregistered ( thread -- thread )
  59     dup thread-registered? [ already-stopped ] when ;
  60
  61 ERROR: not-running thread ;
  62
  63 : check-registered ( thread -- thread )
  64     dup thread-registered? [ not-running ] unless ;
  65
  66 <PRIVATE
  67
  68 : register-thread ( thread -- )
  69     check-unregistered dup id>> threads set-at ;
  70
  71 : unregister-thread ( thread -- )
  72     check-registered id>> threads delete-at ;
  73
  74 : set-self ( thread -- ) 63 set-special-object ; inline
  75
  76 PRIVATE>
  77
  78 : new-thread ( quot name class -- thread )
  79     new
  80         swap >>name
  81         swap >>quot
  82         \ thread counter >>id
  83         <box> >>continuation ; inline
  84
  85 : <thread> ( quot name -- thread )
  86     \ thread new-thread ;
  87
  88 : run-queue ( -- dlist ) 65 special-object ;
  89
  90 : sleep-queue ( -- heap ) 66 special-object ;
  91
  92 : resume ( thread -- )
  93     f >>state
  94     check-registered run-queue push-front ;
  95
  96 : resume-now ( thread -- )
  97     f >>state
  98     check-registered run-queue push-back ;
  99
 100 : resume-with ( obj thread -- )
 101     f >>state
 102     check-registered 2array run-queue push-front ;
 103
 104 : sleep-time ( -- nanos/f )
 105     {
 106         { [ run-queue deque-empty? not ] [ 0 ] }
 107         { [ sleep-queue heap-empty? ] [ f ] }
 108         [ sleep-queue heap-peek nip nano-count [-] ]
 109     } cond ;
 110
 111 DEFER: stop
 112
 113 <PRIVATE
 114
 115 : schedule-sleep ( thread dt -- )
 116     [ check-registered dup ] dip sleep-queue heap-push*
 117     >>sleep-entry drop ;
 118
 119 : expire-sleep? ( heap -- ? )
 120     dup heap-empty?
 121     [ drop f ] [ heap-peek nip nano-count <= ] if ;
 122
 123 : expire-sleep ( thread -- )
 124     f >>sleep-entry resume ;
 125
 126 : expire-sleep-loop ( -- )
 127     sleep-queue
 128     [ dup expire-sleep? ]
 129     [ dup heap-pop drop expire-sleep ]
 130     while
 131     drop ;
 132
 133 : start ( namestack thread -- * )
 134     [
 135         set-self
 136         set-namestack
 137         V{ } set-catchstack
 138         { } set-retainstack
 139         { } set-datastack
 140         self quot>> [ call stop ] call-clear
 141     ] (( namestack thread -- * )) call-effect-unsafe ;
 142
 143 DEFER: next
 144
 145 : no-runnable-threads ( -- * )
 146     ! We should never be in a state where the only threads
 147     ! are sleeping; the I/O wait thread is always runnable.
 148     ! However, if it dies, we handle this case
 149     ! semi-gracefully.
 150     !
 151     ! And if sleep-time outputs f, there are no sleeping
 152     ! threads either... so WTF.
 153     sleep-time {
 154         { [ dup not ] [ drop die ] }
 155         { [ dup 0 = ] [ drop ] }
 156         [ (sleep) ]
 157     } cond next ;
 158
 159 : (next) ( arg thread -- * )
 160     f >>state
 161     dup set-self
 162     dup runnable>> [
 163         continuation>> box> continue-with
 164     ] [
 165         t >>runnable start
 166     ] if ;
 167
 168 : next ( -- * )
 169     expire-sleep-loop
 170     run-queue dup deque-empty? [
 171         drop no-runnable-threads
 172     ] [
 173         pop-back dup array? [ first2 ] [ f swap ] if (next)
 174     ] if ;
 175
 176 PRIVATE>
 177
 178 : stop ( -- )
 179     self [ exit-handler>> call( -- ) ] [ unregister-thread ] bi next ;
 180
 181 : suspend ( quot state -- obj )
 182     [
 183         [ [ self swap call ] dip self (>>state) ] dip
 184         self continuation>> >box
 185         next
 186     ] callcc1 2nip ; inline
 187
 188 : yield ( -- ) [ resume ] f suspend drop ;
 189
 190 GENERIC: sleep-until ( n/f -- )
 191
 192 M: integer sleep-until
 193     '[ _ schedule-sleep ] "sleep" suspend drop ;
 194
 195 M: f sleep-until
 196     drop [ drop ] "interrupt" suspend drop ;
 197
 198 GENERIC: sleep ( dt -- )
 199
 200 M: real sleep
 201     >integer nano-count + sleep-until ;
 202
 203 : interrupt ( thread -- )
 204     dup state>> [
 205         dup sleep-entry>> [ sleep-queue heap-delete ] when*
 206         f >>sleep-entry
 207         dup resume
 208     ] when drop ;
 209
 210 : (spawn) ( thread -- )
 211     [ register-thread ] [ namestack swap resume-with ] bi ;
 212
 213 : spawn ( quot name -- thread )
 214     <thread> [ (spawn) ] keep ;
 215
 216 : spawn-server ( quot name -- thread )
 217     [ '[ _ loop ] ] dip spawn ;
 218
 219 : in-thread ( quot -- )
 220     [ datastack ] dip
 221     '[ _ set-datastack _ call ]
 222     "Thread" spawn drop ;
 223
 224 GENERIC: error-in-thread ( error thread -- )
 225
 226 <PRIVATE
 227
 228 : init-threads ( -- )
 229     H{ } clone 64 set-special-object
 230     <dlist> 65 set-special-object
 231     <min-heap> 66 set-special-object
 232     initial-thread global
 233     [ drop [ ] "Initial" <thread> ] cache
 234     <box> >>continuation
 235     t >>runnable
 236     f >>state
 237     dup register-thread
 238     set-self ;
 239
 240 PRIVATE>
 241
 242 [ init-threads ] "threads" add-startup-hook