test/go-chord/chord_test/chord_test.go

package chord_test

import (
	"fmt"
	"runtime"
	"testing"
	"time"
	"trial/go-chord"
)

type MultiLocalTrans struct {
	remote chord.Transport
	hosts  map[string]*chord.LocalTransport
}

func InitMLTransport() *MultiLocalTrans {
	hosts := make(map[string]*chord.LocalTransport)
	remote := &chord.BlackholeTransport{}
	ml := &MultiLocalTrans{hosts: hosts}
	ml.remote = remote
	return ml
}

func (ml *MultiLocalTrans) ListVnodes(host string) ([]*chord.Vnode, error) {
	if local, ok := ml.hosts[host]; ok {
		return local.ListVnodes(host)
	}
	return ml.remote.ListVnodes(host)
}

// Ping a Vnode, check for liveness
func (ml *MultiLocalTrans) Ping(v *chord.Vnode) (bool, error) {
	if local, ok := ml.hosts[v.Host]; ok {
		return local.Ping(v)
	}
	return ml.remote.Ping(v)
}

// Request a nodes predecessor
func (ml *MultiLocalTrans) GetPredecessor(v *chord.Vnode) (*chord.Vnode, error) {
	if local, ok := ml.hosts[v.Host]; ok {
		return local.GetPredecessor(v)
	}
	return ml.remote.GetPredecessor(v)
}

// Notify our successor of ourselves
func (ml *MultiLocalTrans) Notify(target, self *chord.Vnode) ([]*chord.Vnode, error) {
	if local, ok := ml.hosts[target.Host]; ok {
		return local.Notify(target, self)
	}
	return ml.remote.Notify(target, self)
}

// Find a successor
func (ml *MultiLocalTrans) FindSuccessors(v *chord.Vnode, n int, k []byte) ([]*chord.Vnode, error) {
	if local, ok := ml.hosts[v.Host]; ok {
		return local.FindSuccessors(v, n, k)
	}
	return ml.remote.FindSuccessors(v, n, k)
}

// Clears a predecessor if it matches a given vnode. Used to leave.
func (ml *MultiLocalTrans) ClearPredecessor(target, self *chord.Vnode) error {
	if local, ok := ml.hosts[target.Host]; ok {
		return local.ClearPredecessor(target, self)
	}
	return ml.remote.ClearPredecessor(target, self)
}

// Instructs a node to skip a given successor. Used to leave.
func (ml *MultiLocalTrans) SkipSuccessor(target, self *chord.Vnode) error {
	if local, ok := ml.hosts[target.Host]; ok {
		return local.SkipSuccessor(target, self)
	}
	return ml.remote.SkipSuccessor(target, self)
}

func (ml *MultiLocalTrans) Register(v *chord.Vnode, o chord.VnodeRPC) {
	local, ok := ml.hosts[v.Host]
	if !ok {
		local = chord.InitLocalTransport(nil).(*chord.LocalTransport)
		ml.hosts[v.Host] = local
	}
	local.Register(v, o)
}

func (ml *MultiLocalTrans) Deregister(host string) {
	delete(ml.hosts, host)
}

func TestDefaultConfig(t *testing.T) {
	conf := chord.DefaultConfig("test")
	if conf.Hostname != "test" {
		t.Fatalf("bad hostname")
	}
	if conf.NumVnodes != 8 {
		t.Fatalf("bad num Vnodes")
	}
	if conf.NumSuccessors != 8 {
		t.Fatalf("bad num succ")
	}
	if conf.HashFunc == nil {
		t.Fatalf("bad hash")
	}
	if conf.HashBits != 160 {
		t.Fatalf("bad hash bits")
	}
	if conf.StabilizeMin != time.Duration(15*time.Second) {
		t.Fatalf("bad min stable")
	}
	if conf.StabilizeMax != time.Duration(45*time.Second) {
		t.Fatalf("bad max stable")
	}
	if conf.Delegate != nil {
		t.Fatalf("bad delegate")
	}
}

