<commit_before>require 'rubygems'
require 'bud'
require 'chord/chord_find'

module ChordJoin
  include ChordFind

  state {
    channel :join_req, [:@to, :requestor_addr] => [:start]
    table   :join_pending, join_req.keys => join_req.cols
    # table :finger, [:index] => [:start, 'hi', :succ, :succ_addr]
    # interface output, :succ_resp, [:key] => [:start, :addr]
    channel :finger_table_req, [:@to,:requestor_addr]
    channel :finger_table_resp, [:@requestor_addr] + finger.keys => finger.cols
    channel :pred_req, [:referrer_key, :referrer_index]
    channel :pred_resp, [:referrer_key, :referrer_index, :referrer_addr]
    channel :finger_upd, [:referrer_addr, :referrer_index, :my_start, :my_addr]
    table :offsets, [:val]
  }

  def log2(x)
    Math.log(x)/Math.log(2)
  end

  def bootstrap
    super
    offsets <= [(1..log2(@maxkey)).map{|o| o}]
  end


  declare
  def join_rules_proxy
    # an existing member serves as a proxy for the new node that wishes to join.
    # when it receives a join req from new node, it requests successors on the new
    # node's behalf

    # cache the request
    stdio <~ (join_pending <= join_req).inspected
    # asynchronously, find out who owns start+1
    succ_req <= join_req.map{|j| j.start}
    # upon response to successor request, ask the successor to send the contents
    #  of its finger table directly to the new node.
    finger_table_req <~ join([join_pending, succ_resp]).map do |j, s|
      [s.addr, j.from] if j.start+1 == s.key
    end
  end

  declare
  def join_rules_successor
    # at successor, upon receiving finger_table_req, ship finger table entries directly to new node
    finger_table_resp <~ join([finger_table_req, finger]).map do |ftreq, f|
      # finger tuple prefixed with requestor_addr
      [ftreq.requestor_addr] + f
    end
  end

  declare
  def join_rules_new_node
    # at new member, install finger entries
    finger <= finger_table_resp do |f|
      # construct tuple that contains all the finger columns in f
      (finger.keys + finger.cols).map{|c| f.send{c.to_sym}}
    end
    # update all nodes whose finger tables should refer here
    # first, for each offset o find last node whose o'th finger might be the new node's id
    pred_req <= join([me,offsets]).map do |m,o|
      [m.start - 2**o.val, o.val]
    end
    # upon pred_resp, send a finger_upd message to the node that was found to point here
    finger_upd <~ natjoin([me, pred_resp]).map do |m, resp|
      [resp.referrer_addr, resp.referrer_index, m.start, ip_port]
    end
  end

  declare
  def join_rules_referers
    # update finger entries upon a finger_upd if the new one works: insert new, delete old
    # XXX would be nice to have an update pattern in Bloom for this
    finger <+ join([finger_upd, finger], [finger_upd.referrer_index, finger.index]).map do |u, f|
      [f.index, f.start, f.hi, u.my_start, u.from] if in_range(u.my_start, f.start, f.hi)
    end
    finger <- join([finger_upd, finger], [finger_upd.referrer_index, finger.index]).map do |u, f|
      f if in_range(u.my_start, f.start, f.hi)
    end
  end
end
<commit_msg>Fix likely bug in Chord.<commit_after>require 'rubygems'
require 'bud'
require 'chord/chord_find'

module ChordJoin
  include ChordFind

  state {
    channel :join_req, [:@to, :requestor_addr] => [:start]
    table   :join_pending, join_req.keys => join_req.cols
    # table :finger, [:index] => [:start, 'hi', :succ, :succ_addr]
    # interface output, :succ_resp, [:key] => [:start, :addr]
    channel :finger_table_req, [:@to,:requestor_addr]
    channel :finger_table_resp, [:@requestor_addr] + finger.keys => finger.cols
    channel :pred_req, [:referrer_key, :referrer_index]
    channel :pred_resp, [:referrer_key, :referrer_index, :referrer_addr]
    channel :finger_upd, [:referrer_addr, :referrer_index, :my_start, :my_addr]
    table :offsets, [:val]
  }

  def log2(x)
    Math.log(x)/Math.log(2)
  end

  def bootstrap
    super
    offsets <= [(1..log2(@maxkey)).map{|o| o}]
  end


  declare
  def join_rules_proxy
    # an existing member serves as a proxy for the new node that wishes to join.
    # when it receives a join req from new node, it requests successors on the new
    # node's behalf

    # cache the request
    stdio <~ (join_pending <= join_req).inspected
    # asynchronously, find out who owns start+1
    succ_req <= join_req.map{|j| j.start}
    # upon response to successor request, ask the successor to send the contents
    #  of its finger table directly to the new node.
    finger_table_req <~ join([join_pending, succ_resp]).map do |j, s|
      [s.addr, j.from] if j.start+1 == s.key
    end
  end

  declare
  def join_rules_successor
    # at successor, upon receiving finger_table_req, ship finger table entries directly to new node
    finger_table_resp <~ join([finger_table_req, finger]).map do |ftreq, f|
      # finger tuple prefixed with requestor_addr
      [ftreq.requestor_addr] + f
    end
  end

  declare
  def join_rules_new_node
    # at new member, install finger entries
    finger <= finger_table_resp do |f|
      # construct tuple that contains all the finger columns in f
      (finger.keys + finger.cols).map{|c| f.send{c.to_sym}}
    end
    # update all nodes whose finger tables should refer here
    # first, for each offset o find last node whose o'th finger might be the new node's id
    pred_req <~ join([me,offsets]).map do |m,o|
      [m.start - 2**o.val, o.val]
    end
    # upon pred_resp, send a finger_upd message to the node that was found to point here
    finger_upd <~ natjoin([me, pred_resp]).map do |m, resp|
      [resp.referrer_addr, resp.referrer_index, m.start, ip_port]
    end
  end

  declare
  def join_rules_referers
    # update finger entries upon a finger_upd if the new one works: insert new, delete old
    # XXX would be nice to have an update pattern in Bloom for this
    finger <+ join([finger_upd, finger], [finger_upd.referrer_index, finger.index]).map do |u, f|
      [f.index, f.start, f.hi, u.my_start, u.from] if in_range(u.my_start, f.start, f.hi)
    end
    finger <- join([finger_upd, finger], [finger_upd.referrer_index, finger.index]).map do |u, f|
      f if in_range(u.my_start, f.start, f.hi)
    end
  end
end

{'fraction_non_alphanumeric': 0.10819881732459645, 'fraction_numerical': 0.002237494006712482, 'mean_word_length': 3.3823529411764706, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 13, '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': '1831990', 'n_tokens_mistral': 2183, 'n_tokens_neox': 2134, 'n_words': 680}