core/Kademlia.go

/*
File Username:  Kademlia.go
Copyright:  2021 Peernet s.r.o.
Author:     Peter Kleissner
*/

package core

import (
	"bytes"
	"time"

	"github.com/PeernetOfficial/core/dht"
	"github.com/PeernetOfficial/core/protocol"
)

const alpha = 5       // Count of nodes to be contacted in parallel for finding a key
const bucketSize = 20 // Count of nodes per bucket

func (backend *Backend) initKademlia() {
	backend.nodesDHT = dht.NewDHT(&dht.Node{ID: backend.nodeID}, 256, bucketSize, alpha)

	// ShouldEvict determines whether node 1 shall be evicted in favor of node 2
	backend.nodesDHT.ShouldEvict = func(node1, node2 *dht.Node) bool {
		rttOld := node1.Info.(*PeerInfo).GetRTT()
		rttNew := node2.Info.(*PeerInfo).GetRTT()

		// evict the old node if the new one has a faster ping time
		if rttOld == 0 { // old one has no recent RTT (happens if all connections are inactive)?
			return true
		} else if rttNew > 0 {
			// If new RTT is smaller, evict old one.
			return rttNew < rttOld
		}

		// If here, none has a RTT. Keep the closer (by distance) one.
		return backend.nodesDHT.IsNodeCloser(node1.ID, node2.ID)
	}

	// SendRequestStore sends a store message to the remote node. I.e. asking it to store the given key-value
	backend.nodesDHT.SendRequestStore = func(node *dht.Node, key []byte, dataSize uint64) {
		node.Info.(*PeerInfo).sendAnnouncementStore(key, dataSize)
	}

	// SendRequestFindNode sends an information request to find a particular node. nodes are the nodes to send the request to.
	backend.nodesDHT.SendRequestFindNode = func(request *dht.InformationRequest) {
		for _, node := range request.Nodes {
			node.Info.(*PeerInfo).sendAnnouncementFindNode(request)
		}
	}

	// SendRequestFindValue sends an information request to find data. nodes are the nodes to send the request to.
	backend.nodesDHT.SendRequestFindValue = func(request *dht.InformationRequest) {
		for _, node := range request.Nodes {
			node.Info.(*PeerInfo).sendAnnouncementFindValue(request)
		}
	}

	backend.nodesDHT.FilterSearchStatus = backend.Filters.DHTSearchStatus
}

// autoBucketRefresh refreshes buckets every 5 minutes to meet the alpha nodes per bucket target. Force full refresh every hour.
func (backend *Backend) autoBucketRefresh() {
	for minute := 5; ; minute += 5 {
		time.Sleep(time.Minute * 5)

		target := alpha
		if minute%60 == 0 {
			target = 0
		}

		backend.nodesDHT.RefreshBuckets(target)
	}
}

// bootstrapKademlia bootstraps the Kademlia bucket list
func (backend *Backend) bootstrapKademlia() {
	monitor := make(chan *PeerInfo)
	backend.registerPeerMonitor(monitor)

	// Wait until there are at least 2 peers connected.
	for {
		<-monitor
		if backend.nodesDHT.NumNodes() >= 2 {
			break
		}
	}

	backend.unregisterPeerMonitor(monitor)

	// Refresh every 10 seconds 3 times
	for n := 0; n < 3; n++ {
		backend.nodesDHT.RefreshBuckets(alpha)

		time.Sleep(time.Second)
	}
}

// Future sendAnnouncementX: If it detects that announcements are sent out to the same peer within 50ms it should activate a wait-and-group scheme.

func (peer *PeerInfo) sendAnnouncementFindNode(request *dht.InformationRequest) {
	// If the key is self, send it as FIND_SELF
	if bytes.Equal(request.Key, peer.Backend.nodeID) {
		peer.sendAnnouncement(false, true, nil, nil, nil, request)
	} else {
		peer.sendAnnouncement(false, false, []protocol.KeyHash{{Hash: request.Key}}, nil, nil, request)
	}
}

