Merge pull request #179 from r2dev2/r2dev2/three-pipeline-forwarding

[fw] refactor incoming Interest processing

Author: tianyuan129

.github/workflows/test.yml

@@ -89,3 +89,10 @@ jobs:
 
       - name: Run e2e tests
         run: make e2e
+
+      - name: Upload minindn logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: minindn-logs
+          path: /tmp/minindn

Makefile

@@ -9,6 +9,8 @@ ndnd: clean
 	CGO_ENABLED=0 go build -o ndnd \
 		-ldflags "-X '${STD_PACKAGE}/utils.NDNdVersion=${VERSION}'" \
 		cmd/ndnd/main.go
+	mkdir -p .bin/
+	cp ndnd .bin/
 
 examples: .bin/alo-latest
 

dv/config/config.go

@@ -32,11 +32,6 @@ var BroadcastStrategy = enc.LOCALHOST.
 	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/mgmt.go

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

dv/dv/prefix.go

@@ -425,9 +425,8 @@ func (pfx *PrefixModule) applyPetOps(ops []petEgressOp) {
 		}
 		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))
+				args.Multicast = true
 			}
 		}
 

dv/dv/router.go

@@ -8,6 +8,7 @@ import (
 	"github.com/named-data/ndnd/dv/config"
 	"github.com/named-data/ndnd/dv/nfdc"
 	"github.com/named-data/ndnd/dv/table"
+	"github.com/named-data/ndnd/fw/defn"
 	enc "github.com/named-data/ndnd/std/encoding"
 	"github.com/named-data/ndnd/std/log"
 	"github.com/named-data/ndnd/std/ndn"
@@ -285,64 +286,51 @@ func (dv *Router) register() (err error) {
 	}
 
 	for _, prefix := range pfxs {
-		dv.nfdc.Exec(nfdc.NfdMgmtCmd{
-			Module: "pet",
-			Cmd:    "add-nexthop",
-			Args: &mgmt.ControlArgs{
-				Name: prefix,
-			},
-			Retries: -1,
+		dv.execMgmtRetry("pet", "add-nexthop", &mgmt.ControlArgs{
+			Name: prefix,
 		})
 	}
 	// Allow outgoing local-prefix-sync Interests to use two-phase forwarding.
 	// Incoming Interests still terminate locally on the same prefix.
-	dv.nfdc.Exec(nfdc.NfdMgmtCmd{
-		Module: "pet",
-		Cmd:    "add-egress",
-		Args: &mgmt.ControlArgs{
-			Name:   dv.pfx.SyncPrefix(),
-			Egress: &mgmt.EgressRecord{Name: neighborsPrefix.Clone()},
-		},
-		Retries: -1,
+	dv.execMgmtRetry("pet", "add-egress", &mgmt.ControlArgs{
+		Name:      dv.pfx.SyncPrefix(),
+		Egress:    &mgmt.EgressRecord{Name: neighborsPrefix.Clone()},
+		Multicast: true,
 	})
 	// Set Advertisement Sync to localhop neighbors
-	dv.nfdc.Exec(nfdc.NfdMgmtCmd{
-		Module: "pet",
-		Cmd:    "add-egress",
-		Args: &mgmt.ControlArgs{
-			Name:   dv.config.AdvertisementSyncPrefix(),
-			Egress: &mgmt.EgressRecord{Name: neighborsPrefix.Clone()},
-		},
-		Retries: -1,
-	})
-	// 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.BroadcastStrategy,
-			},
-		},
-		Retries: -1,
-	})
-	dv.nfdc.Exec(nfdc.NfdMgmtCmd{
-		Module: "strategy-choice",
-		Cmd:    "set",
-		Args: &mgmt.ControlArgs{
-			Name: dv.pfx.SyncPrefix(),
-			Strategy: &mgmt.Strategy{
-				Name: config.BroadcastStrategy,
-			},
-		},
-		Retries: -1,
+	dv.execMgmtRetry("pet", "add-egress", &mgmt.ControlArgs{
+		Name:      dv.config.AdvertisementSyncPrefix(),
+		Egress:    &mgmt.EgressRecord{Name: neighborsPrefix.Clone()},
+		Multicast: true,
 	})
 
+	// Force multicast strategy for sync prefixes to broadcast.
+	broadcastPrefixes := []enc.Name{
+		dv.pfx.SyncPrefix(),
+		dv.config.AdvertisementSyncPrefix(),
+	}
+	for _, prefix := range broadcastPrefixes {
+		dv.execMgmtRetry("multicast-strategy-choice", "set", &mgmt.ControlArgs{
+			Name:     prefix,
+			Strategy: &mgmt.Strategy{Name: defn.BROADCAST_STRATEGY},
+		})
+	}
+
 	return nil
 }
 
+func (dv *Router) execMgmtRetry(module, cmd string, args *mgmt.ControlArgs) {
+	var err error
+	for i := 0; ; i++ {
+		if _, err = dv.engine.ExecMgmtCmd(module, cmd, args); err == nil {
+			break
+		}
+		log.Error(dv, "Forwarder command failed", "err", err, "attempt", i,
+			"module", module, "cmd", cmd, "args", args)
+		time.Sleep(100 * time.Millisecond)
+	}
+}
+
 // createFaces creates faces to all neighbors.
 func (dv *Router) createFaces() {
 	neighborsPrefix := enc.LOCALHOP.Append(enc.NewGenericComponent("neighbors"))
@@ -370,15 +358,10 @@ func (dv *Router) createFaces() {
 		dv.mutex.Unlock()
 
 		// Add neighbor to localhop neighbors
-		dv.nfdc.Exec(nfdc.NfdMgmtCmd{
-			Module: "fib",
-			Cmd:    "add-nexthop",
-			Args: &mgmt.ControlArgs{
-				Name:   neighborsPrefix.Clone(),
-				Cost:   optional.Some(uint64(1)),
-				FaceId: optional.Some(faceId),
-			},
-			Retries: 3,
+		dv.execMgmtRetry("fib", "add-nexthop", &mgmt.ControlArgs{
+			Name:   neighborsPrefix.Clone(),
+			Cost:   optional.Some(uint64(1)),
+			FaceId: optional.Some(faceId),
 		})
 	}
 }

dv/dv/table_algo.go

@@ -3,7 +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"
+	"github.com/named-data/ndnd/fw/bier"
 	enc "github.com/named-data/ndnd/std/encoding"
 	"github.com/named-data/ndnd/std/log"
 )
@@ -138,7 +138,7 @@ func (dv *Router) updateFib() {
 
 	// Rebuild the BIFT whenever the FIB changes so BIER forwarding paths
 	// are always consistent with the routing table.
-	if fw.IsBierEnabled() {
-		fw.Bift.BuildFromFib()
+	if bier.IsBierEnabled() {
+		bier.Bift.BuildFromFib()
 	}
 }

e2e/dv_util.py

@@ -53,7 +53,7 @@ def is_converged(nodes: list[Node], network=DEFAULT_NETWORK, use_nfdc=False) ->
             # We don't support that.
             routes = node.cmd('nfdc route list')
         else:
-            routes = node.cmd('ndnd fw route-list')
+            routes = node.cmd('ndnd fw route-list 2>&1')
         for other in nodes:
             if other == node:
                 continue

e2e/test_001.py

@@ -63,16 +63,6 @@ def scenario(ndn: Minindn, network='/minindn'):
         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 = [
-            # Localhop SVS sync interests should use broadcast (#174)
-            "prefix=/minindn/32=DV/32=PES/32=svs strategy=/localhost/nfd/strategy/broadcast/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'

e2e/test_007.py

@@ -81,6 +81,7 @@ def scenario(ndn: Minindn, network='/minindn'):
     producer_prefixes = {sync_prefix, f'{network}/svs/{producer.name}'}
     dv_util.wait_prefix_pet_ready({node: producer_prefixes for node in consumers}, deadline=60)
 
+    beg = time.time()
     # Step 5: Wait for publish → BIER sync interest propagation → unicast data fetch.
     # 20s delay + 30s SVS sync cycle + 20s fetch buffer = 70s total.
     info(f'Waiting for message propagation from {producer.name} (70s)...\n')
@@ -90,6 +91,7 @@ def scenario(ndn: Minindn, network='/minindn'):
     for node in chat_nodes:
         node.cmd("pkill -f 'ndnd svs-chat' 2>/dev/null; true")
     time.sleep(1)  # let log buffers flush
+    print(time.time() - beg, "seconds elapsed")
 
     failures = []
     for consumer in consumers: