archive/1.vm/075channel.mu

   1 # Mu synchronizes between routines using channels rather than locks, like
   2 # Erlang and Go.
   3 #
   4 # The key property of channels: Writing to a full channel or reading from an
   5 # empty one will put the current routine in 'waiting' state until the
   6 # operation can be completed.
   7 #
   8 # Beware of addresses passed into channels. They can cause race conditions.
   9
  10 scenario channel [
  11   run [
  12     local-scope
  13     source:&:source:num, sink:&:sink:num <- new-channel 3/capacity
  14     sink <- write sink, 34
  15     10:num/raw, 11:bool/raw, source <- read source
  16   ]
  17   memory-should-contain [
  18     10 <- 34
  19     11 <- 0  # read was successful
  20   ]
  21 ]
  22
  23 container channel:_elem [
  24   lock:bool  # inefficient but simple: serialize all reads as well as writes
  25   first-full:num  # for write
  26   first-free:num  # for read
  27   # A circular buffer contains values from index first-full up to (but not
  28   # including) index first-free. The reader always modifies it at first-full,
  29   # while the writer always modifies it at first-free.
  30   data:&:@:_elem
  31 ]
  32
  33 # Since channels have two ends, and since it's an error to use either end from
  34 # multiple routines, let's distinguish the ends.
  35
  36 container source:_elem [
  37   chan:&:channel:_elem
  38 ]
  39
  40 container sink:_elem [
  41   chan:&:channel:_elem
  42 ]
  43
  44 def new-channel capacity:num -> in:&:source:_elem, out:&:sink:_elem [
  45   local-scope
  46   load-inputs
  47   result:&:channel:_elem <- new {(channel _elem): type}
  48   *result <- put *result, first-full:offset, 0
  49   *result <- put *result, first-free:offset, 0
  50   capacity <- add capacity, 1  # unused slot for 'full?' below
  51   data:&:@:_elem <- new _elem:type, capacity
  52   *result <- put *result, data:offset, data
  53   in <- new {(source _elem): type}
  54   *in <- put *in, chan:offset, result
  55   out <- new {(sink _elem): type}
  56   *out <- put *out, chan:offset, result
  57 ]
  58
  59 # write a value to a channel
  60 def write out:&:sink:_elem, val:_elem -> out:&:sink:_elem [
  61   local-scope
  62   load-inputs
  63   assert out, [write to null channel]
  64   chan:&:channel:_elem <- get *out, chan:offset
  65   <channel-write-initial>
  66   # block until lock is acquired AND queue has room
  67   lock:location <- get-location *chan, lock:offset
  68 #?   $print [write], 10/newline
  69   {
  70 #?     $print [trying to acquire lock for writing], 10/newline
  71     wait-for-reset-then-set lock
  72 #?     $print [lock acquired for writing], 10/newline
  73     full?:bool <- channel-full? chan
  74     break-unless full?
  75 #?     $print [but channel is full; relinquishing lock], 10/newline
  76     # channel is full; relinquish lock and give a reader the opportunity to
  77     # create room on it
  78     reset lock
  79     current-routine-is-blocked
  80     switch  # avoid spinlocking
  81     loop
  82   }
  83   current-routine-is-unblocked
  84 #?   $print [performing write], 10/newline
  85   # store a deep copy of val
  86   circular-buffer:&:@:_elem <- get *chan, data:offset
  87   free:num <- get *chan, first-free:offset
  88   *circular-buffer <- put-index *circular-buffer, free, val
  89   # mark its slot as filled
  90   free <- add free, 1
  91   {
  92     # wrap free around to 0 if necessary
  93     len:num <- length *circular-buffer
  94     at-end?:bool <- greater-or-equal free, len
  95     break-unless at-end?
  96     free <- copy 0
  97   }
  98   # write back
  99   *chan <- put *chan, first-free:offset, free
 100 #?   $print [relinquishing lock after writing], 10/newline
 101   reset lock
 102 ]
 103
 104 # read a value from a channel
 105 def read in:&:source:_elem -> result:_elem, eof?:bool, in:&:source:_elem [
 106   local-scope
 107   load-inputs
 108   assert in, [read on null channel]
 109   eof? <- copy false  # default result
 110   chan:&:channel:_elem <- get *in, chan:offset
 111   # block until lock is acquired AND queue has data
 112   lock:location <- get-location *chan, lock:offset
 113 #?   $print [read], 10/newline
 114   {
 115 #?     $print [trying to acquire lock for reading], 10/newline
 116     wait-for-reset-then-set lock
 117 #?     $print [lock acquired for reading], 10/newline
 118     empty?:bool <- channel-empty? chan
 119     break-unless empty?
 120 #?     $print [but channel is empty; relinquishing lock], 10/newline
 121     # channel is empty; relinquish lock and give a writer the opportunity to
 122     # add to it
 123     reset lock
 124     current-routine-is-blocked
 125     <channel-read-empty>
 126     switch  # avoid spinlocking
 127     loop
 128   }
 129   current-routine-is-unblocked
 130   # pull result off
 131   full:num <- get *chan, first-full:offset
 132   circular-buffer:&:@:_elem <- get *chan, data:offset
 133   result <- index *circular-buffer, full
 134   # clear the slot
 135   empty:&:_elem <- new _elem:type
 136   *circular-buffer <- put-index *circular-buffer, full, *empty
 137   # mark its slot as empty
 138   full <- add full, 1
 139   {
 140     # wrap full around to 0 if necessary
 141     len:num <- length *circular-buffer
 142     at-end?:bool <- greater-or-equal full, len
 143     break-unless at-end?
 144     full <- copy 0
 145   }
 146   # write back
 147   *chan <- put *chan, first-full:offset, full
 148 #?   $print [relinquishing lock after reading], 10/newline
 149   reset lock
 150 ]
 151
 152 # todo: create a notion of iterator and iterable so we can read/write whole
 153 # aggregates (arrays, lists, ..) of _elems at once.
 154
 155 scenario channel-initialization [
 156   run [
 157     local-scope
 158     source:&:source:num <- new-channel 3/capacity
 159     chan:&:channel:num <- get *source, chan:offset
 160     10:num/raw <- get *chan, first-full:offset
 161     11:num/raw <- get *chan, first-free:offset
 162   ]
 163   memory-should-contain [
 164     10 <- 0  # first-full
 165     11 <- 0  # first-free
 166   ]
 167 ]
 168
 169 scenario channel-write-increments-free [
 170   local-scope
 171   _, sink:&:sink:num <- new-channel 3/capacity
 172   run [
 173     sink <- write sink, 34
 174     chan:&:channel:num <- get *sink, chan:offset
 175     10:num/raw <- get *chan, first-full:offset
 176     11:num/raw <- get *chan, first-free:offset
 177   ]
 178   memory-should-contain [
 179     10 <- 0  # first-full
 180     11 <- 1  # first-free
 181   ]
 182 ]
 183
 184 scenario channel-read-increments-full [
 185   local-scope
 186   source:&:source:num, sink:&:sink:num <- new-channel 3/capacity
 187   sink <- write sink, 34
 188   run [
 189     _, _, source <- read source
 190     chan:&:channel:num <- get *source, chan:offset
 191     10:num/raw <- get *chan, first-full:offset
 192     11:num/raw <- get *chan, first-free:offset
 193   ]
 194   memory-should-contain [
 195     10 <- 1  # first-full
 196     11 <- 1  # first-free
 197   ]
 198 ]
 199
 200 scenario channel-wrap [
 201   local-scope
 202   # channel with just 1 slot
 203   source:&:source:num, sink:&:sink:num <- new-channel 1/capacity
 204   chan:&:channel:num <- get *source, chan:offset
 205   # write and read a value
 206   sink <- write sink, 34
 207   _, _, source <- read source
 208   run [
 209     # first-free will now be 1
 210     10:num/raw <- get *chan, first-free:offset
 211     11:num/raw <- get *chan, first-free:offset
 212     # write second value, verify that first-free wraps
 213     sink <- write sink, 34
 214     20:num/raw <- get *chan, first-free:offset
 215     # read second value, verify that first-full wraps
 216     _, _, source <- read source
 217     30:num/raw <- get *chan, first-full:offset
 218   ]
 219   memory-should-contain [
 220     10 <- 1  # first-free after first write
 221     11 <- 1  # first-full after first read
 222     20 <- 0  # first-free after second write, wrapped
 223     30 <- 0  # first-full after second read, wrapped
 224   ]
 225 ]
 226
 227 scenario channel-new-empty-not-full [
 228   run [
 229     local-scope
 230     source:&:source:num <- new-channel 3/capacity
 231     chan:&:channel:num <- get *source, chan:offset
 232     10:bool/raw <- channel-empty? chan
 233     11:bool/raw <- channel-full? chan
 234   ]
 235   memory-should-contain [
 236     10 <- 1  # empty?
 237     11 <- 0  # full?
 238   ]
 239 ]
 240
 241 scenario channel-write-not-empty [
 242   local-scope
 243   source:&:source:num, sink:&:sink:num <- new-channel 3/capacity
 244   chan:&:channel:num <- get *source, chan:offset
 245   run [
 246     sink <- write sink, 34
 247     10:bool/raw <- channel-empty? chan
 248     11:bool/raw <- channel-full? chan
 249   ]
 250   memory-should-contain [
 251     10 <- 0  # empty?
 252     11 <- 0  # full?
 253   ]
 254 ]
 255
 256 scenario channel-write-full [
 257   local-scope
 258   source:&:source:num, sink:&:sink:num <- new-channel 1/capacity
 259   chan:&:channel:num <- get *source, chan:offset
 260   run [
 261     sink <- write sink, 34
 262     10:bool/raw <- channel-empty? chan
 263     11:bool/raw <- channel-full? chan
 264   ]
 265   memory-should-contain [
 266     10 <- 0  # empty?
 267     11 <- 1  # full?
 268   ]
 269 ]
 270
 271 scenario channel-read-not-full [
 272   local-scope
 273   source:&:source:num, sink:&:sink:num <- new-channel 1/capacity
 274   chan:&:channel:num <- get *source, chan:offset
 275   sink <- write sink, 34
 276   run [
 277     _, _, source <- read source
 278     10:bool/raw <- channel-empty? chan
 279     11:bool/raw <- channel-full? chan
 280   ]
 281   memory-should-contain [
 282     10 <- 1  # empty?
 283     11 <- 0  # full?
 284   ]
 285 ]
 286
 287 scenario channel-clear [
 288   local-scope
 289   # create a channel with a few items
 290   source:&:source:num, sink:&:sink:num <- new-channel 3/capacity
 291   chan:&:channel:num <- get *sink, chan:offset
 292   write sink, 30
 293   write sink, 31
 294   write sink, 32
 295   run [
 296     clear source
 297     10:bool/raw <- channel-empty? chan
 298   ]
 299   memory-should-contain [
 300     10 <- 1  # after the call to 'clear', the channel should be empty
 301   ]
 302 ]
 303
 304 def clear in:&:source:_elem -> in:&:source:_elem [
 305   local-scope
 306   load-inputs
 307   chan:&:channel:_elem <- get *in, chan:offset
 308   {
 309     empty?:bool <- channel-empty? chan
 310     break-if empty?
 311     _, _, in <- read in
 312     loop
 313   }
 314 ]
 315
 316 ## cancelling channels
 317
 318 # every channel comes with a boolean signifying if it's been closed
 319 # initially this boolean is false
 320 container channel:_elem [
 321   closed?:bool
 322 ]
 323
 324 # a channel can be closed from either the source or the sink
 325 # both routines can modify the 'closed?' bit, but they can only ever set it, so this is a benign race
 326 def close x:&:source:_elem -> x:&:source:_elem [
 327   local-scope
 328   load-inputs
 329   chan:&:channel:_elem <- get *x, chan:offset
 330   *chan <- put *chan, closed?:offset, true
 331 ]
 332 def close x:&:sink:_elem -> x:&:sink:_elem [
 333   local-scope
 334   load-inputs
 335   chan:&:channel:_elem <- get *x, chan:offset
 336   *chan <- put *chan, closed?:offset, true
 337 ]
 338
 339 # once a channel is closed from one side, no further operations are expected from that side
 340 # if a channel is closed for reading,
 341 #   no further writes will be let through
 342 # if a channel is closed for writing,
 343 #   future reads continue until the channel empties,
 344 #   then the channel is also closed for reading
 345 after <channel-write-initial> [
 346   closed?:bool <- get *chan, closed?:offset
 347   return-if closed?
 348 ]
 349 after <channel-read-empty> [
 350   closed?:bool <- get *chan, closed?:offset
 351   {
 352     break-unless closed?
 353     empty-result:&:_elem <- new _elem:type
 354     current-routine-is-unblocked
 355     return *empty-result, true
 356   }
 357 ]
 358
 359 ## helpers
 360
 361 # An empty channel has first-free and first-full both at the same value.
 362 def channel-empty? chan:&:channel:_elem -> result:bool [
 363   local-scope
 364   load-inputs
 365   # return chan.first-full == chan.first-free
 366   full:num <- get *chan, first-full:offset
 367   free:num <- get *chan, first-free:offset
 368   result <- equal full, free
 369 ]
 370
 371 # A full channel has first-free just before first-full, wasting one slot.
 372 # (Other alternatives: https://www.snellman.net/blog/archive/2016-12-13-ring-buffers)
 373 def channel-full? chan:&:channel:_elem -> result:bool [
 374   local-scope
 375   load-inputs
 376   # tmp = chan.first-free + 1
 377   tmp:num <- get *chan, first-free:offset
 378   tmp <- add tmp, 1
 379   {
 380     # if tmp == chan.capacity, tmp = 0
 381     len:num <- capacity chan
 382     at-end?:bool <- greater-or-equal tmp, len
 383     break-unless at-end?
 384     tmp <- copy 0
 385   }
 386   # return chan.first-full == tmp
 387   full:num <- get *chan, first-full:offset
 388   result <- equal full, tmp
 389 ]
 390
 391 def capacity chan:&:channel:_elem -> result:num [
 392   local-scope
 393   load-inputs
 394   q:&:@:_elem <- get *chan, data:offset
 395   result <- length *q
 396 ]
 397
 398 ## helpers for channels of characters in particular
 399
 400 def buffer-lines in:&:source:char, buffered-out:&:sink:char -> buffered-out:&:sink:char, in:&:source:char [
 401   local-scope
 402   load-inputs
 403   # repeat forever
 404   eof?:bool <- copy false
 405   {
 406     line:&:buffer:char <- new-buffer 30
 407     # read characters from 'in' until newline, copy into line
 408     {
 409       +next-character
 410       c:char, eof?:bool, in <- read in
 411       break-if eof?
 412       # drop a character on backspace
 413       {
 414         # special-case: if it's a backspace
 415         backspace?:bool <- equal c, 8
 416         break-unless backspace?
 417         # drop previous character
 418         {
 419           buffer-length:num <- get *line, length:offset
 420           buffer-empty?:bool <- equal buffer-length, 0
 421           break-if buffer-empty?
 422           buffer-length <- subtract buffer-length, 1
 423           *line <- put *line, length:offset, buffer-length
 424         }
 425         # and don't append this one
 426         loop +next-character
 427       }
 428       # append anything else
 429       line <- append line, c
 430       line-done?:bool <- equal c, 10/newline
 431       break-if line-done?
 432       loop
 433     }
 434     # copy line into 'buffered-out'
 435     i:num <- copy 0
 436     line-contents:text <- get *line, data:offset
 437     max:num <- get *line, length:offset
 438     {
 439       done?:bool <- greater-or-equal i, max
 440       break-if done?
 441       c:char <- index *line-contents, i
 442       buffered-out <- write buffered-out, c
 443       i <- add i, 1
 444       loop
 445     }
 446     {
 447       break-unless eof?
 448       buffered-out <- close buffered-out
 449       return
 450     }
 451     loop
 452   }
 453 ]
 454
 455 scenario buffer-lines-blocks-until-newline [
 456   run [
 457     local-scope
 458     source:&:source:char, sink:&:sink:char <- new-channel 10/capacity
 459     _, buffered-stdin:&:sink:char/buffered-stdin <- new-channel 10/capacity
 460     buffered-chan:&:channel:char <- get *buffered-stdin, chan:offset
 461     empty?:bool <- channel-empty? buffered-chan
 462     assert empty?, [
 463 F buffer-lines-blocks-until-newline: channel should be empty after init]
 464     # buffer stdin into buffered-stdin, try to read from buffered-stdin
 465     buffer-routine:num <- start-running buffer-lines, source, buffered-stdin
 466     wait-for-routine-to-block buffer-routine
 467     empty? <- channel-empty? buffered-chan
 468     assert empty?:bool, [
 469 F buffer-lines-blocks-until-newline: channel should be empty after buffer-lines bring-up]
 470     # write 'a'
 471     sink <- write sink, 97/a
 472     restart buffer-routine
 473     wait-for-routine-to-block buffer-routine
 474     empty? <- channel-empty? buffered-chan
 475     assert empty?:bool, [
 476 F buffer-lines-blocks-until-newline: channel should be empty after writing 'a']
 477     # write 'b'
 478     sink <- write sink, 98/b
 479     restart buffer-routine
 480     wait-for-routine-to-block buffer-routine
 481     empty? <- channel-empty? buffered-chan
 482     assert empty?:bool, [
 483 F buffer-lines-blocks-until-newline: channel should be empty after writing 'b']
 484     # write newline
 485     sink <- write sink, 10/newline
 486     restart buffer-routine
 487     wait-for-routine-to-block buffer-routine
 488     empty? <- channel-empty? buffered-chan
 489     data-emitted?:bool <- not empty?
 490     assert data-emitted?, [
 491 F buffer-lines-blocks-until-newline: channel should contain data after writing newline]
 492     trace 1, [test], [reached end]
 493   ]
 494   trace-should-contain [
 495     test: reached end
 496   ]
 497 ]
 498
 499 def drain source:&:source:char -> result:text, source:&:source:char [
 500   local-scope
 501   load-inputs
 502   buf:&:buffer:char <- new-buffer 30
 503   {
 504     c:char, done?:bool <- read source
 505     break-if done?
 506     buf <- append buf, c
 507     loop
 508   }
 509   result <- buffer-to-array buf
 510 ]