added a simple, as of now uncomplete, binary

routing table

Author: distributed

lib/Makefile

@@ -6,5 +6,6 @@ GOFILES=callmanager.go\
 	utility.go\
 	webinterface.go\
 	sleepqueue.go
+	routing.go
 
 include $(GOROOT)/src/Make.pkg

lib/callmanager.go

@@ -135,10 +135,11 @@ type CallManager struct {
 	regchan     chan *RunningRPC
 	inchan      chan *RPC
 	timeout  int64
+	rt RoutingTable
 }
 
 
-func NewCallManager(transceiver Transceiver) *CallManager {
+func NewCallManager(transceiver Transceiver, rt RoutingTable) *CallManager {
 	cm := new(CallManager)
 
 	cm.rpcmap = make(map[string](*rpcentry), 8)
@@ -149,6 +150,7 @@ func NewCallManager(transceiver Transceiver) *CallManager {
 	cm.regchan = make(chan *RunningRPC)
 	cm.inchan = make(chan *RPC)
 	cm.timeout = TIMEOUT
+	cm.rt = rt
 
 	return cm
 }
@@ -206,6 +208,9 @@ func (cm *CallManager) constructAnswer(req *RPC, retcall uint8, retis []interfac
 
 // TODO: errors...
 func (cm *CallManager) DispatchRPC(rpc *RPC) {
+	if cm.rt != nil {
+		cm.rt.SeeHost(rpc.From)
+	}
 	switch rpc.Header.Call {
 	case 0x01:
 		retcall, retis, err := cm.DispatchRequest(rpc)

lib/routing.go

@@ -0,0 +1,232 @@
+package malus
+
+
+import (
+	"os"
+	"log"
+	"bytes"
+	"fmt"
+	)
+
+type RoutingTable interface {
+	SeeHost(h *Host)
+}
+
+
+type BRoutingTable struct {
+	id string
+	buckets [](*Bucket)
+	maxbucket int
+
+	seehost chan *Host
+	quitchan chan bool
+
+	Log bool
+	logger *log.Logger
+}
+
+
+type RTHost struct {
+	host *Host
+	distance Distance
+}
+
+type Bucket struct {
+	hosts [](*RTHost)
+}
+
+
+func NewBRoutingTable(id string) (rt *BRoutingTable) {
+	rt = new(BRoutingTable)
+
+	rt.id = id
+
+	rt.buckets = make([](*Bucket), HASHLEN * 8)
+	firstbucket := new(Bucket)
+	firstbucket.hosts = make([](*RTHost), K + 1)[0:0]
+
+	rt.buckets[0] = firstbucket
+	rt.maxbucket = 0
+
+	rt.seehost = make(chan *Host)
+	rt.quitchan = make(chan bool)
+
+	rt.Log = true
+	rt.logger = log.New(os.Stdout, nil, "BRoutingTable: ", 0)
+
+	go rt.main()
+
+	return
+}
+
+
+func (rt *BRoutingTable) main() {
+	for {
+		select {
+		case h := <- rt.seehost:
+			rt.seeHost(h)
+		case <- rt.quitchan:
+			return
+		}
+	}
+}
+
+
+func (rt *BRoutingTable) SeeHost(h *Host) {
+	rt.seehost <- h
+}
+
+
+func (rt *BRoutingTable) seeHost(h *Host) {
+	if rt.Log {
+		rt.logger.Logf("see host: %v dist %v\n", h, XOR(h.Id, rt.id)[0:5])
+	}
+	bucketno, pos, maxbucketno := rt.findHost(h)
+	/*rt.logger.Logf("is in %d/%d maxbucketno %d\n", bucketno, pos, maxbucketno) 
+	rt.logger.Logf("dist %v to %v\n", XOR(h.Id, rt.id), h)
+	rt.logger.Logf("own id is %x\n", rt.id)
+	rt.logger.Logf("h.Id is %x", h.Id)*/
+
+
+	bucket := rt.buckets[bucketno]
+
+	// we do not already have the entry
+	if pos < 0 {
+		if rt.Log {
+			rt.logger.Logf("we don't have an entry yet\n")
+		}
+		rthost := &RTHost{h, XOR(h.Id, rt.id)}
+		hl := len(bucket.hosts)
+		if hl < K {
+			if rt.Log {
+				rt.logger.Logf("bucket %d not full yet -> inserting\n", bucketno)
+			}
+			bucket.hosts = bucket.hosts[0:hl+1]
+			bucket.hosts[hl] = rthost
+		} else {
+			if maxbucketno == bucketno {
+				if rt.Log {
+					rt.logger.Logf("bucket %d full -> dropping\n", bucketno)
+				}
+			} else {
+				rt.newBucket()
+				bucket.hosts = bucket.hosts[0:hl+1]
+				bucket.hosts[hl] = rthost
+				rt.balanceleftright(uint(bucketno))
+			}
+		}
+	} else {
+		if rt.Log {
+			rt.logger.Logf("we already have an entry\n")
+		}
+		host := bucket.hosts[pos]
+		copy(bucket.hosts[pos:], bucket.hosts[pos+1:])
+		bucket.hosts[len(bucket.hosts)-1] = host
+	}
+}
+
+
+func (rt *BRoutingTable) newBucket() *Bucket {
+	if rt.maxbucket == (HASHLEN * 8 - 1) {
+		return nil
+	}
+
+	b := new(Bucket)
+	b.hosts = make([](*RTHost), K + 1)[0:0]
+
+	rt.maxbucket++
+	rt.buckets[rt.maxbucket] = b
+	
+	return b
+}
+
+
+func (rt *BRoutingTable) findHost(h *Host) (bucketno, pos, maxbucketno int) {
+	bucketno = -1
+	pos = -1
+
+	dist := XOR(h.Id, rt.id)
+	maxbucketno = int(BucketNo(dist))
+
+	maxbucket := rt.maxbucket
+	if maxbucket < 0 {
+		return
+	}
+
+	bucketno = maxbucketno
+
+	if bucketno > maxbucket {
+		bucketno = maxbucket
+	}
+
+	hosts := rt.buckets[bucketno].hosts
+	for i, rh := range hosts {
+		pos = i
+		if rh.host.Id == h.Id {
+			return
+		}
+	}
+
+	pos = -1
+	return
+}
+
+
+func (rt *BRoutingTable) balanceleftright(lefti uint) {
+	righti := lefti + 1
+
+	left := rt.buckets[lefti]
+	right := rt.buckets[righti]
+
+	newleft := make([](*RTHost), K + 1)
+	newright := make([](*RTHost), K + 1)
+
+	nleft := 0
+	nright := 0
+
+	for _, rth := range left.hosts {
+		if BucketNo(rth.distance) == lefti {
+			newleft[nleft] = rth
+			nleft++
+		} else {
+			newright[nright] = rth
+			nright++
+		}
+	}
+
+	for _, rth := range right.hosts {
+		if BucketNo(rth.distance) == lefti {
+			newleft[nleft] = rth
+			nleft++
+		} else {
+			newright[nright] = rth
+			nright++
+		}
+	}
+
+	rt.logger.Logf("rebalanced to %d/%d\n", nleft, nright)
+	
+	newleft = newleft[0:nleft]
+	newright = newright[0:nright]
+
+	left.hosts = newleft
+	right.hosts = newright
+	
+	fmt.Printf("%v", rt)
+}
+
+
+func (rt *BRoutingTable) String() string {
+	buf := bytes.NewBuffer(nil)
+
+	buf.WriteString("BRoutingTable ===>\n")
+	for b := 0; b <= rt.maxbucket; b++ {
+		buf.WriteString(fmt.Sprintf("bucket %d\n", b))
+		for _, rth := range rt.buckets[b].hosts {
+			buf.WriteString(fmt.Sprintf("\t%x | %v @ %v\n", rth.host.Id, XOR(rth.host.Id, rt.id)[0:5], rth.host.Addr))
+		}
+	}
+	
+	buf.WriteString("<===\n")
+	return buf.String()
+}

main/main.go

@@ -24,8 +24,12 @@ func main() {
 		panic("could not make transceiver")
 	}
 
-	cm := malus.NewCallManager(tr)
 	id := malus.SHA1String(strconv.Itoa(laddr.Port))
+
+
+	rt := malus.NewBRoutingTable(id)
+
+	cm := malus.NewCallManager(tr, rt)
 	cm.Id = id
 	cm.AddRPC("ping", malus.Ping)
 	cm.AddRPC("store", malus.Store)