func fastConf() *chord.Config {
	conf := chord.DefaultConfig("test")
	conf.StabilizeMin = time.Duration(15 * time.Millisecond)
	conf.StabilizeMax = time.Duration(45 * time.Millisecond)
	return conf
}

func TestCreateShutdown(t *testing.T) {
	// Start the timer thread
	time.After(15)
	conf := fastConf()
	numGo := runtime.NumGoroutine()
	r, err := chord.Create(conf, nil)
	if err != nil {
		t.Fatalf("unexpected err. %s", err)
	}
	r.Shutdown()
	after := runtime.NumGoroutine()
	if after != numGo {
		t.Fatalf("unexpected routines! A:%d B:%d", after, numGo)
	}
}

func TestJoin(t *testing.T) {
	// Create a multi transport
	ml := InitMLTransport()

	// Create the initial ring
	conf := fastConf()
	r, err := chord.Create(conf, ml)
	if err != nil {
		t.Fatalf("unexpected err. %s", err)
	}

	// Create a second ring
	conf2 := fastConf()
	conf2.Hostname = "test2"
	r2, err := chord.Join(conf2, ml, "test")
	if err != nil {
		t.Fatalf("failed to join local node! Got %s", err)
	}

	// Shutdown
	r.Shutdown()
	r2.Shutdown()
}

func TestJoinDeadHost(t *testing.T) {
	// Create a multi transport
	ml := InitMLTransport()

	// Create the initial ring
	conf := fastConf()
	_, err := chord.Join(conf, ml, "noop")
	if err == nil {
		t.Fatalf("expected err!")
	}
}

func TestLeave(t *testing.T) {
	// Create a multi transport
	ml := InitMLTransport()

	// Create the initial ring
	conf := fastConf()
	r, err := chord.Create(conf, ml)
	if err != nil {
		t.Fatalf("unexpected err. %s", err)
	}

	// Create a second ring
	conf2 := fastConf()
	conf2.Hostname = "test2"
	r2, err := chord.Join(conf2, ml, "test")
	if err != nil {
		t.Fatalf("failed to join local node! Got %s", err)
	}

	// Wait for some stabilization
	<-time.After(100 * time.Millisecond)

	// Node 1 should leave
	r.Leave()
	ml.Deregister("test")

	// Wait for stabilization
	<-time.After(100 * time.Millisecond)

	// Verify r2 ring is still in tact
	num := len(r2.Vnodes)
	for idx, vn := range r2.Vnodes {
		if vn.Successors[0] != &r2.Vnodes[(idx+1)%num].Vnode {
			t.Fatalf("bad successor! Got:%s:%s", vn.Successors[0].Host,
				vn.Successors[0])
		}
	}
}

func TestLookupBadN(t *testing.T) {
	// Create a multi transport
	ml := InitMLTransport()

	// Create the initial ring
	conf := fastConf()
	r, err := chord.Create(conf, ml)
	if err != nil {
		t.Fatalf("unexpected err. %s", err)
	}

	vns, err := r.Lookup(5, []byte("test"))
	fmt.Println("vns::", fmt.Sprintf("%s", vns))
	// if err == nil {
	// 	t.Fatalf("expected err!")
	// }
	for idx := range vns {
		fmt.Println(vns[idx].String(), vns[idx].Host, fmt.Sprintf("%x", vns[idx].Id))
	}
}

