about summary refs log tree commit diff stats
path: root/075channel.mu
blob: 985292a325d17f826fe7d461704cb2ec87e6d522 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
# Mu synchronizes between routines using channels rather than locks, like
# Erlang and Go.
#
# Key properties of channels:
#
#   a) Writing to a full channel or reading from an empty one will put the
#   current routine in 'waiting' state until the operation can be completed.
#
#   b) Writing to a channel implicitly performs a deep copy. This prevents
#   addresses from being shared between routines, and therefore eliminates all
#   possibility of race conditions.
#
# There's still a narrow window for race conditions: the ingredients passed in
# to 'start-running'. Pass only channels into routines and you should be fine.
# Any other mutable ingredients will require locks.

scenario channel [
  run [
    local-scope
    source:&:source:num, sink:&:sink:num <- new-channel 3/capacity
    sink <- write sink, 34
    10:num/raw, 11:bool/raw, source <- read source
  ]
  memory-should-contain [
    10 <- 34
    11 <- 0  # read was successful
  ]
]

container channel:_elem [
  lock:bool  # inefficient but simple: serialize all reads as well as writes
  first-full:num  # for write
  first-free:num  # for read
  # A circular buffer contains values from index first-full up to (but not
  # including) index first-empty. The reader always modifies it at first-full,
  # while the writer always modifies it at first-empty.
  data:&:@:_elem
]

# Since channels have two ends, and since it's an error to use either end from
# multiple routines, let's distinguish the ends.

container source:_elem [
  chan:&:channel:_elem
]

container sink:_elem [
  chan:&:channel:_elem
]

def new-channel capacity:num -> in:&:source:_elem, out:&:sink:_elem [
  local-scope
  load-ingredients
  result:&:channel:_elem <- new {(channel _elem): type}
  *result <- put *result, first-full:offset, 0
  *result <- put *result, first-free:offset, 0
  capacity <- add capacity, 1  # unused slot for 'full?' below
  data:&:@:_elem <- new _elem:type, capacity
  *result <- put *result, data:offset, data
  in <- new {(source _elem): type}
  *in <- put *in, chan:offset, result
  out <- new {(sink _elem): type}
  *out <- put *out, chan:offset, result
]

def write out:&:sink:_elem, val:_elem -> out:&:sink:_elem [
  local-scope
  load-ingredients
  assert out, [write to null channel]
  chan:&:channel:_elem <- get *out, chan:offset
  <channel-write-initial>
  # block until lock is acquired AND queue has room
  lock:location <- get-location *chan, lock:offset
#?   $print [write], 10/newline
  {
#?     $print [trying to acquire lock for writing], 10/newline
    wait-for-reset-then-set lock
#?     $print [lock acquired for writing], 10/newline
    full?:bool <- channel-full? chan
    break-unless full?
#?     $print [but channel is full; relinquishing lock], 10/newline
    # channel is full; relinquish lock and give a reader the opportunity to
    # create room on it
    reset lock
    current-routine-is-blocked
    switch  # avoid spinlocking
    loop
  }
  current-routine-is-unblocked
#?   $print [performing write], 10/newline
  # store a deep copy of val
  circular-buffer:&:@:_elem <- get *chan, data:offset
  free:num <- get *chan, first-free:offset
  val-copy:_elem <- deep-copy val  # on this instruction rests all Mu's concurrency-safety
  *circular-buffer <- put-index *circular-buffer, free, val-copy
  # mark its slot as filled
  free <- add free, 1
  {
    # wrap free around to 0 if necessary
    len:num <- length *circular-buffer
    at-end?:bool <- greater-or-equal free, len
    break-unless at-end?
    free <- copy 0
  }
  # write back
  *chan <- put *chan, first-free:offset, free
#?   $print [relinquishing lock after writing], 10/newline
  reset lock
]

