1 files changed, 200 insertions, 105 deletions

diff --git a/src/gnunet/service/dht/routingtable.go b/src/gnunet/service/dht/routingtable.go
index 2933e6a..fe9f956 100644
--- a/src/gnunet/service/dht/routingtable.go
+++ b/src/gnunet/service/dht/routingtable.go
@@ -21,10 +21,11 @@ package dht
 import (
         "bytes"
         "context"
-        "crypto/sha512"
         "encoding/hex"
+        "gnunet/config"
+        "gnunet/crypto"
+        "gnunet/service/dht/blocks"
         "gnunet/util"
-        "math/rand"
         "sync"
         "time"
 
@@ -32,61 +33,63 @@ import (
         "github.com/bfix/gospel/math"
 )
 
-var (
-        // routing table hash function: defines number of
-        // buckets and size of peer addresses
-        rtHash = sha512.New
-)
-
-// Routing table contants (adjust with changing hash function)
+// Routing table constants
 const (
-        numBuckets = 512 // number of bits of hash function result
-        numK       = 20  // number of entries per k-bucket
-        sizeAddr   = 64  // size of peer address in bytes
+        numK = 20 // number of entries per k-bucket
 )
 
 //======================================================================
+// Peer address
 //======================================================================
 
 // PeerAddress is the identifier for a peer in the DHT network.
 // It is the SHA-512 hash of the PeerID (public Ed25519 key).
 type PeerAddress struct {
-        addr      [sizeAddr]byte    // hash value as bytes
-        connected bool              // is peer connected?
-        lastSeen  util.AbsoluteTime // time the peer was last seen
-        lastUsed  util.AbsoluteTime // time the peer was last used
+        Peer     *util.PeerID      // peer identifier
+        Key      *crypto.HashCode  // address key is a sha512 hash
+        lastSeen util.AbsoluteTime // time the peer was last seen
+        lastUsed util.AbsoluteTime // time the peer was last used
 }
 
 // NewPeerAddress returns the DHT address of a peer.
 func NewPeerAddress(peer *util.PeerID) *PeerAddress {
-        r := new(PeerAddress)
-        h := rtHash()
-        h.Write(peer.Key)
-        copy(r.addr[:], h.Sum(nil))
-        r.lastSeen = util.AbsoluteTimeNow()
-        r.lastUsed = util.AbsoluteTimeNow()
-        return r
+        return &PeerAddress{
+                Peer:     peer,
+                Key:      crypto.Hash(peer.Data),
+                lastSeen: util.AbsoluteTimeNow(),
+                lastUsed: util.AbsoluteTimeNow(),
+        }
+}
+
+// NewQueryAddress returns a wrapped peer address for a query key
+func NewQueryAddress(key *crypto.HashCode) *PeerAddress {
+        return &PeerAddress{
+                Peer:     nil,
+                Key:      crypto.NewHashCode(key.Bits),
+                lastSeen: util.AbsoluteTimeNow(),
+                lastUsed: util.AbsoluteTimeNow(),
+        }
 }
 
 // String returns a human-readble representation of an address.
 func (addr *PeerAddress) String() string {
-        return hex.EncodeToString(addr.addr[:])
+        return hex.EncodeToString(addr.Key.Bits)
 }
 
 // Equals returns true if two peer addresses are the same.
 func (addr *PeerAddress) Equals(p *PeerAddress) bool {
-        return bytes.Equal(addr.addr[:], p.addr[:])
+        return bytes.Equal(addr.Key.Bits, p.Key.Bits)
 }
 
 // Distance between two addresses: returns a distance value and a
 // bucket index (smaller index = less distant).
 func (addr *PeerAddress) Distance(p *PeerAddress) (*math.Int, int) {
-        var d PeerAddress
-        for i := range d.addr {
-                d.addr[i] = addr.addr[i] ^ p.addr[i]
+        d := make([]byte, 64)
+        for i := range d {
+                d[i] = addr.Key.Bits[i] ^ p.Key.Bits[i]
         }
-        r := math.NewIntFromBytes(d.addr[:])
-        return r, numBuckets - r.BitLen()
+        r := math.NewIntFromBytes(d)
+        return r, 512 - r.BitLen()
 }
 
 //======================================================================
@@ -98,23 +101,28 @@ func (addr *PeerAddress) Distance(p *PeerAddress) (*math.Int, int) {
 // distance to the reference address, so smaller index means
 // "nearer" to the reference address.
 type RoutingTable struct {
-        ref       *PeerAddress              // reference address for distance
-        buckets   []*Bucket                 // list of buckets
-        list      map[*PeerAddress]struct{} // keep list of peers
-        mtx       sync.RWMutex              // lock for write operations
-        l2nse     float64                   // log2 of estimated network size
-        inProcess bool                      // flag if Process() is running
+        sync.RWMutex
+
+        ref        *PeerAddress                          // reference address for distance
+        buckets    []*Bucket                             // list of buckets
+        list       *util.Map[string, *PeerAddress]       // keep list of peers
+        l2nse      float64                               // log2 of estimated network size
+        inProcess  bool                                  // flag if Process() is running
+        cfg        *config.RoutingConfig                 // routing parameters
+        helloCache *util.Map[string, *blocks.HelloBlock] // HELLO block cache
 }
 
 // NewRoutingTable creates a new routing table for the reference address.
-func NewRoutingTable(ref *PeerAddress) *RoutingTable {
+func NewRoutingTable(ref *PeerAddress, cfg *config.RoutingConfig) *RoutingTable {
         // create routing table
         rt := &RoutingTable{
-                ref:       ref,
-                list:      make(map[*PeerAddress]struct{}),
-                buckets:   make([]*Bucket, numBuckets),
-                l2nse:     0.,
-                inProcess: false,
+                ref:        ref,
+                list:       util.NewMap[string, *PeerAddress](),
+                buckets:    make([]*Bucket, 512),
+                l2nse:      -1,
+                inProcess:  false,
+                cfg:        cfg,
+                helloCache: util.NewMap[string, *blocks.HelloBlock](),
         }
         // fill buckets
         for i := range rt.buckets {
@@ -130,41 +138,69 @@ func NewRoutingTable(ref *PeerAddress) *RoutingTable {
 // Add new peer address to routing table.
 // Returns true if the entry was added, false otherwise.
 func (rt *RoutingTable) Add(p *PeerAddress) bool {
-        // ensure one write and no readers
-        rt.lock(false)
-        defer rt.unlock(false)
+        k := p.String()
+        logger.Printf(logger.DBG, "[RT] Add(%s)", k)
 
         // check if peer is already known
-        if _, ok := rt.list[p]; ok {
+        if px, ok := rt.list.Get(k); ok {
+                logger.Println(logger.DBG, "[RT] --> already known")
+                px.lastSeen = util.AbsoluteTimeNow()
                 return false
         }
 
         // compute distance (bucket index) and insert address.
         _, idx := p.Distance(rt.ref)
         if rt.buckets[idx].Add(p) {
-                rt.list[p] = struct{}{}
+                logger.Println(logger.DBG, "[RT] --> entry added")
+                p.lastUsed = util.AbsoluteTimeNow()
+                rt.list.Put(k, p)
                 return true
         }
         // Full bucket: we did not add the address to the routing table.
+        logger.Println(logger.DBG, "[RT] --> bucket full -- discarded")
         return false
 }
 
 // Remove peer address from routing table.
 // Returns true if the entry was removed, false otherwise.
 func (rt *RoutingTable) Remove(p *PeerAddress) bool {
-        // ensure one write and no readers
-        rt.lock(false)
-        defer rt.unlock(false)
+        k := p.String()
+        logger.Printf(logger.DBG, "[RT] Remove(%s)", k)
 
         // compute distance (bucket index) and remove entry from bucket
+        rc := false
         _, idx := p.Distance(rt.ref)
         if rt.buckets[idx].Remove(p) {
-                delete(rt.list, p)
-                return true
+                logger.Println(logger.DBG, "[RT] --> entry removed from bucket and internal lists")
+                rc = true
+        } else {
+                // remove from internal list
+                logger.Println(logger.DBG, "[RT] --> entry removed from internal lists only")
         }
-        // remove from internal list
-        delete(rt.list, p)
-        return false
+        rt.list.Delete(k)
+        // delete from HELLO cache
+        rt.helloCache.Delete(p.Peer.String())
+        return rc
+}
+
+// Contains checks if a peer is available in the routing table
+func (rt *RoutingTable) Contains(p *PeerAddress) bool {
+        k := p.String()
+        logger.Printf(logger.DBG, "[RT] Contains(%s)?", k)
+
+        // check for peer in internal list
+        px, ok := rt.list.Get(k)
+        if !ok {
+                logger.Println(logger.DBG, "[RT] --> NOT found in current list:")
+                _ = rt.list.ProcessRange(func(key string, val *PeerAddress) error {
+                        logger.Printf(logger.DBG, "[RT]    * %s", val)
+                        return nil
+                }, true)
+        } else {
+                logger.Println(logger.DBG, "[RT] --> found in current list")
+                px.lastSeen = util.AbsoluteTimeNow()
+        }
+        return ok
 }
 
 //----------------------------------------------------------------------
@@ -186,51 +222,53 @@ func (rt *RoutingTable) Process(f func() error, readonly bool) error {
 // Routing functions
 //----------------------------------------------------------------------
 
-// SelectClosestPeer for a given peer address and bloomfilter.
-func (rt *RoutingTable) SelectClosestPeer(p *PeerAddress, bf *PeerBloomFilter) (n *PeerAddress) {
+// SelectClosestPeer for a given peer address and peer filter.
+func (rt *RoutingTable) SelectClosestPeer(p *PeerAddress, pf *blocks.PeerFilter) (n *PeerAddress) {
         // no writer allowed
-        rt.mtx.RLock()
-        defer rt.mtx.RUnlock()
+        rt.RLock()
+        defer rt.RUnlock()
 
-        // find closest address
+        // find closest peer in routing table
         var dist *math.Int
         for _, b := range rt.buckets {
-                if k, d := b.SelectClosestPeer(p, bf); n == nil || (d != nil && d.Cmp(dist) < 0) {
+                if k, d := b.SelectClosestPeer(p, pf); n == nil || (d != nil && d.Cmp(dist) < 0) {
                         dist = d
                         n = k
                 }
         }
         // mark peer as used
-        n.lastUsed = util.AbsoluteTimeNow()
+        if n != nil {
+                n.lastUsed = util.AbsoluteTimeNow()
+        }
         return
 }
 
 // SelectRandomPeer returns a random address from table (that is not
 // included in the bloomfilter)
-func (rt *RoutingTable) SelectRandomPeer(bf *PeerBloomFilter) *PeerAddress {
+func (rt *RoutingTable) SelectRandomPeer(pf *blocks.PeerFilter) (p *PeerAddress) {
         // no writer allowed
-        rt.mtx.RLock()
-        defer rt.mtx.RUnlock()
+        rt.RLock()
+        defer rt.RUnlock()
 
         // select random entry from list
-        if size := len(rt.list); size > 0 {
-                idx := rand.Intn(size)
-                for k := range rt.list {
-                        if idx == 0 {
-                                // mark peer as used
-                                k.lastUsed = util.AbsoluteTimeNow()
-                                return k
-                        }
-                        idx--
+        var ok bool
+        for {
+                if _, p, ok = rt.list.GetRandom(); !ok {
+                        return nil
+                }
+                if !pf.Contains(p.Peer) {
+                        break
                 }
         }
-        return nil
+        // mark peer as used
+        p.lastUsed = util.AbsoluteTimeNow()
+        return
 }
 
 // SelectPeer selects a neighbor depending on the number of hops parameter.
 // If hops < NSE this function MUST return SelectRandomPeer() and
 // SelectClosestpeer() otherwise.
-func (rt *RoutingTable) SelectPeer(p *PeerAddress, hops int, bf *PeerBloomFilter) *PeerAddress {
+func (rt *RoutingTable) SelectPeer(p *PeerAddress, hops int, bf *blocks.PeerFilter) *PeerAddress {
         if float64(hops) < rt.l2nse {
                 return rt.SelectRandomPeer(bf)
         }
@@ -238,9 +276,24 @@ func (rt *RoutingTable) SelectPeer(p *PeerAddress, hops int, bf *PeerBloomFilter
 }
 
 // IsClosestPeer returns true if p is the closest peer for k. Peers with a
-// positive test in the Bloom filter  are not considered.
-func (rt *RoutingTable) IsClosestPeer(p, k *PeerAddress, bf *PeerBloomFilter) bool {
-        n := rt.SelectClosestPeer(k, bf)
+// positive test in the Bloom filter are not considered. If p is nil, our
+// reference address is used.
+func (rt *RoutingTable) IsClosestPeer(p, k *PeerAddress, pf *blocks.PeerFilter) bool {
+        // get closest peer in routing table
+        n := rt.SelectClosestPeer(k, pf)
+        // check SELF?
+        if p == nil {
+                // if no peer in routing table found
+                if n == nil {
+                        // local peer is closest
+                        return true
+                }
+                // check if local distance is smaller than for best peer in routing table
+                d0, _ := n.Distance(k)
+                d1, _ := rt.ref.Distance(k)
+                return d1.Cmp(d0) < 0
+        }
+        // check if p is closest peer
         return n.Equals(p)
 }
 
@@ -248,7 +301,7 @@ func (rt *RoutingTable) IsClosestPeer(p, k *PeerAddress, bf *PeerBloomFilter) bo
 // The arguments are the desired replication level, the hop count of the message so far,
 // and the base-2 logarithm of the current network size estimate (L2NSE) as provided by the
 // underlay. The result is the non-negative number of next hops to select.
-func (rt *RoutingTable) ComputeOutDegree(repl, hop int) int {
+func (rt *RoutingTable) ComputeOutDegree(repl, hop uint16) int {
         hf := float64(hop)
         if hf > 4*rt.l2nse {
                 return 0
@@ -271,22 +324,62 @@ func (rt *RoutingTable) ComputeOutDegree(repl, hop int) int {
 func (rt *RoutingTable) heartbeat(ctx context.Context) {
 
         // check for dead or expired peers
-        timeout := util.NewRelativeTime(3 * time.Hour)
+        logger.Println(logger.DBG, "[dht] RT heartbeat...")
+        timeout := util.NewRelativeTime(time.Duration(rt.cfg.PeerTTL) * time.Second)
         if err := rt.Process(func() error {
-                for addr := range rt.list {
-                        if addr.connected {
-                                continue
-                        }
+                return rt.list.ProcessRange(func(k string, p *PeerAddress) error {
                         // check if we can/need to drop a peer
-                        drop := timeout.Compare(addr.lastSeen.Elapsed()) < 0
-                        if drop || timeout.Compare(addr.lastUsed.Elapsed()) < 0 {
-                                rt.Remove(addr)
+                        drop := timeout.Compare(p.lastSeen.Elapsed()) < 0
+                        if drop || timeout.Compare(p.lastUsed.Elapsed()) < 0 {
+                                logger.Printf(logger.DBG, "[RT] removing %v: %v, %v", p, p.lastSeen.Elapsed(), p.lastUsed.Elapsed())
+                                rt.Remove(p)
                         }
-                }
-                return nil
+                        return nil
+                }, false)
         }, false); err != nil {
-                logger.Println(logger.ERROR, "[dht] RT heartbeat: "+err.Error())
+                logger.Println(logger.ERROR, "[dht] RT heartbeat failed: "+err.Error())
         }
+
+        // drop expired entries from the HELLO cache
+        _ = rt.helloCache.ProcessRange(func(key string, val *blocks.HelloBlock) error {
+                if val.Expires.Expired() {
+                        rt.helloCache.Delete(key)
+                }
+                return nil
+        }, false)
+
+        // update the estimated network size
+        // rt.l2nse = ...
+}
+
+//----------------------------------------------------------------------
+
+func (rt *RoutingTable) BestHello(addr *PeerAddress, rf blocks.ResultFilter) (hb *blocks.HelloBlock, dist *math.Int) {
+        // iterate over cached HELLOs to find (best) match first
+        _ = rt.helloCache.ProcessRange(func(key string, val *blocks.HelloBlock) error {
+                // check if block is excluded by result filter
+                if !rf.Contains(val) {
+                        // check for better match
+                        p := NewPeerAddress(val.PeerID)
+                        d, _ := addr.Distance(p)
+                        if hb == nil || d.Cmp(dist) < 0 {
+                                hb = val
+                                dist = d
+                        }
+                }
+                return nil
+        }, true)
+        return
+}
+
+// CacheHello adds a HELLO block to the list of cached entries.
+func (rt *RoutingTable) CacheHello(hb *blocks.HelloBlock) {
+        rt.helloCache.Put(hb.PeerID.String(), hb)
+}
+
+// GetHello returns a HELLO block for key k (if available)
+func (rt *RoutingTable) GetHello(k string) (*blocks.HelloBlock, bool) {
+        return rt.helloCache.Get(k)
 }
 
 //----------------------------------------------------------------------
@@ -295,9 +388,9 @@ func (rt *RoutingTable) heartbeat(ctx context.Context) {
 func (rt *RoutingTable) lock(readonly bool) {
         if !rt.inProcess {
                 if readonly {
-                        rt.mtx.RLock()
+                        rt.RLock()
                 } else {
-                        rt.mtx.Lock()
+                        rt.Lock()
                 }
         }
 }
@@ -306,9 +399,9 @@ func (rt *RoutingTable) lock(readonly bool) {
 func (rt *RoutingTable) unlock(readonly bool) {
         if !rt.inProcess {
                 if readonly {
-                        rt.mtx.RUnlock()
+                        rt.RUnlock()
                 } else {
-                        rt.mtx.Unlock()
+                        rt.Unlock()
                 }
         }
 }
@@ -319,8 +412,9 @@ func (rt *RoutingTable) unlock(readonly bool) {
 
 // Bucket holds peer entries with approx. same distance from node
 type Bucket struct {
-        list   []*PeerAddress
-        rwlock sync.RWMutex
+        sync.RWMutex
+
+        list []*PeerAddress // list of peer addresses in bucket.
 }
 
 // NewBucket creates a new entry list of given size
@@ -334,8 +428,8 @@ func NewBucket(n int) *Bucket {
 // Returns true if entry is added, false otherwise.
 func (b *Bucket) Add(p *PeerAddress) bool {
         // only one writer and no readers
-        b.rwlock.Lock()
-        defer b.rwlock.Unlock()
+        b.Lock()
+        defer b.Unlock()
 
         // check for free space in bucket
         if len(b.list) < numK {
@@ -343,6 +437,7 @@ func (b *Bucket) Add(p *PeerAddress) bool {
                 b.list = append(b.list, p)
                 return true
         }
+        // full bucket: no further additions
         return false
 }
 
@@ -350,8 +445,8 @@ func (b *Bucket) Add(p *PeerAddress) bool {
 // Returns true if entry is removed (found), false otherwise.
 func (b *Bucket) Remove(p *PeerAddress) bool {
         // only one writer and no readers
-        b.rwlock.Lock()
-        defer b.rwlock.Unlock()
+        b.Lock()
+        defer b.Unlock()
 
         for i, pe := range b.list {
                 if pe.Equals(p) {
@@ -365,14 +460,14 @@ func (b *Bucket) Remove(p *PeerAddress) bool {
 
 // SelectClosestPeer returns the entry with minimal distance to the given
 // peer address; entries included in the bloom flter are ignored.
-func (b *Bucket) SelectClosestPeer(p *PeerAddress, bf *PeerBloomFilter) (n *PeerAddress, dist *math.Int) {
+func (b *Bucket) SelectClosestPeer(p *PeerAddress, pf *blocks.PeerFilter) (n *PeerAddress, dist *math.Int) {
         // no writer allowed
-        b.rwlock.RLock()
-        defer b.rwlock.RUnlock()
+        b.RLock()
+        defer b.RUnlock()
 
         for _, addr := range b.list {
                 // skip addresses in bloomfilter
-                if bf.Contains(addr) {
+                if pf.Contains(addr.Peer) {
                         continue
                 }
                 // check for shorter distance