chord.rb

#!/usr/bin/env ruby

require 'eventmachine'
require 'securerandom'
require 'optparse'
require 'ostruct'

class Chord
  BITS = 5
end

# Represents the local or remote node
class Node
  attr_accessor :id, :host, :port
  include EventMachine::Deferrable

  def initialize(id, host, port)
    @id = id
    @host = host
    @port = port
  end

  # This is the messaging implementation
  # The dht protocol is implemented in the Client class
  def find_successor(id)
    # Send successor message to node and return answer
    puts "--- asking #{@host}:#{@port} SUCCESSOR of #{id}"
    return EventMachine.connect("www.google.com", 80, NodeConnection)
  end

  def predecessor()
    # Asks the node what is its predecessor
    puts "--- asking #{@host}:#{@port} PREDECESSOR"
    return Node.new(SecureRandom.random_number(2**Chord::BITS), "localhost", 9999)
  end

  def notify(id)
    # Notifies the node we're its predecessor
    puts "--- sending #{@host}:#{@port} NOTIFY #{id}"
  end
end

class NodeConnection < EventMachine::Connection
  def post_init()
    send_data "GET / HTTP/1.1\r\nHost: _\r\n\r\n"
  end

  def receive_data(data)
    puts data.inspect
    close_connection()
  end

  def unbind()
    puts "--- node connection unbound"
  end
end

class Client
  attr_accessor :finger

  def initialize(options = OpenStruct.new)
    # This key is for testing purposes only
    id = SecureRandom.random_number(2**Chord::BITS)

    # Creating this client's node object
    @node = Node.new(id, "localhost", 10000 + id)

    # While the client doesn't join a ring, this is pretty much true
    @predecessor = nil
    @successor = @node

    # Creating this client's finger table with the known succesors
    # (which, until we join something or be joined by someone, is
    # always this node itself, so ronery, sosad)
    @finger = Chord::BITS.times.map do |i|
      @node
    end

    # Creating a "server" to listen for incoming requests
    EventMachine.run {
      EventMachine.start_server(@node.host, @node.port, ClientConnection) do |con|
        con.client = self
      end

      puts "Client listening on #{@node.host}:#{@node.port}"
      puts "Client finger: #{@finger}"

      if(options.host)
        host = options.host.split(/:/).first
        port = options.host.split(/:/).last.to_i
        node = Node.new(nil, host, port)

        join(node)
      end

      timer = EventMachine::PeriodicTimer.new(5) do
        stabilize()
        fix_fingers()
        check_predecessor()
      end
    }
  end

  def find_successor(id)
    if(in_range?(id, @node.id, @successor))
      return @successor
    else
      deferred = closest_preceding_node(id).find_successor(id)
    end
  end

  def closest_preceding_node(id)
    Chord::BITS.downto(1).each do |i|
      if(in_range?(@finger[i].id, @node.id, id))
        return @finger[i]
      end
    end

    return @node
  end

  def stabilize()
    n = @successor.predecessor()

    if(in_range?(n.id, @node.id, @successor.id))
      @successor = n
    end

    @successor.notify(@node.id)
  end

  def notify(node)
    if(@predecessor.nil? or in_range?(node.id, @predecessor.id, @node.id))
      @predecessor = node
    end
  end

  def fix_fingers()
    puts "--- fixing fingers table"
  end

  def check_predecessor()
    puts "--- checking predecessor for connectivity"
  end

  def join(node)
    @predecessor = nil
    @successor = node.find_successor(@node.id)
  end

  #######
  private
  #######

  def in_range?(key, lower, upper)
    if(lower <= upper)
      return ((lower..upper) === key)
    else
      return ((lower..Chord::BITS) === key or (0..upper) === key)
    end
  end
end

class ClientConnection < EventMachine::Connection
  attr_accessor :client

  def receive_data(data)
    puts "--- client: #{@client.inspect}"
    puts "--- received data: #{data}"
  end

  def unbind
    puts "--- client disconnected!"
  end
end

# Tests
options = OpenStruct.new

OptionParser.new do |opts|
  opts.banner = "Usage: chord.rb [options]"

  opts.on("-j HOST", "--join HOST", String, "One node of the DHT to join (format: host:port)") do |h|
    options.host = h
  end
end.parse!

c = Client.new(options)