def read in:&:source:_elem -> result:_elem, eof?:bool, in:&:source:_elem [
  local-scope
  load-ingredients
  assert in, [read on null channel]
  eof? <- copy 0/false  # default result
  chan:&:channel:_elem <- get *in, chan:offset
  # block until lock is acquired AND queue has data
  lock:location <- get-location *chan, lock:offset
#?   $print [read], 10/newline
  {
#?     $print [trying to acquire lock for reading], 10/newline
    wait-for-reset-then-set lock
#?     $print [lock acquired for reading], 10/newline
    empty?:bool <- channel-empty? chan
    break-unless empty?
#?     $print [but channel is empty; relinquishing lock], 10/newline
    # channel is empty; relinquish lock and give a writer the opportunity to
    # add to it
    reset lock
    current-routine-is-blocked
    <channel-read-empty>
    switch  # avoid spinlocking
    loop
  }
  current-routine-is-unblocked
  # pull result off
  full:num <- get *chan, first-full:offset
  circular-buffer:&:@:_elem <- get *chan, data:offset
  result <- index *circular-buffer, full
  # clear the slot
  empty:&:_elem <- new _elem:type
  *circular-buffer <- put-index *circular-buffer, full, *empty
  # mark its slot as empty
  full <- add full, 1
  {
    # wrap full around to 0 if necessary
    len:num <- length *circular-buffer
    at-end?:bool <- greater-or-equal full, len
    break-unless at-end?
    full <- copy 0
  }
  # write back
  *chan <- put *chan, first-full:offset, full
#?   $print [relinquishing lock after reading], 10/newline
  reset lock
]

def clear in:&:source:_elem -> in:&:source:_elem [
  local-scope
  load-ingredients
  chan:&:channel:_elem <- get *in, chan:offset
  {
    empty?:bool <- channel-empty? chan
    break-if empty?
    _, _, in <- read in
  }
]

scenario channel-initialization [
  run [
    local-scope
    source:&:source:num <- new-channel 3/capacity
    chan:&:channel:num <- get *source, chan:offset
    10:num/raw <- get *chan, first-full:offset
    11:num/raw <- get *chan, first-free:offset
  ]
  memory-should-contain [
    10 <- 0  # first-full
    11 <- 0  # first-free
  ]
]

scenario channel-write-increments-free [
  local-scope
  _, sink:&:sink:num <- new-channel 3/capacity
  run [
    sink <- write sink, 34
    chan:&:channel:num <- get *sink, chan:offset
    10:num/raw <- get *chan, first-full:offset
    11:num/raw <- get *chan, first-free:offset
  ]
  memory-should-contain [
    10 <- 0  # first-full
    11 <- 1  # first-free
  ]
]

scenario channel-read-increments-full [
  local-scope
  source:&:source:num, sink:&:sink:num <- new-channel 3/capacity
  sink <- write sink, 34
  run [
    _, _, source <- read source
    chan:&:channel:num <- get *source, chan:offset
    10:num/raw <- get *chan, first-full:offset
    11:num/raw <- get *chan, first-free:offset
  ]
  memory-should-contain [
    10 <- 1  # first-full
    11 <- 1  # first-free
  ]
]

scenario channel-wrap [
  local-scope
  # channel with just 1 slot
  source:&:source:num, sink:&:sink:num <- new-channel 1/capacity
  chan:&:channel:num <- get *source, chan:offset
  # write and read a value
  sink <- write sink, 34
  _, _, source <- read source
  run [
    # first-free will now be 1
    10:num/raw <- get *chan, first-free:offset
    11:num/raw <- get *chan, first-free:offset
    # write second value, verify that first-free wraps
    sink <- write sink, 34
    20:num/raw <- get *chan, first-free:offset
    # read second value, verify that first-full wraps
    _, _, source <- read source
    30:num/raw <- get *chan, first-full:offset
  ]
  memory-should-contain [
    10 <- 1  # first-free after first write
    11 <- 1  # first-full after first read
    20 <- 0  # first-free after second write, wrapped
    30 <- 0  # first-full after second read, wrapped
  ]
]