func (peer *PeerInfo) sendAnnouncementFindValue(request *dht.InformationRequest) {

	findSelf := false
	var findPeer []protocol.KeyHash
	var findValue []protocol.KeyHash

	findValue = append(findValue, protocol.KeyHash{Hash: request.Key})

	peer.sendAnnouncement(false, findSelf, findPeer, findValue, nil, request)
}

func (peer *PeerInfo) sendAnnouncementStore(fileHash []byte, fileSize uint64) {
	peer.sendAnnouncement(false, false, nil, nil, []protocol.InfoStore{{ID: protocol.KeyHash{Hash: fileHash}, Size: fileSize, Type: 0}}, nil)
}

// ---- CORE DATA FUNCTIONS ----

// Data2Hash returns the hash for the data
func Data2Hash(data []byte) (hash []byte) {
	return protocol.HashData(data)
}

// GetData returns the requested data. It checks first the local store and then tries via DHT.
func (backend *Backend) GetData(hash []byte) (data []byte, senderNodeID []byte, found bool) {
	if data, found = backend.GetDataLocal(hash); found {
		return data, backend.nodeID, found
	}

	return backend.GetDataDHT(hash)
}

// GetDataLocal returns data from the local warehouse.
func (backend *Backend) GetDataLocal(hash []byte) (data []byte, found bool) {
	return backend.dhtStore.Get(hash)
}

// GetDataDHT requests data via DHT
func (backend *Backend) GetDataDHT(hash []byte) (data []byte, senderNodeID []byte, found bool) {
	data, senderNodeID, found, _ = backend.nodesDHT.Get(hash)
	return data, senderNodeID, found
}

// StoreDataLocal stores data into the local warehouse.
func (backend *Backend) StoreDataLocal(data []byte) error {
	key := protocol.HashData(data)
	return backend.dhtStore.Set(key, data)
}

// StoreDataDHT stores data locally and informs closestCount peers in the DHT about it.
// Remote peers may choose to keep a record (in case another peers asks) or mirror the full data.
func (backend *Backend) StoreDataDHT(data []byte, closestCount int) error {
	key := protocol.HashData(data)
	if err := backend.dhtStore.Set(key, data); err != nil {
		return err
	}
	return backend.nodesDHT.Store(key, uint64(len(data)), closestCount)
}

// ---- NODE FUNCTIONS ----

// IsNodeContact checks if the node is a contact in the local DHT routing table
func (backend *Backend) IsNodeContact(nodeID []byte) (node *dht.Node, peer *PeerInfo) {
	node = backend.nodesDHT.IsNodeContact(nodeID)
	if node == nil {
		return nil, nil
	}

	return node, node.Info.(*PeerInfo)
}

// FindNode finds a node via the DHT
func (backend *Backend) FindNode(nodeID []byte, Timeout time.Duration) (node *dht.Node, peer *PeerInfo, err error) {
	// first check if in mirrored node list
	if peer = backend.NodelistLookup(nodeID); peer != nil {
		return nil, peer, nil
	}

	// Search the node via DHT.
	node, err = backend.nodesDHT.FindNode(nodeID)
	if node == nil {
		return nil, nil, err
	}

	return node, node.Info.(*PeerInfo), err
}

// ---- Asynchronous Search ----

// AsyncSearch creates an async search for the given key in the DHT.
// Timeout is the total time the search may take, covering all information requests. TimeoutIR is the time an information request may take.
// Alpha is the number of concurrent requests that will be performed.
func (backend *Backend) AsyncSearch(Action int, Key []byte, Timeout, TimeoutIR time.Duration, Alpha int) (client *dht.SearchClient) {
	return backend.nodesDHT.NewSearch(Action, Key, Timeout, TimeoutIR, Alpha)
}