diff --git a/drsm/chunk.go b/drsm/chunk.go index 3a8eb94..3465d92 100644 --- a/drsm/chunk.go +++ b/drsm/chunk.go @@ -66,6 +66,44 @@ func (d *Drsm) GetNewChunk() (*chunk, error) { return c, nil } +// appendScanIds seeds the ids still to be scanned. FreeIds, ScanIds and AllocIds are shared +// with AllocateInt32ID and ReleaseInt32ID, which hold mutex, so the scan goroutine started by +// claimChunk has to hold it too. The slice is built first so the lock covers only the append. +func (c *chunk) appendScanIds(n int32) { + ids := make([]int32, 0, n) + for i := int32(0); i < n; i++ { + ids = append(ids, i) + } + mutex.Lock() + defer mutex.Unlock() + c.ScanIds = append(c.ScanIds, ids...) +} + +// nextScanId takes the next id to scan, reporting false once every id has been scanned. +func (c *chunk) nextScanId() (int32, bool) { + mutex.Lock() + defer mutex.Unlock() + if len(c.ScanIds) == 0 { + return 0, false + } + id := c.ScanIds[len(c.ScanIds)-1] + c.ScanIds = c.ScanIds[:len(c.ScanIds)-1] + return id, true +} + +// recordScanResult files a scanned id as free or as already in use. The caller invokes +// resourceValidCb outside the lock on purpose: it is supplied by the NF and may re-enter +// drsm, which would deadlock on a non-reentrant mutex. +func (c *chunk) recordScanResult(id int32, free bool) { + mutex.Lock() + defer mutex.Unlock() + if free { + c.FreeIds = append(c.FreeIds, id) + } else { + c.AllocIds[id] = true // Id is in use + } +} + func (c *chunk) AllocateIntID() (int32, error) { if len(c.FreeIds) == 0 { err := fmt.Errorf("freeIds in chunk 0") diff --git a/drsm/scan.go b/drsm/scan.go index d86dc39..ab21207 100644 --- a/drsm/scan.go +++ b/drsm/scan.go @@ -21,10 +21,7 @@ func (c *chunk) scanChunk(d *Drsm) { } c.State = Scanning d.scanChunks[c.Id] = c - var i int32 - for i = 0; i < 1000; i++ { - c.ScanIds = append(c.ScanIds, i) - } + c.appendScanIds(1000) ticker := time.NewTicker(5000 * time.Millisecond) defer ticker.Stop() @@ -35,16 +32,10 @@ func (c *chunk) scanChunk(d *Drsm) { // TODO : find candidate and then scan that Id. // once all Ids are scanned then we can start using this block if c.resourceValidCb != nil { - if len(c.ScanIds) != 0 { - id := c.ScanIds[len(c.ScanIds)-1] - c.ScanIds = c.ScanIds[:len(c.ScanIds)-1] + if id, ok := c.nextScanId(); ok { rid := c.Id<<10 | id res := c.resourceValidCb(rid) - if res { - c.FreeIds = append(c.FreeIds, id) - } else { - c.AllocIds[id] = true // Id is in use - } + c.recordScanResult(id, res) } else { // mark as owned. and remove from scan list and add to local table c.State = Owned diff --git a/drsm/scan_test.go b/drsm/scan_test.go new file mode 100644 index 0000000..f08fe44 --- /dev/null +++ b/drsm/scan_test.go @@ -0,0 +1,109 @@ +// SPDX-FileCopyrightText: 2026 Forsway Scandinavia AB +// +// SPDX-License-Identifier: Apache-2.0 + +package drsm + +import ( + "sync" + "testing" +) + +const ( + scanSeed = 1024 + scanRounds = 2000 +) + +func newScanTestChunk(id int32) *chunk { + return &chunk{Id: id, State: Scanning, AllocIds: make(map[int32]bool)} +} + +// TestScanFieldsConcurrentAccess drives the scan goroutine's handling of ScanIds, FreeIds and +// AllocIds against the two API paths that touch the same fields. api.go calls AllocateIntID +// and ReleaseIntID with mutex held, so the test does the same, which is what makes the scan +// goroutine's accesses the unsynchronised side. +// +// Releasing an id that belongs to a chunk still being scanned is a supported path: +// ReleaseInt32ID looks the chunk up in scanChunks precisely so that can happen, and +// ReleaseIntID then walks ScanIds while the scan goroutine is popping from it. +func TestScanFieldsConcurrentAccess(t *testing.T) { + c := newScanTestChunk(1) + c.appendScanIds(scanSeed) + + var ( + seenMu sync.Mutex + seen = make(map[int32]bool) + ) + + var wg sync.WaitGroup + wg.Add(3) + + // The scan goroutine: pop an id, consult the callback outside the lock, record the result. + go func() { + defer wg.Done() + for { + id, ok := c.nextScanId() + if !ok { + return + } + seenMu.Lock() + if seen[id] { + t.Errorf("id %d was handed out twice by nextScanId", id) + seenMu.Unlock() + return + } + seen[id] = true + seenMu.Unlock() + + c.recordScanResult(id, id%3 != 0) + } + }() + + // ReleaseInt32ID's inner call. With State == Scanning this also walks ScanIds. + go func() { + defer wg.Done() + for i := range int32(scanRounds) { + mutex.Lock() + c.ReleaseIntID(i % scanSeed) + mutex.Unlock() + } + }() + + // AllocateInt32ID's inner call. + go func() { + defer wg.Done() + for range scanRounds { + mutex.Lock() + _, _ = c.AllocateIntID() + mutex.Unlock() + } + }() + + wg.Wait() + + // appendScanIds seeds distinct ids and both mutators only remove, so nothing may be left. + if _, ok := c.nextScanId(); ok { + t.Error("ScanIds should be drained once the scan goroutine has finished") + } +} + +// TestAppendScanIdsSeedsDistinctIds pins the seeding behaviour the drain assertion relies on. +func TestAppendScanIdsSeedsDistinctIds(t *testing.T) { + c := newScanTestChunk(2) + c.appendScanIds(8) + + got := make(map[int32]bool) + for { + id, ok := c.nextScanId() + if !ok { + break + } + if got[id] { + t.Fatalf("id %d seeded more than once", id) + } + got[id] = true + } + if len(got) != 8 { + t.Errorf("seeded %d ids, want 8", len(got)) + } +}