fw: implement cs/erase management command

fw/dispatch/fw.go

@@ -9,6 +9,7 @@ package dispatch
 
 import (
 	"github.com/named-data/ndnd/fw/defn"
+	enc "github.com/named-data/ndnd/std/encoding"
 )
 
 // FWThread provides an interface that forwarding threads can satisfy
@@ -19,6 +20,8 @@ type FWThread interface {
 	QueueData(packet *defn.Pkt)
 	QueueInterest(packet *defn.Pkt)
 
+	EraseCsDataUnderPrefix(name enc.Name, limit int) (int, bool)
+
 	Counters() defn.FWThreadCounters
 }
 

fw/fw/thread.go

@@ -68,6 +68,7 @@ type Thread struct {
 	deadNonceList *table.DeadNonceList
 	shouldQuit    chan interface{}
 	HasQuit       chan interface{}
+	csErase       chan csEraseRequest
 
 	// Counters
 	nInInterests          atomic.Uint64
@@ -80,6 +81,20 @@ type Thread struct {
 	nCsMisses             atomic.Uint64
 }
 
+// csEraseRequest is a request from the management thread to erase Content
+// Store entries under a name prefix. The reply channel is buffered so the
+// forwarding goroutine never blocks on it.
+type csEraseRequest struct {
+	name  enc.Name
+	limit int
+	reply chan csEraseResult
+}
+
+type csEraseResult struct {
+	nErased int
+	more    bool
+}
+
 // NewThread creates a new forwarding thread
 func NewThread(id int) *Thread {
 	t := new(Thread)
@@ -89,6 +104,7 @@ func NewThread(id int) *Thread {
 	t.strategies = InstantiateStrategies(t)
 	t.deadNonceList = table.NewDeadNonceList()
 	t.shouldQuit = make(chan interface{}, 1)
+	t.csErase = make(chan csEraseRequest)
 	t.HasQuit = make(chan interface{})
 	return t
 }
@@ -143,6 +159,9 @@ func (t *Thread) Run() {
 			t.deadNonceList.RemoveExpiredEntries()
 		case <-t.pitCS.UpdateTicker():
 			t.pitCS.Update()
+		case req := <-t.csErase:
+			n, more := t.pitCS.EraseCsDataUnderPrefix(req.name, req.limit)
+			req.reply <- csEraseResult{nErased: n, more: more}
 		case <-t.shouldQuit:
 			continue
 		}
@@ -172,6 +191,17 @@ func (t *Thread) QueueData(data *defn.Pkt) {
 	}
 }
 
+// EraseCsDataUnderPrefix erases up to limit Content Store entries under the
+// given prefix in this thread's Content Store. It blocks until the forwarding
+// goroutine has serviced the request, and reports whether further matching
+// entries remain.
+func (t *Thread) EraseCsDataUnderPrefix(name enc.Name, limit int) (int, bool) {
+	reply := make(chan csEraseResult, 1)
+	t.csErase <- csEraseRequest{name: name, limit: limit, reply: reply}
+	result := <-reply
+	return result.nErased, result.more
+}
+
 // (AI GENERATED DESCRIPTION): Processes an incoming Interest packet: verifies its validity, enforces hop limits and scope, checks for nonces and dead‑nonce loops, updates the PIT and content store, selects and filters next‑hops via the FIB, and forwards the Interest according to the chosen forwarding strategy.
 func (t *Thread) processIncomingInterest(packet *defn.Pkt) {
 	interest := packet.L3.Interest

fw/mgmt/cs.go

@@ -8,6 +8,8 @@
 package mgmt
 
 import (
+	"math"
+
 	"github.com/named-data/ndnd/fw/core"
 	"github.com/named-data/ndnd/fw/dispatch"
 	"github.com/named-data/ndnd/fw/fw"
@@ -51,8 +53,7 @@ func (c *ContentStoreModule) handleIncomingInterest(interest *Interest) {
 	case "config":
 		c.config(interest)
 	case "erase":
-		// TODO
-		//c.erase(interest)
+		c.erase(interest)
 	case "info":
 		c.info(interest)
 	default:
@@ -111,6 +112,65 @@ func (c *ContentStoreModule) config(interest *Interest) {
 	})
 }
 
+// eraseLimit mirrors NFD's ERASE_LIMIT: a single cs/erase command never
+// erases more than this many entries, and the client is expected to repeat the
+// command for the remainder.
+const csEraseLimit = 256
+
+// erase handles a cs/erase command Interest: it erases Content Store entries
+// under the prefix in ControlParameters.Name on every forwarding thread and
+// replies with the number of entries erased.
+func (c *ContentStoreModule) erase(interest *Interest) {
+	if len(interest.Name()) < len(LOCAL_PREFIX)+3 {
+		// Name not long enough to contain ControlParameters
+		core.Log.Warn(c, "Missing ControlParameters", "name", interest.Name())
+		c.manager.sendCtrlResp(interest, 400, "ControlParameters is incorrect", nil)
+		return
+	}
+
+	params := decodeControlParameters(c, interest)
+	if params == nil {
+		c.manager.sendCtrlResp(interest, 400, "ControlParameters is incorrect", nil)
+		return
+	}
+
+	if len(params.Name) == 0 {
+		core.Log.Warn(c, "Missing Prefix in ControlParameters", "name", interest.Name())
+		c.manager.sendCtrlResp(interest, 400, "ControlParameters is incorrect", nil)
+		return
+	}
+
+	requested := uint64(math.MaxUint64)
+	if count, ok := params.Count.Get(); ok {
+		requested = count
+	}
+	remaining := min(requested, csEraseLimit)
+
+	nErased := uint64(0)
+	more := false
+	for threadID := 0; threadID < fw.CfgNumThreads(); threadID++ {
+		thread := dispatch.GetFWThread(threadID)
+		if thread == nil {
+			continue
+		}
+		n, threadHasMore := thread.EraseCsDataUnderPrefix(params.Name, int(remaining))
+		nErased += uint64(n)
+		more = more || threadHasMore
+		remaining -= uint64(n)
+	}
+
+	// NFD signals "more entries remain" by setting Capacity to ERASE_LIMIT when
+	// the command hit the limit and the client asked for more than that.
+	args := &mgmt.ControlArgs{
+		Name:  params.Name,
+		Count: optional.Some(nErased),
+	}
+	if nErased == csEraseLimit && requested > csEraseLimit && more {
+		args.Capacity = optional.Some(uint64(csEraseLimit))
+	}
+	c.manager.sendCtrlResp(interest, 200, "OK", args)
+}
+
 // (AI GENERATED DESCRIPTION): Collects content‑store statistics from all threads and replies to the Interest with a status dataset containing the CS capacity, flags, entry count, hit and miss counts.
 func (c *ContentStoreModule) info(interest *Interest) {
 	if len(interest.Name()) > len(LOCAL_PREFIX)+2 {

fw/table/pit-cs-tree.go

@@ -407,6 +407,47 @@ func (p *PitCsTree) eraseCsDataFromReplacementStrategy(index uint64) {
 	}
 }
 
+// EraseCsDataUnderPrefix erases up to limit Content Store entries whose names
+// fall under the given prefix, and reports whether further matching entries
+// remain. With limit < 1 nothing is erased and only the remain flag is
+// reported. It must be called from the goroutine that owns the table.
+func (p *PitCsTree) EraseCsDataUnderPrefix(name enc.Name, limit int) (int, bool) {
+	node := p.root.findExactMatchEntryEnc(name)
+	if node == nil {
+		return 0, false
+	}
+
+	// Collect one node past the limit so the caller can tell whether another
+	// erase command would still find entries under this prefix.
+	var targets []*pitCsTreeNode
+	var collect func(n *pitCsTreeNode)
+	collect = func(n *pitCsTreeNode) {
+		if len(targets) > limit {
+			return
+		}
+		if n.csEntry != nil {
+			targets = append(targets, n)
+		}
+		for _, child := range n.children {
+			collect(child)
+		}
+	}
+	collect(node)
+
+	more := len(targets) > limit
+	if more {
+		targets = targets[:limit]
+	}
+	for _, target := range targets {
+		entry := target.csEntry
+		p.csReplacement.BeforeErase(entry.index, entry.wire)
+		target.csEntry = nil
+		delete(p.csMap, entry.index)
+		p.nCsEntries.Add(-1)
+	}
+	return len(targets), more
+}
+
 // Given a pitCsTreeNode that is the longest prefix match of an interest, look for any
 // CS data rechable from this pitCsTreeNode. This function must be called only after
 // the interest as far as possible with the nodes components in the PitCSTree.

fw/table/pit-cs-tree_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"math/rand"
 	"sort"
+	"strconv"
 	"testing"
 	"time"
 
@@ -120,6 +121,69 @@ func TestIsCsServing(t *testing.T) {
 }
 
 // (AI GENERATED DESCRIPTION): Unit test that verifies PitCS.InsertInterest correctly creates or updates PIT entries, detects duplicate nonces, preserves entry state, and supports prefix relationships between interests.
+func TestEraseCsDataUnderPrefix(t *testing.T) {
+	setReplacementPolicy("lru")
+	CfgSetCsCapacity(1024)
+	pitCS := NewPitCS(func(PitEntry) {})
+
+	// Empty tree: nothing erased, nothing remains
+	name, _ := enc.NameFromStr("/ndn")
+	n, more := pitCS.EraseCsDataUnderPrefix(name, 10)
+	assert.Equal(t, n, 0)
+	assert.False(t, more)
+
+	// Insert data under /ndn and elsewhere
+	name1, _ := enc.NameFromStr("/ndn/a")
+	name2, _ := enc.NameFromStr("/ndn/b/c")
+	name3, _ := enc.NameFromStr("/other")
+	pitCS.InsertData(makeData(name1), VALID_DATA_1)
+	pitCS.InsertData(makeData(name2), VALID_DATA_2)
+	pitCS.InsertData(makeData(name3), VALID_DATA_1)
+	assert.Equal(t, pitCS.CsSize(), 3)
+
+	// Detect-only mode erases nothing but reports matches
+	n, more = pitCS.EraseCsDataUnderPrefix(name, 0)
+	assert.Equal(t, n, 0)
+	assert.True(t, more)
+	assert.Equal(t, pitCS.CsSize(), 3)
+
+	// Limit below the match count: erases up to the limit, reports more
+	n, more = pitCS.EraseCsDataUnderPrefix(name, 1)
+	assert.Equal(t, n, 1)
+	assert.True(t, more)
+	assert.Equal(t, pitCS.CsSize(), 2)
+
+	// Erased entries no longer satisfy interests
+	interest1 := makeInterest(name1)
+	interest1.CanBePrefixV = false
+	interest2 := makeInterest(name2)
+	interest2.CanBePrefixV = false
+	found := pitCS.FindMatchingDataFromCS(interest1) != nil ||
+		pitCS.FindMatchingDataFromCS(interest2) != nil
+	assert.True(t, found) // exactly one remains under /ndn
+
+	// Erase the rest under the prefix; entries outside it are untouched
+	n, more = pitCS.EraseCsDataUnderPrefix(name, 10)
+	assert.Equal(t, n, 1)
+	assert.False(t, more)
+	assert.Equal(t, pitCS.CsSize(), 1)
+
+	interest3 := makeInterest(name3)
+	interest3.CanBePrefixV = false
+	assert.NotNil(t, pitCS.FindMatchingDataFromCS(interest3))
+
+	// Erased entries are also removed from the replacement strategy, so the
+	// next insert after refilling to capacity must not consult stale entries
+	setReplacementPolicy("lru")
+	for i := 0; i < 10; i++ {
+		dname, _ := enc.NameFromStr("/fill/" + strconv.Itoa(i))
+		pitCS.InsertData(makeData(dname), VALID_DATA_1)
+	}
+	n, _ = pitCS.EraseCsDataUnderPrefix(name3, 10)
+	assert.Equal(t, n, 1)
+	assert.Equal(t, pitCS.CsSize(), 10)
+}
+
 func TestInsertInterest(t *testing.T) {
 	setReplacementPolicy("lru")
 

fw/table/pit-cs.go

@@ -32,6 +32,8 @@ type PitCsTable interface {
 	InsertData(data *defn.FwData, wire []byte)
 	// FindMatchingDataFromCS finds a matching Data in the CS.
 	FindMatchingDataFromCS(interest *defn.FwInterest) CsEntry
+	// EraseCsDataUnderPrefix erases up to limit Data under the given prefix from the CS.
+	EraseCsDataUnderPrefix(name enc.Name, limit int) (int, bool)
 	// CsSize returns the number of entries in the CS.
 	CsSize() int
 	// IsCsAdmitting returns whether the CS is admitting new entries.