Distributed Hash Table

　　Chord Algorithm is used to search the Distributed Hash Table (DHT) in a Peer-to-Peer Network, where a peer only remembers the addresses of a few companions but can get access to any peer who is responsible for a given key. This demo program is largely based on such thinking although there may be some modifications and simplifications.

　　Moreover, in this program I used Java UDP sockets to simulate a simple reliable data transfer protocol, which can deal with bit error, packet error and packet loss. A message is encapsulated with a SEQ number, an ACK number and a CRC checksum. To simplify the application layer programming, I provide a Client class and a Server class with some convenient APIs. A Server should always listen and answer alternatively, and can be informed or inquired by calling a Client method.

 

　　Makefile:

all: ./bin/Chord.class ./bin/Peer.class

./bin/Chord.class: ./src/Chord.java ./src/Peer.java
    javac -classpath ./bin/ -d ./bin/ ./src/Chord.java

./bin/Peer.class: ./src/Peer.java
    javac -d ./bin/ ./src/Peer.java

clean:
    rm ./bin/*.class

1. Chord.java

import java.util.concurrent.*;
import java.util.*;
import java.io.*;


public class Chord extends Thread {
    private static int num;
    private static Process[] p;
    private static Chord[] t;
    private static int[] port;
    private BufferedReader in;
    private PrintWriter out;
    
    public Chord(Process proc) {
        in = new BufferedReader(new InputStreamReader(proc.getInputStream()));
        out = new PrintWriter(new OutputStreamWriter(proc.getOutputStream()));
        start();
    }
    public void run() {
        try {
            out.println(num);
            for (int i=0;i<num;i++) {
                out.println(port[i]);
            }
            out.close();
            String str = null;
            while (true) {
                str = in.readLine();
                if (str.charAt(0)!='#') {
                    System.out.println(str);
                } else {
                    break;
                }
            }
            in.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) {
        num = Integer.parseInt(args[0]);
        p = new Process[num];
        t = new Chord[num];
        port = new int[num];
        Random rand = new Random();
        for (int i=0;i<num;i++) {
            port[i] = 1024+rand.nextInt(8192);
        }
        System.out.println("
Wait Please ~
");
        try {
            for (int i=0;i<num;i++) {
                /* Thread t[i] is resp for the standard IO of Process p[i] */
                p[i] = Runtime.getRuntime().exec("java -classpath ./bin/ Peer "+port[i]);
                t[i] = new Chord(p[i]);
                TimeUnit.MILLISECONDS.sleep(100);
            }
            for (int i=0;i<num;i++) {
                p[i].waitFor();
                t[i].join();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println();
    }
}

2. Peer.java

import java.util.concurrent.atomic.*;
import java.util.concurrent.*;
import java.util.*;
import java.text.*;
import java.net.*;
import java.io.*;


abstract class RDT {
    /** RDT is a class using UDP sockets to simulate
     *        reliable data transfer that implements packet
     *        loss/error detection and retransmission.
     *    An RDT message = SEQ + ACK + content + CRC checksum
     *        if (SEQ==0)
     *            no reply is expected, a client "ACK" message
     *        else
     *            a reply with the identical ACK is expected
     */
    
    private static final int GEN = 0x04C11DB7;
    private static final int SIZE = 256;
    protected static final int TIME_LIMIT = 10;
    protected static Random rand;
    protected InetAddress addr;
    protected int targetPort;
    protected DatagramSocket socket;
    private DatagramPacket packet;
    private byte[] buffer;
    private byte ack = 0;
        
