Merge pull request #176 from asoberai/dv2

[fw] initial BIER support

Author: tianyuan129

cmd/cmd.go

@@ -51,6 +51,7 @@ func init() {
 	CmdNDNd.AddCommand(tools.CmdPingServer())
 	CmdNDNd.AddCommand(tools.CmdCatChunks())
 	CmdNDNd.AddCommand(tools.CmdPutChunks())
+	CmdNDNd.AddCommand(tools.CmdSvsChat())
 }
 
 // (AI GENERATED DESCRIPTION): Creates the top‑level “fw” command for managing the NDN Forwarding Daemon, adding a “run” subcommand to start the daemon and a set of “nfdc” control subcommands for configuring it.

cmd/svs-chat/main.go

@@ -0,0 +1,106 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+	"time"
+
+	enc "github.com/named-data/ndnd/std/encoding"
+	eng "github.com/named-data/ndnd/std/engine"
+	"github.com/named-data/ndnd/std/log"
+	"github.com/named-data/ndnd/std/ndn"
+	"github.com/named-data/ndnd/std/object"
+	"github.com/named-data/ndnd/std/object/storage"
+	"github.com/named-data/ndnd/std/sync"
+)
+
+func main() {
+	log.Default().SetLevel(log.LevelTrace)
+	prefixStr := flag.String("prefix", "/minindn/svs", "SVS group prefix")
+	nameStr := flag.String("name", "", "Participant name")
+	msgStr := flag.String("msg", "", "Message to send")
+	delaySecs := flag.Int("delay", 0, "Wait N seconds before sending the message")
+	waitSecs := flag.Int("wait", 10, "Seconds to wait before exiting")
+	flag.Parse()
+
+	if *nameStr == "" {
+		fmt.Println("Error: -name is required")
+		os.Exit(1)
+	}
+
+	groupPrefix, _ := enc.NameFromStr(*prefixStr)
+	nodeName, _ := enc.NameFromStr(*nameStr)
+
+	app := eng.NewBasicEngine(eng.NewDefaultFace())
+	if err := app.Start(); err != nil {
+		fmt.Printf("Failed to start engine: %v\n", err)
+		os.Exit(1)
+	}
+	defer app.Stop()
+
+	store := storage.NewMemoryStore()
+	client := object.NewClient(app, store, nil)
+
+	err := client.Start()
+	if err != nil {
+		panic(err)
+	}
+	defer client.Stop()
+
+	syncPrefix := groupPrefix.Append(enc.NewKeywordComponent("svs"))
+	client.AnnouncePrefix(ndn.Announcement{Name: syncPrefix, Expose: true, Multicast: true})
+
+	dataPrefix := groupPrefix.Append(nodeName...)
+	client.AnnouncePrefix(ndn.Announcement{Name: dataPrefix, Expose: true})
+
+	alo, err := sync.NewSvsALO(sync.SvsAloOpts{
+		Name: nodeName,
+		Svs: sync.SvSyncOpts{
+			Client:      client,
+			GroupPrefix: groupPrefix,
+			BootTime:    1,
+		},
+		Snapshot: &sync.SnapshotNodeLatest{
+			Client: client,
+			SnapMe: func(_ enc.Name) (enc.Wire, error) {
+				if *msgStr != "" {
+					return enc.Wire{[]byte(*msgStr)}, nil
+				}
+				return enc.Wire{[]byte("(no message)")}, nil
+			},
+			Threshold: 5,
+		},
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	alo.SetOnPublisher(func(name enc.Name) {
+		alo.SubscribePublisher(name, func(pub sync.SvsPub) {
+			fmt.Printf("CHAT %s: %s\n", pub.Publisher.String(), string(pub.Bytes()))
+		})
+	})
+
+	if err := alo.Start(); err != nil {
+		panic(err)
+	}
+
+	// Important: let the sync initialization happen
+	time.Sleep(2 * time.Second)
+
+	if *msgStr != "" {
+		if *delaySecs > 0 {
+			time.Sleep(time.Duration(*delaySecs) * time.Second)
+		}
+		_, _, err := alo.Publish(enc.Wire{[]byte(*msgStr)})
+		if err != nil {
+			fmt.Printf("Publish error: %v\n", err)
+		} else {
+			fmt.Printf("Published message: %s\n", *msgStr)
+		}
+	}
+
+	time.Sleep(time.Duration(*waitSecs) * time.Second)
+	alo.Stop()
+}

dv/dv/insertion.go

@@ -213,7 +213,7 @@ func (pfx *PrefixModule) onPrefixInsertionObject(object ndn.Data, faceId uint64)
 		return resError
 	}
 
-	pfx.Announce(prefix, faceId, cost, params.ValidityPeriod)
+	pfx.Announce(prefix, faceId, cost, false, params.ValidityPeriod)
 
 	return &mgmt.ControlResponse{
 		Val: &mgmt.ControlResponseVal{

dv/dv/mgmt.go

@@ -292,8 +292,9 @@ func (dv *Router) mgmtOnPrefix(args ndn.InterestHandlerArgs) {
 			responseParams.ExpirationPeriod = optional.Some(expires)
 		}
 
+		multicast := params.Val.Flags.GetOr(0)&1 != 0
 		dv.mutex.Lock()
-		dv.pfx.Announce(name, faceID, cost, validity)
+		dv.pfx.Announce(name, faceID, cost, multicast, validity)
 		dv.mutex.Unlock()
 
 		res.Val.StatusCode = 200

dv/dv/prefix.go

@@ -45,9 +45,10 @@ type PrefixModule struct {
 }
 
 type petEgressOp struct {
-	add    bool
-	name   enc.Name
-	egress enc.Name
+	add       bool
+	name      enc.Name
+	egress    enc.Name
+	multicast bool
 }
 
 type petNextHopOp struct {
@@ -225,10 +226,14 @@ func (pfx *PrefixModule) Reset() {
 
 // Announce adds or updates a local prefix in prefix egress state.
 // Use face=0 and cost=0 for route-only semantics.
-func (pfx *PrefixModule) Announce(name enc.Name, face uint64, cost uint64, validity *spec.ValidityPeriod) {
+func (pfx *PrefixModule) Announce(name enc.Name, face uint64, cost uint64, multicast bool, validity *spec.ValidityPeriod) {
+	pfx.announce(name, face, cost, multicast, validity)
+}
+
+func (pfx *PrefixModule) announce(name enc.Name, face uint64, cost uint64, multicast bool, validity *spec.ValidityPeriod) {
 	pfx.mu.Lock()
-	petOps := pfx.addRouterPrefixPet(pfx.routerName, name)
-	pfx.pfx.Announce(name, face, cost, validity)
+	petOps := pfx.addRouterPrefixPet(pfx.routerName, name, multicast)
+	pfx.pfx.Announce(name, face, cost, multicast, validity)
 	pfx.mu.Unlock()
 
 	pfx.applyPetOps(petOps)
@@ -330,7 +335,7 @@ func (pfx *PrefixModule) processUpdate(wire enc.Wire) (dirty bool, petOps []petE
 			petOps = append(petOps, pfx.resetRouterPet(router)...)
 		}
 		for _, add := range ops.PrefixOpAdds {
-			petOps = append(petOps, pfx.addRouterPrefixPet(router, add.Name)...)
+			petOps = append(petOps, pfx.addRouterPrefixPet(router, add.Name, add.Multicast)...)
 		}
 		for _, remove := range ops.PrefixOpRemoves {
 			petOps = append(petOps, pfx.removeRouterPrefixPet(router, remove.Name)...)
@@ -360,7 +365,7 @@ func (pfx *PrefixModule) resetRouterPet(router enc.Name) []petEgressOp {
 	return ops
 }
 
-func (pfx *PrefixModule) addRouterPrefixPet(router enc.Name, prefix enc.Name) []petEgressOp {
+func (pfx *PrefixModule) addRouterPrefixPet(router enc.Name, prefix enc.Name, multicast bool) []petEgressOp {
 	routerHash := router.Hash()
 	prefixes := pfx.petPrefixes[routerHash]
 	if prefixes == nil {
@@ -376,9 +381,10 @@ func (pfx *PrefixModule) addRouterPrefixPet(router enc.Name, prefix enc.Name) []
 
 	egress := router.Clone()
 	return []petEgressOp{{
-		add:    true,
-		name:   prefix.Clone(),
-		egress: egress,
+		add:       true,
+		name:      prefix.Clone(),
+		egress:    egress,
+		multicast: multicast,
 	}}
 }
 
@@ -413,17 +419,22 @@ func (pfx *PrefixModule) applyPetOps(ops []petEgressOp) {
 
 	for _, op := range ops {
 		cmd := "remove-egress"
+		args := &mgmt.ControlArgs{
+			Name:   op.name,
+			Egress: &mgmt.EgressRecord{Name: op.egress},
+		}
 		if op.add {
 			cmd = "add-egress"
+			// Signal the Sync group (multicast) flag to the forwarder's PET via Flags bit 0.
+			if op.multicast {
+				args.Flags = optional.Some(uint64(1))
+			}
 		}
 
 		pfx.nfdc.Exec(nfdc.NfdMgmtCmd{
-			Module: "pet",
-			Cmd:    cmd,
-			Args: &mgmt.ControlArgs{
-				Name:   op.name,
-				Egress: &mgmt.EgressRecord{Name: op.egress},
-			},
+			Module:  "pet",
+			Cmd:     cmd,
+			Args:    args,
 			Retries: -1,
 		})
 	}

dv/dv/table_algo.go

@@ -3,6 +3,7 @@ package dv
 import (
 	"github.com/named-data/ndnd/dv/config"
 	"github.com/named-data/ndnd/dv/table"
+	fw "github.com/named-data/ndnd/fw/fw"
 	enc "github.com/named-data/ndnd/std/encoding"
 	"github.com/named-data/ndnd/std/log"
 )
@@ -134,4 +135,10 @@ func (dv *Router) updateFib() {
 		}
 	}
 	dv.fib.RemoveUnmarked()
+
+	// Rebuild the BIFT whenever the FIB changes so BIER forwarding paths
+	// are always consistent with the routing table.
+	if fw.IsBierEnabled() {
+		fw.Bift.BuildFromFib()
+	}
 }

dv/table/prefix_state.go

@@ -94,16 +94,21 @@ func (pt *PrefixEgreState) Reset() {
 
 // Announce updates or creates a local prefix with an optional validity period.
 // Use face=0 and cost=0 for route-only semantics.
-func (pt *PrefixEgreState) Announce(name enc.Name, face uint64, cost uint64, validity *spec.ValidityPeriod) {
+// multicast=true marks this as a Sync group prefix (vs. a producer prefix).
+func (pt *PrefixEgreState) Announce(name enc.Name, face uint64, cost uint64, multicast bool, validity *spec.ValidityPeriod) {
 	hash := name.TlvStr()
 	entry := pt.me.Prefixes[hash]
 	publishAdd := false
 	if entry == nil {
 		entry = &PrefixEntry{
-			Name: name,
+			Name:      name,
+			Multicast: multicast,
 		}
 		pt.me.Prefixes[hash] = entry
 		publishAdd = true
+	} else if multicast && !entry.Multicast {
+		entry.Multicast = true
+		publishAdd = true
 	}
 
 	if !sameValidityPeriod(entry.ValidityPeriod, validity) {

e2e-run.sh

@@ -0,0 +1,33 @@
+#!/usr/bin/env bash
+# e2e-run.sh — Docker entrypoint: install binaries and run e2e tests.
+set -euo pipefail
+
+log() { echo "==> $*"; }
+
+# OpenVSwitch is required for Mininet virtual switches.
+# ovs-vsctl can hang on macOS linuxkit kernels, so both calls are guarded with timeout.
+log "Starting OpenVSwitch..."
+timeout 15 service openvswitch-switch start 2>/dev/null || true
+timeout 5  ovs-vsctl set-manager ptcp:6640  2>/dev/null || true
+
+log "Installing binaries..."
+install -m 755 /ndnd/.bin/ndnd /usr/local/bin/ndnd
+log "  ndnd: $(ndnd --version 2>&1 | head -1 || echo unknown)"
+
+# Mini-NDN's NFD config template needs readvertise_nlsr enabled.
+sed -i 's/readvertise_nlsr no/readvertise_nlsr yes/g' \
+  /usr/local/etc/ndn/nfd.conf.sample 2>/dev/null || true
+
+# runner.py calls `go build` to ensure fresh binaries.
+# Pre-built binaries are already in .bin/ so we shim `go` to be a no-op.
+mkdir -p /usr/local/fake-go
+cat > /usr/local/fake-go/go << 'EOF'
+#!/bin/bash
+exit 0
+EOF
+chmod +x /usr/local/fake-go/go
+export PATH="/usr/local/fake-go:$PATH"
+
+log "Running e2e tests..."
+cd /ndnd
+python3 e2e/runner.py e2e/topo.sprint.conf

e2e/dv_util.py

@@ -88,3 +88,25 @@ def wait_prefix_pet_ready(node_to_prefixes: dict[Node, set[str]], deadline=30) -
         time.sleep(1)
 
     raise Exception('PET prefix replication did not converge')
+
+def populate_bift(nodes: list[Node], bier_map: dict, network=DEFAULT_NETWORK):
+    info(f'Deploying BIER indices to BIFT ({len(bier_map)} routers)...\n')
+    for node in nodes:
+        for router, idx in bier_map.items():
+            router_name = f'{network}/{router.name}'
+            node.cmd(f'ndnd fw bift-register prefix="{router_name}" cost={idx}')
+        node.cmd('ndnd fw bift-rebuild')
+    info('BIFT populated on all nodes\n')
+
+
+def dump_bier_logs(nodes: list[Node], label: str = '', lines: int = 40) -> str:
+    """Return BIER-relevant log lines from each node's ndnd log (for failure diagnostics)."""
+    out = f'\n=== BIER forwarder logs{" (" + label + ")" if label else ""} ===\n'
+    for node in nodes:
+        log = node.cmd(
+            f'cat /tmp/minindn/{node.name}/log/yanfd.log'
+            f' 2>/dev/null | grep -iE "bier|bift|bfir|bfr|bfer|strategy" | tail -{lines}'
+        )
+        if log.strip():
+            out += f'--- {node.name} ---\n{log}\n'
+    return out

e2e/fw.py

@@ -5,7 +5,7 @@
 from minindn.apps.application import Application
 
 class NDNd_FW(Application):
-    def __init__(self, node, config={}, logLevel='INFO', threads=2, network='/minindn'):
+    def __init__(self, node, config={}, logLevel='INFO', threads=2, network='/minindn', bier_index=-1):
         Application.__init__(self, node)
 
         if not shutil.which('ndnd'):
@@ -37,6 +37,8 @@ def __init__(self, node, config={}, logLevel='INFO', threads=2, network='/minind
                 'threads': threads,
                 # Two-phase PET forwarding needs the local router identity.
                 'router_name': f'{network}/{node.name}',
+                # BIER bit index for this router (-1 = disabled)
+                'bier_index': bier_index,
             },
         }