Message encoding of announcement and response messages.

More changes coming.
Rename of organization to Peernet s.r.o.
InformationRequest proper termination signal.

dht/DHT Lite.go

@@ -40,15 +40,14 @@ type DHT struct {
 
 	// SendRequest sends an information request to the remote node. I.e. requesting information.
 	// The returned results channel will be closed when no more results are to be expected.
-	SendRequest func(request *InformationRequest, nodes []*Node) (results chan *message2)
+	SendRequest func(request *InformationRequest, nodes []*Node)
 
 	// The maximum time to wait for a response to any message
 	TMsgTimeout time.Duration
 }
 
 // NewDHT initializes a new DHT node with default values.
-// Store must be set by the caller.
-func NewDHT(store Store, self Node, bits, bucketSize int) *DHT {
+func NewDHT(self *Node, bits, bucketSize int) *DHT {
 	return &DHT{
 		ht:          newHashTable(self, bits, bucketSize),
 		alpha:       3,
@@ -112,10 +111,10 @@ func (dht *DHT) FindNode(key []byte) (value []byte, found bool, err error) {
 //     IterateStore - Used to store new information in the network.
 //     IterateFindNode - Used to bootstrap the network.
 //     IterateFindValue - Used to find a value among the network given a key.
-func (dht *DHT) iterate(t int, target []byte, data []byte) (value []byte, closest []*Node, err error) {
+func (dht *DHT) iterate(action int, target []byte, data []byte) (value []byte, closest []*Node, err error) {
 	if len(target) != dht.ht.bBits {
 		return nil, nil, errors.New("invalid key")
-	} else if t < IterateStore || t > IterateFindValue {
+	} else if action < IterateStore || action > IterateFindValue {
 		return nil, nil, errors.New("unknown iterate type")
 	}
 
@@ -133,15 +132,16 @@ func (dht *DHT) iterate(t int, target []byte, data []byte) (value []byte, closes
 	closestNode := sl.Nodes[0]
 
 	for {
-		resultsChan := dht.SendRequest(&InformationRequest{Action: t, Key: target}, sl.GetUncontacted(dht.alpha, !queryRest))
-		results := infoCollectResults(resultsChan, dht.TMsgTimeout)
+		info := NewInformationRequest(action, target)
+		dht.SendRequest(info, sl.GetUncontacted(dht.alpha, !queryRest))
+		results := info.CollectResults(dht.TMsgTimeout)
 
 		for _, result := range results {
 			if result.Error != nil {
 				sl.RemoveNode(result.SenderID)
 				continue
 			}
-			switch t {
+			switch action {
 			case IterateFindNode:
 				sl.AppendUniqueNodes(result.Closest...)
 				// TODO: Accept contact info?
@@ -161,7 +161,7 @@ func (dht *DHT) iterate(t int, target []byte, data []byte) (value []byte, closes
 		// If closestNode is unchanged then we are done
 		if bytes.Compare(sl.Nodes[0].ID, closestNode.ID) == 0 || queryRest {
 			// We are done
-			switch t {
+			switch action {
 			case IterateFindNode:
 				if !queryRest {
 					queryRest = true

dht/Hash Table.go

@@ -19,7 +19,7 @@ import (
 // hashTable represents the hashtable state
 type hashTable struct {
 	// The ID of the local node
-	Self Node
+	Self *Node
 
 	// the size in bits of the keys used to identify nodes and store and
 	// retrieve data; in basic Kademlia this is 160, the length of a SHA1
@@ -38,12 +38,12 @@ type hashTable struct {
 	mutex *sync.Mutex
 }
 
-func newHashTable(n Node, bits, bucketSize int) *hashTable {
+func newHashTable(self *Node, bits, bucketSize int) *hashTable {
 	ht := &hashTable{
 		bBits: bits,
 		bSize: bucketSize,
 		mutex: &sync.Mutex{},
-		Self:  n,
+		Self:  self,
 	}
 
 	ht.RoutingTable = make([][]*Node, ht.bBits)

dht/Information Request.go

@@ -8,36 +8,62 @@ Information requests are asynchronous queries sent to nodes.
 
 package dht
 
-import "time"
+import (
+	"sync"
+	"time"
+)
 
 // InformationRequest is an asynchronous request sent. It tracks any asynchronous replies and handles timeouts.
 type InformationRequest struct {
-	Action int    // IterateFindNode or IterateFindValue
-	Key    []byte // Target key
-
-	// TODO: Include results channel? Timeout settings? Cancelation?
+	Action          int               // IterateFindNode or IterateFindValue
+	Key             []byte            // Target key
+	ResultChan      chan *NodeMessage // Result channel
+	IsTerminated    bool              // If true, it was signaled for termination
+	TerminateSignal chan interface{}  // gets closed on termination signal, can be used in select via "case _ = <- network.terminateSignal:"
+	sync.Mutex                        // for sychronized closing
 }
 
-type message2 struct {
-	SenderID []byte // Sender of this message
-	Data     []byte
-	Closest  []*Node
-	Error    error
+// NewInformationRequest creates a new information request
+func NewInformationRequest(Action int, Key []byte) (ir *InformationRequest) {
+	return &InformationRequest{
+		ResultChan: make(chan *NodeMessage),
+		Action:     Action,
+		Key:        Key,
+	}
 }
 
-// infoCollectResults collects all information request responses within the given timeout.
-func infoCollectResults(resultChan chan *message2, timeout time.Duration) (results []*message2) {
+// CollectResults collects all information request responses within the given timeout.
+func (ir *InformationRequest) CollectResults(timeout time.Duration) (results []*NodeMessage) {
 	for {
 		select {
-		case result, ok := <-resultChan:
-			if !ok {
-				return
-			}
+		case result := <-ir.ResultChan:
 			results = append(results, result)
+
 		case <-time.After(timeout):
-			// send cancelation signal ?
-			//close(resultChan)
+			ir.Terminate()
+			return
+
+		case <-ir.TerminateSignal:
 			return
 		}
 	}
 }
+
+// Terminate sends the termination signal to all workers. It is safe to call Terminate multiple times.
+func (ir *InformationRequest) Terminate() {
+	ir.Lock()
+	defer ir.Unlock()
+
+	if ir.IsTerminated {
+		return
+	}
+
+	// set the termination signal
+	ir.IsTerminated = true
+	close(ir.TerminateSignal) // safety guaranteed via lock
+
+	close(ir.ResultChan)
+}
+
+// TODO: Incoming information request responses should be handled in batches, i.e. every X ms (for example 100ms) without waiting for all results to finish.
+// This could seriously speed up discovery.

dht/Node.go

@@ -123,3 +123,11 @@ func (n *shortList) GetUncontacted(count int, useCount bool) (Nodes []*Node) {
 
 	return Nodes
 }
+
+// NodeMessage is a message sent by a node
+type NodeMessage struct {
+	SenderID []byte // Sender of this message
+	Data     []byte
+	Closest  []*Node
+	Error    error
+}