scenario channel-new-empty-not-full [
  run [
    local-scope
    source:&:source:num <- new-channel 3/capacity
    chan:&:channel:num <- get *source, chan:offset
    10:bool/raw <- channel-empty? chan
    11:bool/raw <- channel-full? chan
  ]
  memory-should-contain [
    10 <- 1  # empty?
    11 <- 0  # full?
  ]
]

scenario channel-write-not-empty [
  local-scope
  source:&:source:num, sink:&:sink:num <- new-channel 3/capacity
  chan:&:channel:num <- get *source, chan:offset
  run [
    sink <- write sink, 34
    10:bool/raw <- channel-empty? chan
    11:bool/raw <- channel-full? chan
  ]
  memory-should-contain [
    10 <- 0  # empty?
    11 <- 0  # full?
  ]
]

scenario channel-write-full [
  local-scope
  source:&:source:num, sink:&:sink:num <- new-channel 1/capacity
  chan:&:channel:num <- get *source, chan:offset
  run [
    sink <- write sink, 34
    10:bool/raw <- channel-empty? chan
    11:bool/raw <- channel-full? chan
  ]
  memory-should-contain [
    10 <- 0  # empty?
    11 <- 1  # full?
  ]
]

scenario channel-read-not-full [
  local-scope
  source:&:source:num, sink:&:sink:num <- new-channel 1/capacity
  chan:&:channel:num <- get *source, chan:offset
  sink <- write sink, 34
  run [
    _, _, source <- read source
    10:bool/raw <- channel-empty? chan
    11:bool/raw <- channel-full? chan
  ]
  memory-should-contain [
    10 <- 1  # empty?
    11 <- 0  # full?
  ]
]

## cancelling channels

# every channel comes with a boolean signifying if it's been closed
# initially this boolean is false
container channel:_elem [
  closed?:bool
]

# a channel can be closed from either the source or the sink
# both routines can modify the 'closed?' bit, but they can only ever set it, so this is a benign race
def close x:&:source:_elem -> x:&:source:_elem [
  local-scope
  load-ingredients
  chan:&:channel:_elem <- get *x, chan:offset
  *chan <- put *chan, closed?:offset, 1/true
]
def close x:&:sink:_elem -> x:&:sink:_elem [
  local-scope
  load-ingredients
  chan:&:channel:_elem <- get *x, chan:offset
  *chan <- put *chan, closed?:offset, 1/true
]

# once a channel is closed from one side, no further operations are expected from that side
# if a channel is closed for reading,
#   no further writes will be let through
# if a channel is closed for writing,
#   future reads continue until the channel empties,
#   then the channel is also closed for reading
after <channel-write-initial> [
  closed?:bool <- get *chan, closed?:offset
  return-if closed?
]
after <channel-read-empty> [
  closed?:bool <- get *chan, closed?:offset
  {
    break-unless closed?
    empty-result:&:_elem <- new _elem:type
    current-routine-is-unblocked
    return *empty-result, 1/true
  }
]

## helpers

# An empty channel has first-empty and first-full both at the same value.
def channel-empty? chan:&:channel:_elem -> result:bool [
  local-scope
  load-ingredients
  # return chan.first-full == chan.first-free
  full:num <- get *chan, first-full:offset
  free:num <- get *chan, first-free:offset
  result <- equal full, free
]

# A full channel has first-empty just before first-full, wasting one slot.
# (Other alternatives: https://en.wikipedia.org/wiki/Circular_buffer#Full_.2F_Empty_Buffer_Distinction)
def channel-full? chan:&:channel:_elem -> result:bool [
  local-scope
  load-ingredients
  # tmp = chan.first-free + 1
  tmp:num <- get *chan, first-free:offset
  tmp <- add tmp, 1
  {
    # if tmp == chan.capacity, tmp = 0
    len:num <- capacity chan
    at-end?:bool <- greater-or-equal tmp, len
    break-unless at-end?
    tmp <- copy 0
  }
  # return chan.first-full == tmp
  full:num <- get *chan, first-full:offset
  result <- equal full, tmp
]

