[fw] broadcast/replicast forwarding strategy + best-route override

dv/config/config.go

@@ -27,6 +27,16 @@ var MulticastStrategy = enc.LOCALHOST.
 	Append(enc.NewGenericComponent("strategy")).
 	Append(enc.NewGenericComponent("multicast"))
 
+var BroadcastStrategy = enc.LOCALHOST.
+	Append(enc.NewGenericComponent("nfd")).
+	Append(enc.NewGenericComponent("strategy")).
+	Append(enc.NewGenericComponent("broadcast"))
+
+var ReplicastStrategy = enc.LOCALHOST.
+	Append(enc.NewGenericComponent("nfd")).
+	Append(enc.NewGenericComponent("strategy")).
+	Append(enc.NewGenericComponent("replicast"))
+
 //go:embed schema.tlv
 var SchemaBytes []byte
 

dv/dv/router.go

@@ -315,15 +315,15 @@ func (dv *Router) register() (err error) {
 		},
 		Retries: -1,
 	})
-	// Set strategy to multicast for advertisement sync Interests, so
+	// Set strategy to broadcast for advertisement sync Interests, so
 	// /localhop/.../DV/ADS traffic fan-outs to all neighbor nexthops.
 	dv.nfdc.Exec(nfdc.NfdMgmtCmd{
 		Module: "strategy-choice",
 		Cmd:    "set",
 		Args: &mgmt.ControlArgs{
 			Name: dv.config.AdvertisementSyncPrefix(),
 			Strategy: &mgmt.Strategy{
-				Name: config.MulticastStrategy,
+				Name: config.BroadcastStrategy,
 			},
 		},
 		Retries: -1,
@@ -334,7 +334,7 @@ func (dv *Router) register() (err error) {
 		Args: &mgmt.ControlArgs{
 			Name: dv.pfx.SyncPrefix(),
 			Strategy: &mgmt.Strategy{
-				Name: config.MulticastStrategy,
+				Name: config.ReplicastStrategy,
 			},
 		},
 		Retries: -1,

e2e/run_scenario.py

@@ -16,11 +16,12 @@ def ensure_local_ndnd(repo_root: Path) -> None:
     local_bin = repo_root / ".bin"
     local_ndnd = local_bin / "ndnd"
     local_bin.mkdir(parents=True, exist_ok=True)
-    info("Building local ndnd binary for E2E scenario\n")
-    subprocess.check_call(
-        ["go", "build", "-o", str(local_ndnd), "./cmd/ndnd"],
-        cwd=repo_root,
-    )
+    if not local_ndnd.exists():
+        info("Building local ndnd binary for E2E scenario\n")
+        subprocess.check_call(
+            ["go", "build", "-o", str(local_ndnd), "./cmd/ndnd"],
+            cwd=repo_root,
+        )
     os.environ["PATH"] = f"{local_bin}:{os.environ.get('PATH', '')}"
 
 
@@ -37,7 +38,6 @@ def main():
     ensure_local_ndnd(repo_root)
 
     setLogLevel("info")
-    random.seed(0)
 
     Minindn.cleanUp()
     Minindn.verifyDependencies()
@@ -49,6 +49,8 @@ def main():
         scenario = getattr(mod, "scenario")
         sig = inspect.signature(scenario)
 
+        random.seed(0)
+
         info("===================================================\n")
         start = time.time()
         if "network" in sig.parameters:

e2e/runner.py

@@ -14,18 +14,21 @@
 import test_002
 import test_003
 import test_004
+import test_005
+import test_006
 
 
 def ensure_local_ndnd() -> None:
     repo_root = Path(__file__).resolve().parent.parent
     local_bin = repo_root / ".bin"
     local_ndnd = local_bin / "ndnd"
     local_bin.mkdir(parents=True, exist_ok=True)
-    info("Building local ndnd binary for E2E scenarios\n")
-    subprocess.check_call(
-        ["go", "build", "-o", str(local_ndnd), "./cmd/ndnd"],
-        cwd=repo_root,
-    )
+    if not local_ndnd.exists():
+        info("Building local ndnd binary for E2E scenarios\n")
+        subprocess.check_call(
+            ["go", "build", "-o", str(local_ndnd), "./cmd/ndnd"],
+            cwd=repo_root,
+        )
     os.environ["PATH"] = f"{local_bin}:{os.environ.get('PATH', '')}"
 
 
@@ -66,5 +69,7 @@ def run(scenario: FunctionType, **kwargs) -> None:
     run(test_002.scenario)
     run(test_003.scenario)
     run(test_004.scenario)
+    run(test_005.scenario)
+    run(test_006.scenario)
 
     ndn.stop()

e2e/test_001.py

@@ -58,6 +58,22 @@ def scenario(ndn: Minindn, network='/minindn'):
         if f'/localhop{network}/' in fib and '/32=DV' in fib:
             raise Exception(f'Router-specific localhop DV entries unexpectedly present in FIB on {node.name}')
 
+    # Validate the deprecation of multicast in DV code (#174)
+    for node in ndn.net.hosts:
+        strategy = node.cmd('ndnd fw strategy-list')
+        if "multicast" in strategy:
+            raise Exception(f'Multicast is to be retired, unexpectedly present in strategy on {node.name}')
+        expected_strategies = [
+            # SVS sync interests should use replicast (#174)
+            "prefix=/minindn/32=DV/32=PES/32=svs strategy=/localhost/nfd/strategy/replicast/v=1",
+
+            # Localhop advertisement sync interests should use broadcast strategy (#174)
+            "prefix=/localhop/minindn/32=DV/32=ADS strategy=/localhost/nfd/strategy/broadcast/v=1",
+        ]
+        for expected_strat in expected_strategies:
+            if expected_strat not in strategy:
+                raise Exception(f'Strategy {expected_strat!r} not in strategy on {node.name}')
+
     for node, put_node in cat_requests:
         cmd = f'ndnd cat "{network}/{put_node.name}/test" > recv.test.bin 2> cat.log'
         info(f'{node.name} {cmd}\n')

e2e/test_005.py

@@ -0,0 +1,73 @@
+import random
+import os
+
+from mininet.log import info
+from minindn.minindn import Minindn
+from minindn.apps.app_manager import AppManager
+
+from fw import NDNd_FW
+import dv_util
+
+def scenario(ndn: Minindn, network='/minindn'):
+    """
+    Simple file transfer scenario with NDNd forwarder.
+    This tests routing convergence and cat/put operations.
+
+    Identical to test_001 but uses broadcast forwarding strategy.
+    """
+    info('Starting forwarder on nodes\n')
+    AppManager(ndn, ndn.net.hosts, NDNd_FW, network=network)
+
+    dv_util.setup(ndn, network=network)
+    dv_util.converge(ndn.net.hosts, network=network)
+
+    info('Testing file transfer\n')
+    test_file = '/tmp/test.bin'
+    os.system(f'dd if=/dev/urandom of={test_file} bs=64K count=1')
+
+    sample_size = min(8, len(ndn.net.hosts)-1)
+    put_nodes = random.sample(ndn.net.hosts, sample_size)
+    cat_nodes = random.sample(ndn.net.hosts, sample_size)
+    cat_requests = [(cat_node, random.choice(put_nodes)) for cat_node in cat_nodes]
+    put_prefixes = {f"{network}/{node.name}/test" for node in put_nodes}
+    control_prefixes = {
+        f"{network}/32=DV/32=PES/32=svs",
+        f"/localhop{network}/32=DV/32=ADS/32=PSV",
+    }
+
+
+    for node in put_nodes:
+        prefix = f"{network}/{node.name}/test"
+        cmd = f'ndnd put --expose "{prefix}" < {test_file} &'
+        info(f'{node.name} {cmd}\n')
+        node.cmd(cmd)
+
+    # New pipeline requires PET propagation before Interests can be forwarded.
+    expected = {node: set(put_prefixes) for node in ndn.net.hosts}
+    dv_util.wait_prefix_pet_ready(expected, deadline=180)
+
+    # Prefix traffic should remain PET-driven; app prefixes must not be injected into FIB.
+    for node in ndn.net.hosts:
+        fib = node.cmd('ndnd fw fib-list')
+        for prefix in put_prefixes:
+            if prefix in fib:
+                raise Exception(f'App prefix {prefix} unexpectedly present in FIB on {node.name}')
+        for prefix in control_prefixes:
+            if prefix in fib:
+                raise Exception(f'Control prefix {prefix} unexpectedly present in FIB on {node.name}')
+        if f'{network}/32=DV/32=PES/' in fib:
+            raise Exception(f'Router-specific PES entries unexpectedly present in FIB on {node.name}')
+        if f'/localhop{network}/' in fib and '/32=DV' in fib:
+            raise Exception(f'Router-specific localhop DV entries unexpectedly present in FIB on {node.name}')
+
+    # Set default forwarding strategy to broadcast for the cat interest packets
+    for node in ndn.net.hosts:
+        node.cmd("ndnd fw strategy-set prefix=/ strategy=/localhost/nfd/strategy/broadcast/v=1")
+
+    for node, put_node in cat_requests:
+        cmd = f'ndnd cat "{network}/{put_node.name}/test" > recv.test.bin 2> cat.log'
+        info(f'{node.name} {cmd}\n')
+        node.cmd(cmd)
+        if node.cmd(f'diff {test_file} recv.test.bin').strip():
+            info(node.cmd(f'cat cat.log'))
+            raise Exception(f'Test file contents do not match on {node.name}')

e2e/test_006.py

@@ -0,0 +1,73 @@
+import random
+import os
+
+from mininet.log import info
+from minindn.minindn import Minindn
+from minindn.apps.app_manager import AppManager
+
+from fw import NDNd_FW
+import dv_util
+
+def scenario(ndn: Minindn, network='/minindn'):
+    """
+    Simple file transfer scenario with NDNd forwarder.
+    This tests routing convergence and cat/put operations.
+
+    Identical to test_001 but uses replicast forwarding strategy.
+    """
+    info('Starting forwarder on nodes\n')
+    AppManager(ndn, ndn.net.hosts, NDNd_FW, network=network)
+
+    dv_util.setup(ndn, network=network)
+    dv_util.converge(ndn.net.hosts, network=network)
+
+    info('Testing file transfer\n')
+    test_file = '/tmp/test.bin'
+    os.system(f'dd if=/dev/urandom of={test_file} bs=64K count=1')
+
+    sample_size = min(8, len(ndn.net.hosts)-1)
+    put_nodes = random.sample(ndn.net.hosts, sample_size)
+    cat_nodes = random.sample(ndn.net.hosts, sample_size)
+    cat_requests = [(cat_node, random.choice(put_nodes)) for cat_node in cat_nodes]
+    put_prefixes = {f"{network}/{node.name}/test" for node in put_nodes}
+    control_prefixes = {
+        f"{network}/32=DV/32=PES/32=svs",
+        f"/localhop{network}/32=DV/32=ADS/32=PSV",
+    }
+
+
+    for node in put_nodes:
+        prefix = f"{network}/{node.name}/test"
+        cmd = f'ndnd put --expose "{prefix}" < {test_file} &'
+        info(f'{node.name} {cmd}\n')
+        node.cmd(cmd)
+
+    # New pipeline requires PET propagation before Interests can be forwarded.
+    expected = {node: set(put_prefixes) for node in ndn.net.hosts}
+    dv_util.wait_prefix_pet_ready(expected, deadline=180)
+
+    # Prefix traffic should remain PET-driven; app prefixes must not be injected into FIB.
+    for node in ndn.net.hosts:
+        fib = node.cmd('ndnd fw fib-list')
+        for prefix in put_prefixes:
+            if prefix in fib:
+                raise Exception(f'App prefix {prefix} unexpectedly present in FIB on {node.name}')
+        for prefix in control_prefixes:
+            if prefix in fib:
+                raise Exception(f'Control prefix {prefix} unexpectedly present in FIB on {node.name}')
+        if f'{network}/32=DV/32=PES/' in fib:
+            raise Exception(f'Router-specific PES entries unexpectedly present in FIB on {node.name}')
+        if f'/localhop{network}/' in fib and '/32=DV' in fib:
+            raise Exception(f'Router-specific localhop DV entries unexpectedly present in FIB on {node.name}')
+
+    # Set default forwarding strategy to replicast for the cat interest packets
+    for node in ndn.net.hosts:
+        node.cmd("ndnd fw strategy-set prefix=/ strategy=/localhost/nfd/strategy/replicast/v=1")
+
+    for node, put_node in cat_requests:
+        cmd = f'ndnd cat "{network}/{put_node.name}/test" > recv.test.bin 2> cat.log'
+        info(f'{node.name} {cmd}\n')
+        node.cmd(cmd)
+        if node.cmd(f'diff {test_file} recv.test.bin').strip():
+            info(node.cmd(f'cat cat.log'))
+            raise Exception(f'Test file contents do not match on {node.name}')

fw/defn/name.go

@@ -11,7 +11,9 @@ var NON_LOCAL_PREFIX = enc.Name{enc.LOCALHOP, enc.NewGenericComponent("nfd")}
 // Prefix for all stratgies
 var STRATEGY_PREFIX = LOCAL_PREFIX.Append(enc.NewGenericComponent("strategy"))
 
-// Default forwarding strategy name
-var DEFAULT_STRATEGY = STRATEGY_PREFIX.
+var BEST_ROUTE_STRATEGY = STRATEGY_PREFIX.
 	Append(enc.NewGenericComponent("best-route")).
 	Append(enc.NewVersionComponent(1))
+
+// Default forwarding strategy name
+var DEFAULT_STRATEGY = BEST_ROUTE_STRATEGY

fw/fw/bestroute.go

@@ -73,7 +73,8 @@ func (s *BestRoute) AfterReceiveInterest(
 		return
 	}
 
-	// Sort nexthops by cost and send to best-possible nexthop
+	// sort nexthops / nextER by cost and send to best-possible nexthop
+	sort.Slice(nextER, func(i, j int) bool { return nexthops[i].Cost < nexthops[j].Cost })
 	sort.Slice(nexthops, func(i, j int) bool { return nexthops[i].Cost < nexthops[j].Cost })
 
 	now := time.Now()

fw/fw/broadcast.go

@@ -0,0 +1,87 @@
+/**
+ * Broadcast forwarding strategy.
+ *
+ * Using this strategy, the interest packet will be forwarded to all nexthops.
+ */
+
+package fw
+
+import (
+	"github.com/named-data/ndnd/fw/core"
+	"github.com/named-data/ndnd/fw/defn"
+	"github.com/named-data/ndnd/fw/table"
+	enc "github.com/named-data/ndnd/std/encoding"
+)
+
+// The strategy to forward interests to all nexthops.
+type Broadcast struct {
+	StrategyBase
+}
+
+// (AI GENERATED DESCRIPTION): Registers the Broadcast strategy with version 1 in the strategy registry by appending its constructor to the init list.
+func init() {
+	strategyInit = append(strategyInit, func() Strategy { return &Broadcast{} })
+	StrategyVersions["broadcast"] = []uint64{1}
+}
+
+// (AI GENERATED DESCRIPTION): Initializes the *Broadcast strategy by setting up its base with the name “broadcast” and priority 1 on the provided forwarding thread.
+func (s *Broadcast) Instantiate(fwThread *Thread) {
+	s.NewStrategyBase(fwThread, "broadcast", 1)
+}
+
+// (AI GENERATED DESCRIPTION): Sends a cached Data packet (retrieved from the Content Store) back to the requester via the specified PIT entry, using the requesting face and indicating the Content Store as the data source.
+func (s *Broadcast) AfterContentStoreHit(
+	packet *defn.Pkt,
+	pitEntry table.PitEntry,
+	inFace uint64,
+) {
+	core.Log.Trace(s, "AfterContentStoreHit", "name", packet.Name, "faceid", inFace)
+	s.SendData(packet, pitEntry, inFace, 0) // 0 indicates ContentStore is source
+}
+
+// (AI GENERATED DESCRIPTION): Forwards a received Data packet to every face recorded in the PIT entry, logging each forwarding step and invoking SendData for each destination.
+func (s *Broadcast) AfterReceiveData(
+	packet *defn.Pkt,
+	pitEntry table.PitEntry,
+	inFace uint64,
+) {
+	core.Log.Trace(s, "AfterReceiveData", "name", packet.Name, "inrecords", len(pitEntry.InRecords()))
+	for faceID := range pitEntry.InRecords() {
+		core.Log.Trace(s, "Forwarding Data", "name", packet.Name, "faceid", faceID)
+		s.SendData(packet, pitEntry, faceID, inFace)
+	}
+}
+
+// Forwards an incoming Interest to all next-hops
+func (s *Broadcast) AfterReceiveInterest(
+	packet *defn.Pkt,
+	pitEntry table.PitEntry,
+	inFace uint64,
+	nexthops []*table.FibNextHopEntry,
+	nextER []enc.Name,
+) {
+	if len(nexthops) == 0 {
+		core.Log.Debug(s, "No nexthop found - DROP", "name", packet.Name)
+		return
+	}
+
+	successfulForward := false
+
+	for _, nh := range nexthops {
+		if sent := s.SendInterest(packet, pitEntry, nh.Nexthop, inFace); sent {
+			core.Log.Trace(s, "Forwarded Interest", "name", packet.Name, "faceid", nh.Nexthop)
+			successfulForward = true
+		} else {
+			core.Log.Trace(s, "Error forwarding interest", "name", packet.Name, "faceid", nh.Nexthop)
+		}
+	}
+
+	if !successfulForward {
+		core.Log.Debug(s, "No usable nexthop for Interest - DROP", "name", packet.Name)
+	}
+}
+
+// (AI GENERATED DESCRIPTION): No‑op hook invoked before satisfying an Interest in the Broadcast strategy – it performs no action.
+func (s *Broadcast) BeforeSatisfyInterest(pitEntry table.PitEntry, inFace uint64) {
+	// This does nothing in Broadcast
+}