dv: switch prefix table to object client

Author: pulsejet

dv/dv/prefix_sync.go

@@ -7,9 +7,8 @@ import (
 	"github.com/named-data/ndnd/dv/tlv"
 	enc "github.com/named-data/ndnd/std/encoding"
 	"github.com/named-data/ndnd/std/log"
-	"github.com/named-data/ndnd/std/ndn"
+	"github.com/named-data/ndnd/std/object"
 	ndn_sync "github.com/named-data/ndnd/std/sync"
-	"github.com/named-data/ndnd/std/utils"
 )
 
 // Fetch all required prefix data
@@ -60,31 +59,26 @@ func (dv *Router) prefixDataFetch(nodeId enc.Name) {
 	// Fetch the prefix data
 	log.Debugf("prefixDataFetch: fetching prefix data for %s [%d => %d]", nodeId, router.Known, router.Latest)
 
-	cfg := &ndn.InterestConfig{
-		MustBeFresh: true,
-		Lifetime:    utils.IdPtr(4 * time.Second),
-		Nonce:       utils.ConvertNonce(dv.engine.Timer().Nonce()),
-	}
-
-	isSnap := router.Latest-router.Known > 100
+	// Prefix object for other router
 	name := append(nodeId,
 		enc.NewStringComponent(enc.TypeKeywordNameComponent, "DV"),
 		enc.NewStringComponent(enc.TypeKeywordNameComponent, "PFX"),
 	)
-	if isSnap {
-		name = append(name, enc.NewStringComponent(enc.TypeKeywordNameComponent, "SNAP"))
-		cfg.CanBePrefix = true
+	if router.Latest-router.Known > table.PrefixTableSnapThreshold {
+		// no version - discover the latest snapshot
+		name = append(name, table.PrefixTableSnap)
 	} else {
-		name = append(name, enc.NewSequenceNumComponent(router.Known+1))
+		name = append(name,
+			enc.NewSequenceNumComponent(router.Known+1),
+			enc.NewVersionComponent(0), // immutable
+		)
 	}
 
-	interest, err := dv.engine.Spec().MakeInterest(name, cfg, nil, nil)
-	if err != nil {
-		log.Warnf("prefixDataFetch: failed to make Interest: %+v", err)
-		return
-	}
+	dv.client.Consume(name, func(state *object.ConsumeState) bool {
+		if !state.IsComplete() {
+			return true
+		}
 
-	err = dv.engine.Express(interest, func(args ndn.ExpressCallbackArgs) {
 		go func() {
 			// Done fetching, restart if needed
 			defer func() {
@@ -95,30 +89,22 @@ func (dv *Router) prefixDataFetch(nodeId enc.Name) {
 				go dv.prefixDataFetch(nodeId) // recheck
 			}()
 
-			// Sleep this goroutine if no data was received
-			if args.Result != ndn.InterestResultData {
-				log.Warnf("prefixDataFetch: failed to fetch prefix data %s: %d", interest.FinalName, args.Result)
-
-				// see advertDataFetch
-				if args.Result != ndn.InterestResultTimeout {
-					time.Sleep(2 * time.Second)
-				} else {
-					time.Sleep(100 * time.Millisecond)
-				}
+			// Wait before retry if there was a failure
+			if err := state.Error(); err != nil {
+				log.Warnf("prefixDataFetch: failed to fetch prefix data %s: %+v", name, err)
+				time.Sleep(1 * time.Second)
 				return
 			}
 
-			dv.processPrefixData(args.Data, router)
+			dv.processPrefixData(state.Name(), state.Content(), router)
 		}()
+
+		return true
 	})
-	if err != nil {
-		log.Warnf("prefixDataFetch: failed to express Interest: %+v", err)
-		return
-	}
 }
 
-func (dv *Router) processPrefixData(data ndn.Data, router *table.PrefixTableRouter) {
-	ops, err := tlv.ParsePrefixOpList(enc.NewWireReader(data.Content()), true)
+func (dv *Router) processPrefixData(name enc.Name, data []byte, router *table.PrefixTableRouter) {
+	ops, err := tlv.ParsePrefixOpList(enc.NewBufferReader(data), true)
 	if err != nil {
 		log.Warnf("prefixDataFetch: failed to parse PrefixOpList: %+v", err)
 		return
@@ -127,18 +113,21 @@ func (dv *Router) processPrefixData(data ndn.Data, router *table.PrefixTableRout
 	dv.mutex.Lock()
 	defer dv.mutex.Unlock()
 
-	// Update sequence number
-	dataName := data.Name()
-	seqNo := dataName[len(dataName)-1]
-	if seqNo.Typ != enc.TypeSequenceNumNameComponent {
+	// Get sequence number from name
+	seqNo := name[len(name)-2]
+	if seqNo.Equal(table.PrefixTableSnap) && name[len(name)-1].Typ == enc.TypeVersionNameComponent {
+		// version is sequence number for snapshot
+		seqNo = name[len(name)-1]
+	} else if seqNo.Typ != enc.TypeSequenceNumNameComponent {
+		// version is immutable, sequence number is in name
 		log.Warnf("prefixDataFetch: unexpected sequence number type: %s", seqNo.Typ)
 		return
 	}
 
 	// Update the prefix table
 	router.Known = seqNo.NumberVal()
 	if dv.pfx.Apply(ops) {
-		// Update the local fib if prefix table changed (very expensive)
-		go dv.fibUpdate()
+		// Update the local fib if prefix table changed
+		go dv.fibUpdate() // very expensive
 	}
 }

dv/dv/router.go

@@ -11,6 +11,7 @@ import (
 	"github.com/named-data/ndnd/std/log"
 	"github.com/named-data/ndnd/std/ndn"
 	mgmt "github.com/named-data/ndnd/std/ndn/mgmt_2022"
+	"github.com/named-data/ndnd/std/object"
 	ndn_sync "github.com/named-data/ndnd/std/sync"
 	"github.com/named-data/ndnd/std/utils"
 )
@@ -20,6 +21,8 @@ type Router struct {
 	engine ndn.Engine
 	// config for this router
 	config *config.Config
+	// object client
+	client *object.Client
 	// nfd management thread
 	nfdc *nfdc.NfdMgmtThread
 	// single mutex for all operations
@@ -59,6 +62,7 @@ func NewRouter(config *config.Config, engine ndn.Engine) (*Router, error) {
 	dv := &Router{
 		engine: engine,
 		config: config,
+		client: object.NewClient(engine, object.NewMemoryStore()),
 		nfdc:   nfdc.NewNfdMgmtThread(engine),
 		mutex:  sync.Mutex{},
 	}
@@ -74,7 +78,7 @@ func NewRouter(config *config.Config, engine ndn.Engine) (*Router, error) {
 	// Create tables
 	dv.neighbors = table.NewNeighborTable(config, dv.nfdc)
 	dv.rib = table.NewRib(config)
-	dv.pfx = table.NewPrefixTable(config, engine, dv.pfxSvs)
+	dv.pfx = table.NewPrefixTable(config, dv.client, dv.pfxSvs)
 	dv.fib = table.NewFib(config, dv.nfdc)
 
 	return dv, nil
@@ -94,6 +98,10 @@ func (dv *Router) Start() (err error) {
 	defer dv.heartbeat.Stop()
 	defer dv.deadcheck.Stop()
 
+	// Start object client
+	dv.client.Start()
+	defer dv.client.Stop()
+
 	// Start management thread
 	go dv.nfdc.Start()
 	defer dv.nfdc.Stop()
@@ -179,15 +187,6 @@ func (dv *Router) register() (err error) {
 		return err
 	}
 
-	// Prefix Data
-	err = dv.engine.AttachHandler(dv.config.PrefixTableDataPrefix(),
-		func(args ndn.InterestHandlerArgs) {
-			go dv.pfx.OnDataInterest(args)
-		})
-	if err != nil {
-		return err
-	}
-
 	// Readvertise Data
 	err = dv.engine.AttachHandler(dv.config.ReadvertisePrefix(),
 		func(args ndn.InterestHandlerArgs) {

dv/scripts/add_routes.sh

@@ -0,0 +1,27 @@
+#!/bin/bash
+
+# This test script adds and removes routes from a local DV forwarder.
+# This can be used to test a large number of changes to the RIB.
+
+MAX_RIB_SIZE=3
+
+ITER=0
+while true; do
+    ITER=$((ITER+1))
+
+    RIB_SIZE=0
+    while [ $RIB_SIZE -lt $MAX_RIB_SIZE ]; do
+        RIB_SIZE=$((RIB_SIZE+1))
+        ndnd fw route add prefix=/my/route/$RIB_SIZE/$ITER origin=65 face=1
+        sleep 0.05
+    done
+
+    sleep 0.1
+
+    RIB_SIZE=0
+    while [ $RIB_SIZE -lt $MAX_RIB_SIZE ]; do
+        RIB_SIZE=$((RIB_SIZE+1))
+        ndnd fw route remove prefix=/my/route/$RIB_SIZE/$ITER origin=65 face=1
+        sleep 0.05
+    done
+done

dv/table/prefix_table.go

@@ -1,29 +1,27 @@
 package table
 
 import (
-	"sync"
-	"time"
-
 	"github.com/named-data/ndnd/dv/config"
 	"github.com/named-data/ndnd/dv/tlv"
 	enc "github.com/named-data/ndnd/std/encoding"
 	"github.com/named-data/ndnd/std/log"
-	"github.com/named-data/ndnd/std/ndn"
-	"github.com/named-data/ndnd/std/security"
+	"github.com/named-data/ndnd/std/object"
 	ndn_sync "github.com/named-data/ndnd/std/sync"
 	"github.com/named-data/ndnd/std/utils"
 )
 
+const PrefixTableSnapThreshold = 100
+
+var PrefixTableSnap = enc.NewStringComponent(enc.TypeKeywordNameComponent, "SNAP")
+
 type PrefixTable struct {
 	config *config.Config
-	engine ndn.Engine
+	client *object.Client
 	svs    *ndn_sync.SvSync
 
 	routers map[uint64]*PrefixTableRouter
 	me      *PrefixTableRouter
 
-	repo       map[uint64][]byte
-	repoMutex  sync.RWMutex
 	snapshotAt uint64
 }
 
@@ -41,19 +39,16 @@ type PrefixEntry struct {
 
 func NewPrefixTable(
 	config *config.Config,
-	engine ndn.Engine,
+	client *object.Client,
 	svs *ndn_sync.SvSync,
 ) *PrefixTable {
 	pt := &PrefixTable{
 		config: config,
-		engine: engine,
+		client: client,
 		svs:    svs,
 
 		routers: make(map[uint64]*PrefixTableRouter),
 		me:      nil,
-
-		repo:      make(map[uint64][]byte),
-		repoMutex: sync.RWMutex{},
 	}
 
 	pt.me = pt.GetRouter(config.RouterName())
@@ -155,12 +150,19 @@ func (pt *PrefixTable) publishOp(content enc.Wire) {
 	pt.me.Known = seq
 	pt.me.Latest = seq
 
-	// Create the new data
-	name := append(pt.config.PrefixTableDataPrefix(), enc.NewSequenceNumComponent(seq))
-	pt.publish(name, content)
+	// Produce the operation
+	_, err := pt.client.Produce(object.ProduceArgs{
+		Name:    append(pt.config.PrefixTableDataPrefix(), enc.NewSequenceNumComponent(seq)),
+		Content: content,
+		Version: utils.IdPtr(uint64(0)), // immutable
+	})
+	if err != nil {
+		log.Errorf("prefix-table: failed to produce op: %v", err)
+		return
+	}
 
 	// Create snapshot if needed
-	if pt.snapshotAt-seq >= 100 {
+	if seq-pt.snapshotAt >= PrefixTableSnapThreshold/2 {
 		pt.publishSnap()
 	}
 }
@@ -179,59 +181,14 @@ func (pt *PrefixTable) publishSnap() {
 		})
 	}
 
-	// Store snapshot in repo
-	// TODO: this can be a segmented object
-	pt.snapshotAt = pt.me.Latest
-	snapPfx := append(pt.config.PrefixTableDataPrefix(),
-		enc.NewStringComponent(enc.TypeKeywordNameComponent, "SNAP"))
-	snapName := append(snapPfx, enc.NewSequenceNumComponent(pt.snapshotAt))
-	pt.publish(snapName, snap.Encode())
-
-	// Point prefix to the snapshot
-	pt.repoMutex.Lock()
-	defer pt.repoMutex.Unlock()
-	pt.repo[snapPfx.Hash()] = pt.repo[snapName.Hash()]
-}
-
-func (pt *PrefixTable) publish(name enc.Name, content enc.Wire) {
-	// TODO: sign the prefix table data
-	signer := security.NewSha256Signer()
-
-	data, err := pt.engine.Spec().MakeData(
-		name,
-		&ndn.DataConfig{
-			ContentType: utils.IdPtr(ndn.ContentTypeBlob),
-			Freshness:   utils.IdPtr(1 * time.Second),
-		},
-		content,
-		signer)
+	_, err := pt.client.Produce(object.ProduceArgs{
+		Name:    append(pt.config.PrefixTableDataPrefix(), PrefixTableSnap),
+		Content: snap.Encode(),
+		Version: utils.IdPtr(pt.me.Latest),
+	})
 	if err != nil {
-		log.Warnf("prefix-table: publish failed to make data: %+v", err)
-		return
-	}
-
-	// Store the data packet in our mem repo
-	pt.repoMutex.Lock()
-	defer pt.repoMutex.Unlock()
-	pt.repo[name.Hash()] = data.Wire.Join()
-}
-
-// Received prefix data Interest
-func (pt *PrefixTable) OnDataInterest(args ndn.InterestHandlerArgs) {
-	// TODO: remove old entries from repo
-
-	pt.repoMutex.RLock()
-	defer pt.repoMutex.RUnlock()
-
-	// Find exact match in repo
-	name := args.Interest.Name()
-	if data := pt.repo[name.Hash()]; data != nil {
-		err := args.Reply(enc.Wire{data})
-		if err != nil {
-			log.Warnf("prefix-table: failed to reply: %+v", err)
-		}
-		return
+		log.Errorf("prefix-table: failed to produce snap: %v", err)
 	}
 
-	log.Warnf("prefix-table: repo failed to find data for for %s", name)
+	pt.snapshotAt = pt.me.Latest
 }

std/object/client.go

@@ -67,6 +67,11 @@ func (c *Client) Engine() ndn.Engine {
 	return c.engine
 }
 
+// Get the underlying store
+func (c *Client) Store() ndn.Store {
+	return c.store
+}
+
 // Main goroutine for all client processing
 func (c *Client) run() {
 	for {

std/object/client_consume.go

@@ -42,7 +42,7 @@ type ConsumeState struct {
 
 // returns the name of the object being consumed
 func (a *ConsumeState) Name() enc.Name {
-	return a.name
+	return a.fetchName
 }
 
 // returns the error that occurred during fetching

std/object/store_memory.go

@@ -90,6 +90,14 @@ func (s *MemoryStore) Rollback() error {
 	return nil
 }
 
+func (s *MemoryStore) MemSize() int {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+	size := 0
+	s.root.walk(func(n *memoryStoreNode) { size += len(n.wire) })
+	return size
+}
+
 func (n *memoryStoreNode) find(name enc.Name) *memoryStoreNode {
 	if len(name) == 0 {
 		return n
@@ -183,3 +191,10 @@ func (n *memoryStoreNode) merge(tx *memoryStoreNode) {
 		}
 	}
 }
+
+func (n *memoryStoreNode) walk(f func(*memoryStoreNode)) {
+	f(n)
+	for _, child := range n.children {
+		child.walk(f)
+	}
+}