func TestLookup(t *testing.T) {
	// Create a multi transport
	ml := InitMLTransport()

	// Create the initial ring
	conf := fastConf()
	conf.NumVnodes = 8
	conf.NumSuccessors = 8
	r, err := chord.Create(conf, ml)
	if err != nil {
		t.Fatalf("unexpected err. %s", err)
	}
	fmt.Println("conf::", fmt.Sprintf("%#v", conf))
	fmt.Println("ml::", fmt.Sprintf("%#v", ml))
	fmt.Println("r::", fmt.Sprintf("%#v", r))

	// Create a second ring
	conf2 := fastConf()
	conf2.NumVnodes = 8
	conf2.NumSuccessors = 8
	conf2.Hostname = "test2"
	r2, err := chord.Join(conf2, ml, "test")
	if err != nil {
		t.Fatalf("failed to join local node! Got %s", err)
	}
	fmt.Println("conf2::", fmt.Sprintf("%#v", conf2))
	fmt.Println("ml::", fmt.Sprintf("%#v", ml))
	fmt.Println("r2::", fmt.Sprintf("%#v", r2))
	fmt.Println("r::", fmt.Sprintf("%#v", r))

	for n := 3; n <= 3; n++ {
		// Create a second ring
		confx := fastConf()
		confx.NumVnodes = 8
		confx.NumSuccessors = 8
		confx.Hostname = fmt.Sprintf("%s%d", "test", n)
		_, err = chord.Join(confx, ml, "test")
		if err != nil {
			t.Fatalf("failed to join local node! Got %s", err)
		}
	}

	// Wait for some stabilization
	<-time.After(100 * time.Millisecond)

	// Try key lookup
	keys := [][]byte{[]byte("test"), []byte("foo"), []byte("bar"), []byte("trial")}
	for _, k := range keys {
		fmt.Println("lookup", fmt.Sprintf("%s", k))
		vn1, err := r.Lookup(2, k)
		if err != nil {
			t.Fatalf("unexpected err %s", err)
		}
		vn2, err := r2.Lookup(2, k)
		if err != nil {
			t.Fatalf("unexpected err %s", err)
		}
		if len(vn1) != len(vn2) {
			t.Fatalf("result len differs!")
		}
		fmt.Println("vn1", fmt.Sprintf("%#v", vn1))
		fmt.Println("vn2", fmt.Sprintf("%#v", vn2))
		for idx := range vn1 {
			fmt.Println(vn1[idx].String(), vn1[idx].Host, fmt.Sprintf("%x", vn1[idx].Id))
			if vn1[idx].String() != vn2[idx].String() {
				t.Fatalf("results differ!")
			}
		}
	}
}

func TestLookupN(t *testing.T) {
	// Create a multi transport
	ml := InitMLTransport()

	// Create the initial ring
	conf := fastConf()
	conf.NumVnodes = 16
	conf.NumSuccessors = 8
	r, err := chord.Create(conf, ml)
	if err != nil {
		t.Fatalf("unexpected err. %s", err)
	}
	fmt.Println("conf::", fmt.Sprintf("%#v", conf))
	fmt.Println("ml::", fmt.Sprintf("%#v", ml))
	fmt.Println("r::", fmt.Sprintf("%#v", r))

	for n := 2; n <= 4; n++ {
		// Create a second ring
		confx := fastConf()
		confx.NumVnodes = 16
		confx.NumSuccessors = 8
		confx.Hostname = fmt.Sprintf("%s%d", "test", n)
		_, err = chord.Join(confx, ml, "test")
		if err != nil {
			t.Fatalf("failed to join local node! Got %s", err)
		}
	}

	// Wait for some stabilization
	<-time.After(100 * time.Millisecond)

	kc := map[string]int{}
	for i := 0; i < 10000; i++ {
		k := []byte(fmt.Sprint("key", i))
		fmt.Println("lookup", fmt.Sprintf("%s", k))
		vns, err := r.Lookup(3, k)
		if err != nil {
			t.Fatalf("unexpected err %s", err)
		}
		for idx := range vns {
			kc[vns[idx].Host]++
			fmt.Println(vns[idx].String(), vns[idx].Host, fmt.Sprintf("%x", vns[idx].Id))
		}
	}
	fmt.Println("count:", kc)
}