    static {
        rand = new Random();
    }
    protected String rcvd() {
        /* Receive a message with desired ACK # */
        try {
            do {
                buffer = new byte[SIZE];
                packet = new DatagramPacket(buffer,SIZE);
                socket.receive(packet);
            } while (checksum()!=0||buffer[1]!=ack);
            ack = buffer[0];    // next ACK # to be written
            if (ack!=0) {
                addr = packet.getAddress();
                targetPort = packet.getPort();
            }
            return new String(buffer,"ASCII").substring(2,SIZE-4);
        } catch (Exception e) {
            // Timeout Exception: the packet is lost!
            return null;
        }
    }
    protected void send(byte seq,String str) {
        /* Send a message with 'best effort' */
        buffer = new byte[SIZE];
        buffer[0] = seq;
        buffer[1] = ack;
        try {    // write the data content
            byte[] tmp = str.getBytes("ASCII");
            System.arraycopy(tmp,0,buffer,2,tmp.length);
        } catch (Exception e) {
            e.printStackTrace();
        }
        // write the checksum field
        int cks = checksum();
        buffer[SIZE-4] = (byte)(cks&0xFF);
        cks >>= 8;
        buffer[SIZE-3] = (byte)(cks&0xFF);
        cks >>= 8;
        buffer[SIZE-2] = (byte)(cks&0xFF);
        cks >>= 8;
        buffer[SIZE-1] = (byte)(cks&0xFF);
        // encapsulate and send the packet
        packet = new DatagramPacket(buffer,SIZE,addr,targetPort);
        try {
            socket.send(packet);
        } catch (Exception e) {
            System.out.println("Cannot send msg: "+str);
        }
        // next ACK # to be read
        ack = buffer[0];
    }
    private int checksum() {
        /* Calculate the CRC checksum */
        int rem = 0, N = (SIZE<<3);
        for (int pos=0;pos<N;pos++) {
            byte tmp = (byte)(buffer[pos>>3]>>(pos&7));
            if (rem>=0) {
                rem = (rem<<1)|(tmp&1);
            } else {
                rem = (rem<<1)|(tmp&1);
                rem ^= GEN;
            }
        }
        return revBits(rem);
    }
    private int revBits(int val) {
        /* Reverse the bits of an integer */
        int tmp = 0;
        for (int i=0;i<16;i++) {
            tmp |= (((1<<i)&val)<<(31-i-i));
            tmp |= (((1<<(31-i))&val)>>>(31-i-i));
        }
        return tmp;
    }
    protected static boolean isACK(String str) throws Exception {
        /* Test whether a string is equal to "ACK" */
        if (str==null) {
            return false;
        } else {
            byte[] b = str.getBytes("ASCII");
            return (b.length>=3 && b[0]=='A'
                && b[1]=='C' && b[2]=='K');
        }
    }
}


class Client extends RDT {
    /**    An RDT client should always send a message first,
     *        then receives a reply, and finally send an "ACK"
     *         message to terminate the conversation.
     *     An "ACK" message from the client indicates that
     *        it has received the reply and left.
     */
    
