AoC 2019/07, but it crashes the Pony runtime

cpu.pony

use "collections"

interface Sendable
  fun tag send(word: I64)

primitive Running
primitive Blocked
primitive Halted
primitive IllegalInstruction

type Status is (Running | Blocked | Halted | IllegalInstruction)

actor Cpu is Sendable
  var _memory: Memory
  var _ip: USize = 0
  var _input: Array[I64 val] = Array[I64 val]
  var _status: Status = Running
  var _subscribers: SetIs[Sendable tag] = SetIs[Sendable tag]
  var _on_halt: SetIs[{ref ()} iso] = SetIs[{ref ()} iso]
  var _on_crash: SetIs[{ref ()} iso] = SetIs[{ref ()} iso]
  var _running: Bool = false

  new create(memory: Memory iso) =>
    _memory = consume memory

  be subscribe(rcvr: Sendable tag) =>
    _subscribers.set(rcvr)

  be unsubscribe(rcvr: Sendable tag) =>
    _subscribers.unset(rcvr)

  be subscribe_halt(notify: {ref ()} iso) =>
    _on_halt.set(consume notify)

  be unsubscribe_halt(notify: {ref ()} tag) =>
    _on_halt.unset(notify)

  be subscribe_crash(notify: {ref ()} iso) =>
    _on_crash.set(consume notify)

  be unsubscribe_crash(notify: {ref ()} tag) =>
    _on_crash.unset(notify)

  be step() =>
    if not (_status is Running) then
      return
    end

    try
      (let instr, let ip') = _memory.decode(_ip)?
      match instr
      | (Add, let lhs: Src, let rhs: Src, let dst: Dst) =>
          _write(dst, _read(lhs) + _read(rhs))
          _ip = ip'
      | (Mul, let lhs: Src, let rhs: Src, let dst: Dst) =>
          _write(dst, _read(lhs) * _read(rhs))
          _ip = ip'
      | (Inp, let dst: Dst) =>
          try
            _write(dst, _input.shift()?)
            _ip = ip'
          else
            _status = Blocked
          end
      | (Out, let src: Src) =>
          let word = _read(src)
          for rcvr in _subscribers.values() do
            rcvr.send(word)
          end
          _ip = ip'
      | (Jnz, let cnd: Src, let tgt: Src) =>
          _ip = if _read(cnd) != 0 then
            let tgt' = _read(tgt)
            if tgt' < 0 then error else tgt'.usize() end
          else
            ip'
          end
      | (Jz, let cnd: Src, let tgt: Src) =>
          _ip = if _read(cnd) == 0 then
            let tgt' = _read(tgt)
            if tgt' < 0 then error else tgt'.usize() end
          else
            ip'
          end
      | (Lt, let lhs: Src, let rhs: Src, let dst: Dst) =>
          _write(dst, if _read(lhs) < _read(rhs) then 1 else 0 end)
          _ip = ip'
      | (Eq, let lhs: Src, let rhs: Src, let dst: Dst) =>
          _write(dst, if _read(lhs) == _read(rhs) then 1 else 0 end)
          _ip = ip'
      | Hlt =>
          _status = Halted
          _ip = ip'
          for notify in _on_halt.values() do
            notify()
          end
      end
    else
      _status = IllegalInstruction
      for notify in _on_crash.values() do
        notify()
      end
    end

    if _running and (_status is Running) then
      step()
    end

  be run() =>
    if _status is Running then
      _running = true
      step()
    end

  be send(word: I64) =>
    _input.push(word)
    if _status is Blocked then
      _status = Running
      if _running then
        step()
      end
    end

  fun box _read(src: Src): I64 =>
    (let mode, let value) = src
    match mode
    | Imm => value
    | Mem => _memory(value.usize())
    end

  fun ref _write(dst: Dst, word: I64) =>
    (let mode, let value) = dst
    match mode
    | Mem => _memory(value.usize()) = word
    end

input.txt

3,8,1001,8,10,8,105,1,0,0,21,42,63,76,101,114,195,276,357,438,99999,3,9,101,2,9,9,102,5,9,9,1001,9,3,9,1002,9,5,9,4,9,99,3,9,101,4,9,9,102,5,9,9,1001,9,5,9,102,2,9,9,4,9,99,3,9,1001,9,3,9,1002,9,5,9,4,9,99,3,9,1002,9,2,9,101,5,9,9,102,3,9,9,101,2,9,9,1002,9,3,9,4,9,99,3,9,101,3,9,9,102,2,9,9,4,9,99,3,9,1001,9,2,9,4,9,3,9,102,2,9,9,4,9,3,9,101,2,9,9,4,9,3,9,102,2,9,9,4,9,3,9,101,1,9,9,4,9,3,9,1001,9,2,9,4,9,3,9,1001,9,1,9,4,9,3,9,1001,9,2,9,4,9,3,9,1001,9,2,9,4,9,3,9,1001,9,1,9,4,9,99,3,9,102,2,9,9,4,9,3,9,1001,9,2,9,4,9,3,9,102,2,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,1001,9,1,9,4,9,3,9,102,2,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,102,2,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,102,2,9,9,4,9,99,3,9,102,2,9,9,4,9,3,9,102,2,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,1001,9,2,9,4,9,3,9,1002,9,2,9,4,9,3,9,1001,9,1,9,4,9,3,9,1002,9,2,9,4,9,3,9,1002,9,2,9,4,9,3,9,101,2,9,9,4,9,3,9,1001,9,2,9,4,9,99,3,9,1001,9,1,9,4,9,3,9,101,2,9,9,4,9,3,9,102,2,9,9,4,9,3,9,1001,9,2,9,4,9,3,9,1001,9,1,9,4,9,3,9,102,2,9,9,4,9,3,9,1001,9,2,9,4,9,3,9,1001,9,2,9,4,9,3,9,102,2,9,9,4,9,3,9,1001,9,2,9,4,9,99,3,9,102,2,9,9,4,9,3,9,101,1,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,1002,9,2,9,4,9,3,9,1002,9,2,9,4,9,3,9,101,2,9,9,4,9,3,9,1001,9,2,9,4,9,3,9,101,2,9,9,4,9,3,9,1002,9,2,9,4,9,3,9,101,2,9,9,4,9,99

instructions.pony

primitive Mem
primitive Rel
primitive Imm

type Dst is (Mem, I64)
type Src is ((Mem | Imm), I64)

primitive Add
primitive Mul
primitive Inp
primitive Out
primitive Jnz
primitive Jz
primitive Lt
primitive Eq
primitive Hlt

type Instruction is (
    (Add, Src, Src, Dst)
  | (Mul, Src, Src, Dst)
  | (Inp, Dst)
  | (Out, Src)
  | (Jnz, Src, Src)
  | (Jz, Src, Src)
  | (Lt, Src, Src, Dst)
  | (Eq, Src, Src, Dst)
  | Hlt
)

main.pony

use "collections"
use "files"
use "promises"

use "debug"

actor Main
  new create(env: Env) =>
    let path = try
      env.args(1)?
    else
      env.exitcode(1)
      let name = try env.args(0)? else "intpony" end
      env.err.print("Usage: " + name + " code-file")
      return
    end

    let file = match OpenFile(FilePath(FileAuth(env.root), path))
    | let f: File => FileLines(f)
    else
      env.exitcode(1)
      env.err.print("Unable to read \"" + path + "\"")
      return
    end

    let code = recover Array[I64] end
    for line in file do
      for word in line.split(",").values() do
        try
          code.push(word.i64()?)
        else
          env.exitcode(1)
          env.err.print("Unable to parse \"" + word + "\"")
          return
        end
      end
    end
    let mem = recover val Memory(consume code) end

    let p1 = problem1(mem, env)
    p1.next[None]({(max: I64)(env) =>
      env.out.print("Problem 1: " + max.string())
    } iso)

  fun problem1(mem: Memory val, env: Env): Promise[I64] =>
    let mk = MaxKeeper
    let p = Promise[I64]
    let waiter = CpuWaiter({()(p) =>
      mk.query({(max: (I64 | None)) =>
        match max
        | let m: I64 => p(m)
        else
          p.reject()
        end
      })
    })


    /* the real code cycles permutations of these, but it's
     *   distracting and not required to crash.
     * the goal is to capture the max of the last output
     *   from the last CPU in the chain across all permutations of "phases".
     */
    let phases = [as U8: 0; 1; 2; 3; 4]

    // the larger this number is, the more likely a segfault
    for iter in Range(0, 100) do
      var cpus = Array[Cpu]
      for (i, phase) in phases.pairs() do
        let mem' = mem.clone()
        let cpu = Cpu(consume mem')

        if i > 0 then
          try
            cpus(cpus.size() - 1)?.subscribe(cpu)
          end
        end

        cpu.subscribe_crash({() =>
          env.exitcode(1)
          env.err.print("cpu " + i.string() + " crashed")
        })

        cpu.send(phase.i64())

        cpus.push(cpu)
      end

      try
        cpus(cpus.size() - 1)?.subscribe(mk)
        waiter.wait(cpus(cpus.size() - 1)?)
      end

      try
        cpus(0)?.send(0)
      end

      for cpu in cpus.values() do
        Debug("RUN")
        cpu.run()
      end
    end
    waiter.start_waiting()
    p

actor MaxKeeper
  var _max: (I64 | None) = None

  be send(word: I64) =>
    _max = match _max
    | let m: I64 => m.max(word)
    else
      word
    end

  be query(fn: {((I64 | None))} val) =>
    fn(_max)

actor CpuWaiter
  var _waiting: SetIs[Cpu tag] = SetIs[Cpu tag]
  var _wait: Bool = false
  let _when_done: {()} val

  new create(when_done: {()} val) =>
    _when_done = when_done

  be wait(cpu: Cpu tag) =>
    _waiting.set(cpu)
    cpu.subscribe_halt({()(self = recover tag this end) =>
      Debug("DONE HALT")
      self._done(cpu)
    })
    cpu.subscribe_crash({()(self = recover tag this end) =>
      Debug("DONE CRASH")
      self._done(cpu)
    })

  be start_waiting() =>
    _wait = true
    if _waiting.size() == 0 then
      Debug("already done, triggering")
      _when_done()
    end

  be _done(cpu: Cpu tag) =>
    _waiting.unset(cpu)
    if _wait and (_waiting.size() == 0) then
      Debug("done, triggering")
      _when_done()
    end

memory.pony

use "itertools"

type _Mode is (Mem | Imm)

class Memory
  var _memory: Array[I64]

  new create(image: Array[I64]) =>
    _memory = image

  fun box clone(): Memory iso^ =>
    let sz = _memory.size()
    let memory' = recover Array[I64](sz) end
    for m in _memory.values() do
      memory'.push(m)
    end
    recover Memory(consume memory') end

  /* NOTE: this originally implemented an "expanding" memory, but the problem
   *   doesn't exercise it, and it's not required to trigger the crash, so it's
   *   been removed.
   */

  fun box apply(i: USize val): I64 =>
    try
      _memory(i)?
    else
      0
    end

  fun ref update(i: USize val, value: I64) =>
    try
      _memory(i)? = value
    end

  fun box decode(ip: USize val): (Instruction, USize)? =>
    let word = this(ip)
    let op = word % 100
    let mode1 = _DecodeMode((word / 100) % 10)?
    let mode2 = _DecodeMode((word / 1000) % 10)?
    let mode3 = _DecodeMode((word / 10000) % 10)?
    match op
    | 1 => ((Add,
        _decode_src(ip + 1, mode1),
        _decode_src(ip + 2, mode2),
        _decode_dst(ip + 3, mode3)?
      ), ip + 4)
    | 2 => ((Mul,
        _decode_src(ip + 1, mode1),
        _decode_src(ip + 2, mode2),
        _decode_dst(ip + 3, mode3)?
      ), ip + 4)
    | 3 => ((Inp,
        _decode_dst(ip + 1, mode1)?
      ), ip + 2)
    | 4 => ((Out,
        _decode_src(ip + 1, mode1)
      ), ip + 2)
    | 5 => ((Jnz,
        _decode_src(ip + 1, mode1),
        _decode_src(ip + 2, mode2)
      ), ip + 3)
    | 6 => ((Jz,
        _decode_src(ip + 1, mode1),
        _decode_src(ip + 2, mode2)
      ), ip + 3)
    | 7 => ((Lt,
        _decode_src(ip + 1, mode1),
        _decode_src(ip + 2, mode2),
        _decode_dst(ip + 3, mode3)?
      ), ip + 4)
    | 8 => ((Eq,
        _decode_src(ip + 1, mode1),
        _decode_src(ip + 2, mode2),
        _decode_dst(ip + 3, mode3)?
      ), ip + 4)
    | 99 => (Hlt, ip + 1)
    else
      error
    end

  fun box _decode_src(ptr: USize val, mode: _Mode): Src =>
    match mode
    | Mem => (Mem, this(ptr))
    | Imm => (Imm, this(ptr))
    end

  fun box _decode_dst(ptr: USize val, mode: _Mode): Dst? =>
    match mode
    | Mem => (Mem, this(ptr))
    else
      error
    end

primitive  _DecodeMode
  fun apply(digit: I64): _Mode? =>
    match digit
    | 0 => Mem
    | 1 => Imm
    else
      error
    end

notes.txt

Run: intpony.exe input.txt

The three observed behaviors are:
  - Almost instant exit with no output (~60-70% of the time)
  - Almost instant exit with correct output (~15% of the time)
  - Slooooow exit AFTER correct output (~15% of the time)

The only runtime option I've found to have any effect on this is --ponymaxthreads 1, which seemingly guarantees the intended output (with fast exit).

Compiled with -d, I get additional possible outcomes including mismatches between the count of RUN outputs and DONE outputs.

The program ends in a segfault, usually, on either the release or debug binary. It's often reported (by gdb) in "pony_os_peername" on Windows, and in "Array_I64_val_Trace" on WSL2. Oddly, runs with no segfault also have no output, and successful runs produce output before segfaulting. With --ponymaxthreads 1, no segfault on Windows, but I still get segfaults on WSL2.

FYI we create 100 Cpu actors total, each with a 519-"word" memory (i.e. an Array[I64] with 519 entries).

Full "bt" from a crash on WSL2:
#0  0x0000555555568160 in Array_I64_val_Trace ()
#1  0x000055555556895d in Array_u3_t2_$1$10_iso_$1$10_iso_collections__MapEmpty_val_collections__MapDeleted_val_Trace
    ()
#2  0x00007fffe6807600 in ?? ()
#3  0x00007fffe657d600 in ?? ()
#4  0x00007fffffffe750 in ?? ()
#5  0x00007ffff7c91c48 in ?? ()
#6  0x00007ffff7c91e00 in ?? ()
#7  0x0000000000000700 in ?? ()
#8  0x000055555557d534 in ponyint_actor_final ()
#9  0x000055555557f07b in ponyint_cycle_terminate ()
#10 0x000055555558726f in ponyint_sched_shutdown ()
#11 0x0000555555585acf in ponyint_sched_start ()
#12 0x00005555555877d9 in pony_start ()
#13 0x000055555557ce17 in main ()

I've also seen:
#0  0x00007fffe671ccc0 in ?? ()
#1  0x000055555557d534 in ponyint_actor_final ()
#2  0x000055555557f07b in ponyint_cycle_terminate ()
#3  0x000055555558726f in ponyint_sched_shutdown ()
#4  0x0000555555585acf in ponyint_sched_start ()
#5  0x00005555555877d9 in pony_start ()
#6  0x000055555557ce17 in main ()

My current thinking is that a segfault is occurring in the runtime during the final wrap-up, which sometimes interrupts the output. If I'm following the name mangling, it might be in the cleanup for a `SetIs` of either Cpu actors or objects/actors which close over Cpu tag references; these occur in a few places.