<commit_before>require "rubygems"
require "bud"

$nodes = []
$addr_map = {}

class VcAgent
  include Bud

  state do
    scratch :kickoff, [] => [:v]
    scratch :done, [] => [:v]

    scratch :to_send, [:addr, :msg, :slow]
    channel :chn, [:@addr, :msg, :from, :clock]
    table :send_buf, [:addr, :msg, :send_at_time]
    scratch :buf_chosen, send_buf.schema
    periodic :tik, 1

    lat_map :my_vc
  end

  bootstrap do
    my_vc <= [[ip_port, MaxLattice.wrap(0)]]
  end

  bloom do
    stdio <~ chn {|c| ["Got message @ #{port} (# = #{$addr_map[ip_port]}): #{c.inspect}"]}

    to_send <= kickoff { [$nodes[2].ip_port, 1, true]}
    to_send <= kickoff { [$nodes[1].ip_port, 2, false]}

    send_buf <= to_send {|s| [s.addr, s.msg, s.slow ? (@budtime + 2) : @budtime]}
    buf_chosen <= send_buf {|s| s if s.send_at_time == @budtime}
    send_buf <- buf_chosen
    chn <~ buf_chosen {|s| [s.addr, s.msg, $addr_map[ip_port], Marshal.dump(my_vc)]}

    # When we get a message, bump the local VC
    my_vc <+ chn {|c| [ip_port, MaxLattice.wrap(my_vc[ip_port].reveal + 1)]}
    my_vc <+ chn {|c| Marshal.load(c.clock)}

    to_send <= chn {|c| [$nodes[2].ip_port, 3, false] if $addr_map[ip_port] == 1}
    done <= chn {|c| [true] if ($addr_map[ip_port] == 2 and c.msg == 1)}
    stdio <~ my_vc.inspected(self)
  end
end

3.times do |i|
  b = VcAgent.new
  b.run_bg
  $addr_map[b.ip_port] = i
  $nodes << b
  puts "Started: #{b.port}"
end

q = Queue.new
$nodes.last.register_callback(:done) do |t|
  q.push(true)
end

n = $nodes.first
n.sync_do {
  n.kickoff <+ [[true]]
}

q.pop

$nodes.each do |n|
  n.stop
end<commit_msg>Checkpoint work.<commit_after>require "rubygems"
require "bud"

$nodes = []
$addr_map = {}

class VcAgent
  include Bud

  state do
    scratch :kickoff, [] => [:v]
    scratch :done, [] => [:v]

    scratch :to_send, [:addr, :msg, :slow]
    channel :chn, [:@addr, :msg, :from, :clock]
    table :send_buf, [:addr, :msg, :send_at_time]
    scratch :buf_chosen, send_buf.schema
    periodic :tik, 1

    lat_map :my_vc
  end

  bootstrap do
    my_vc <= [[ip_port, MaxLattice.wrap(0)]]
  end

  bloom do
    stdio <~ chn {|c| ["Got message @ #{port} (# = #{$addr_map[ip_port]}), ID = #{c.msg}"]}

    # Setup a specific messaging scenario: node 3 will (usually) receive message
    # 3 and then 1, violating causal order
    to_send <= kickoff { [$nodes[2].ip_port, 1, true]}
    to_send <= kickoff { [$nodes[1].ip_port, 2, false]}
    to_send <= chn {|c| [$nodes[2].ip_port, 3, false] if $addr_map[ip_port] == 1}
    done <= chn {|c| [true] if ($addr_map[ip_port] == 2 and c.msg == 1)}

    send_buf <= to_send {|s| [s.addr, s.msg, s.slow ? (@budtime + 2) : @budtime]}
    buf_chosen <= send_buf {|s| s if s.send_at_time == @budtime}
    send_buf <- buf_chosen
    chn <~ buf_chosen {|s| [s.addr, s.msg, $addr_map[ip_port], Marshal.dump(my_vc)]}

    # When we get a message, bump the local VC
    my_vc <+ chn {|c| [ip_port, MaxLattice.wrap(my_vc[ip_port].reveal + 1)]}
    my_vc <+ chn {|c| Marshal.load(c.clock)}
  end
end

3.times do |i|
  b = VcAgent.new
  b.run_bg
  $addr_map[b.ip_port] = i
  $nodes << b
end

puts "Started: #{$nodes.map{|n| n.port}.inspect}"

q = Queue.new
$nodes.last.register_callback(:done) do |t|
  q.push(true)
end

n = $nodes.first
n.sync_do {
  n.kickoff <+ [[true]]
}

q.pop

$nodes.each do |n|
  n.sync_do {
    puts n.my_vc.inspected(n).inspect
  }
  n.stop
end
{'fraction_non_alphanumeric': 0.18404907975460122, 'fraction_numerical': 0.009056383289512124, 'mean_word_length': 3.274656679151061, 'pattern_counts': {'":': 0, '<': 34, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3547036', 'n_tokens_mistral': 1528, 'n_tokens_neox': 1463, 'n_words': 357}