    private Client(int port) {
        try {
            addr = InetAddress.getByName("localhost");
            targetPort = port;
            socket = new DatagramSocket();
            socket.setSoTimeout(TIME_LIMIT);
        } catch(Exception e) {
            e.printStackTrace();
        }
    }
    public static String inquire(int port,String str) {
        /* Send a inquiry and return a pending reply */
        Client c = new Client(port);
        String res = null;
        while (res==null) {
            c.send((byte)(1+rand.nextInt(255)),str);
            res = c.rcvd();
        }
        c.send((byte)0,"ACK");
        c.socket.close();
        return res;
    }
    public static void inform(int port,String str) {
        /* Send a piece of info and get an "ACK" */
        Client c = new Client(port);
        String res = null;
        try {
            while (!isACK(res)) {
                c.send((byte)(1+rand.nextInt(255)),str);
                res = c.rcvd();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        c.send((byte)0,"ACK");
        c.socket.close();
    }
}


class Server extends RDT {
    /**    An RDT server should always listen to a request 
     *        first, then respond with a reply, and finally
     *        get an "ACK" message.
     *    An "ACK" message from the server indicates that
     *        it has been informed of something.
     */
    
    public Server(int port) {
        try {
            socket = new DatagramSocket(port);
            socket.setSoTimeout(TIME_LIMIT);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public String listen() {
        String req = null;
        while (req==null) {
            req = rcvd();
        }
        return req;
    }
    public void answer(String str) {
        String res = null;
        try {
            while (!isACK(res)) {
                send((byte)(1+rand.nextInt(255)),str);
                res = rcvd();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}


public class Peer extends Thread {
    /**    A node on a Peer-to-Peer Network
     *    For consecutive peers with indexes i and j:
     *         peer[i] is resp for key k iff i <= k < j;
     *     finger[k] is the port # of the on-line peer with
     *         the minimum index not less than index+2^{k}.
     */
    
    private static final int LEN = 16;
    private static final int NUM = (1<<LEN);
    private static DecimalFormat df;
    private static int port, index;
    private static AtomicInteger[] finger;
    private Server sv;
    
    private Peer() {
        sv = new Server(port);
        start();
    }
    public void run() {
        while (true) {
            String msg = sv.listen();
            if (msg.charAt(0)=='?') {
                msg = msg.substring(1,6);
                int next = search(Integer.parseInt(msg));
                if (next==port) {
                    sv.answer("!");
                } else {
                    sv.answer("@"+next);
                }
            } else if (msg.charAt(0)=='#') {
                sv.answer("ACK");
                break;
            }
        }
        System.out.println("#");
    }
    private static int calDelt(int val) {
        int delt = hashIdx(val)-index;
        return delt+((delt<=0)? NUM:0);
    }
    private static boolean within(int a,int x,int b) {
        a -= index;
        a += (a<0)? NUM:0;
        x -= index;
        x += (x<0)? NUM:0;
        b -= index;
        b += (b<0)? NUM:0;
        return a<=x && x<b;
    }
    private static boolean insfinger(int val) {
        /* Insert a port number to the finger table */
        boolean flag = false;
        if (val==port) {
            return flag;
        }
        int delt = calDelt(val);
        for (int i=0;(1<<i)<=delt;i++) {
            if (delt<calDelt(finger[i].get())) {
                finger[i].set(val);
                flag = true;
            }
        }
        return flag;
    }
    private static int search(int key) {
        /* Chord Algorithm Implementation */
        if (within(index,key,hashIdx(finger[0].get()))) {
            return port;
        }
        for (int i=0;i<LEN-1;i++) {
            int a = hashIdx(finger[i].get());
            int b = hashIdx(finger[i+1].get());
            if (b==index||within(a,key,b)) {
                return finger[i].get();
            }
        }
        return finger[LEN-1].get();
    }
    private static int locate(int key) {
        /* Locate the peer who is resp for a given key */
        int next = search(key);
        if (next==port) {
            return port;
        }
        String s = null;
        while (true) {
            s = Client.inquire(next,"?"+df.format(key));
            if (s.charAt(0)=='@') {
                s = s.substring(1,5);
                next = Integer.parseInt(s);
            } else {
                break;
            }
        }
        return next;
    }
    private static int hashIdx(int key) {
        /* Hash function using multiplication */
        double tmp = 0.61803*key;
        tmp -= Math.floor(tmp);
        return (int)Math.floor(NUM*tmp);
    }
    public static void main(String[] args) {
        df = new DecimalFormat("00000");
        try {
            port = Integer.parseInt(args[0]);
            index = hashIdx(port);
            finger = new AtomicInteger[LEN];
            for (int i=0;i<LEN;i++) {
                finger[i] = new AtomicInteger(port);
            }
            // acquaint all the on-line peers
            Scanner in = new Scanner(System.in);
            int num = in.nextInt();
            for (int i=0;i<num;i++) {
                insfinger(in.nextInt());
            }
            Peer server = new Peer();
            // for all peers to set up the finger table
            TimeUnit.MILLISECONDS.sleep(num*100);
            Random rand = new Random();
            for (int i=0;i<3;i++) {
                // Each peer has three searching tasks.
                int key = rand.nextInt(NUM);
                System.out.println("Port "+port+": port "+locate(key)
                        +" is resp for key "+df.format(key));
            }
            // for all peers to finish the searching task
            TimeUnit.MILLISECONDS.sleep(num*100);
            Client.inform(port,"#");
            server.join();
        } catch (Exception e) {
            System.out.println("Main Error: "+e);
        }
    }
}

References:

　　1. Kurose, James F., Keith W. Ross. Computer Networking: a top-down approach[M]. 北京：高等教育出版社, 2009-08

　　2. Tanenbaum, Andrew S., David J. Wetherall. Computer Networks 5th edition[M]. 北京：清华大学出版社, 2011