def capacity chan:&:channel:_elem -> result:num [
  local-scope
  load-ingredients
  q:&:@:_elem <- get *chan, data:offset
  result <- length *q
]

## helpers for channels of characters in particular

def buffer-lines in:&:source:char, buffered-out:&:sink:char -> buffered-out:&:sink:char, in:&:source:char [
  local-scope
  load-ingredients
  # repeat forever
  eof?:bool <- copy 0/false
  {
    line:&:buffer <- new-buffer 30
    # read characters from 'in' until newline, copy into line
    {
      +next-character
      c:char, eof?:bool, in <- read in
      break-if eof?
      # drop a character on backspace
      {
        # special-case: if it's a backspace
        backspace?:bool <- equal c, 8
        break-unless backspace?
        # drop previous character
        {
          buffer-length:num <- get *line, length:offset
          buffer-empty?:bool <- equal buffer-length, 0
          break-if buffer-empty?
          buffer-length <- subtract buffer-length, 1
          *line <- put *line, length:offset, buffer-length
        }
        # and don't append this one
        loop +next-character
      }
      # append anything else
      line <- append line, c
      line-done?:bool <- equal c, 10/newline
      break-if line-done?
      loop
    }
    # copy line into 'buffered-out'
    i:num <- copy 0
    line-contents:text <- get *line, data:offset
    max:num <- get *line, length:offset
    {
      done?:bool <- greater-or-equal i, max
      break-if done?
      c:char <- index *line-contents, i
      buffered-out <- write buffered-out, c
      i <- add i, 1
      loop
    }
    {
      break-unless eof?
      buffered-out <- close buffered-out
      return
    }
    loop
  }
]

scenario buffer-lines-blocks-until-newline [
  run [
    local-scope
    source:&:source:char, sink:&:sink:char <- new-channel 10/capacity
    _, buffered-stdin:&:sink:char/buffered-stdin <- new-channel 10/capacity
    buffered-chan:&:channel:char <- get *buffered-stdin, chan:offset
    empty?:bool <- channel-empty? buffered-chan
    assert empty?, [ 
F buffer-lines-blocks-until-newline: channel should be empty after init]
    # buffer stdin into buffered-stdin, try to read from buffered-stdin
    buffer-routine:num <- start-running buffer-lines, source, buffered-stdin
    wait-for-routine-to-block buffer-routine
    empty? <- channel-empty? buffered-chan
    assert empty?:bool, [ 
F buffer-lines-blocks-until-newline: channel should be empty after buffer-lines bring-up]
    # write 'a'
    sink <- write sink, 97/a
    restart buffer-routine
    wait-for-routine-to-block buffer-routine
    empty? <- channel-empty? buffered-chan
    assert empty?:bool, [ 
F buffer-lines-blocks-until-newline: channel should be empty after writing 'a']
    # write 'b'
    sink <- write sink, 98/b
    restart buffer-routine
    wait-for-routine-to-block buffer-routine
    empty? <- channel-empty? buffered-chan
    assert empty?:bool, [ 
F buffer-lines-blocks-until-newline: channel should be empty after writing 'b']
    # write newline
    sink <- write sink, 10/newline
    restart buffer-routine
    wait-for-routine-to-block buffer-routine
    empty? <- channel-empty? buffered-chan
    data-emitted?:bool <- not empty?
    assert data-emitted?, [ 
F buffer-lines-blocks-until-newline: channel should contain data after writing newline]
    trace 1, [test], [reached end]
  ]
  trace-should-contain [
    test: reached end
  ]
]

def drain source:&:source:char -> result:text, source:&:source:char [
  local-scope
  load-ingredients
  buf:&:buffer <- new-buffer 30
  {
    c:char, done?:bool <- read source
    break-if done?
    buf <- append buf, c
    loop
  }
  result <- buffer-to-array buf
]