Harden cgroups snapshot cache concurrency

Author: ktsaou

.agents/sow/done/SOW-0025-20260627-cgroups-cache-concurrency-contract.md

@@ -1041,12 +1041,22 @@ static int run_lookup_bench(int duration_sec)
     }
 
     nipc_cgroups_cache_t cache;
-    memset(&cache, 0, sizeof(cache));
-    cache.items = items;
-    cache.item_count = 16;
-    cache.populated = true;
-    cache.systemd_enabled = 1;
-    cache.generation = 1;
+    nipc_client_config_t ccfg = {
+        .supported_profiles = NIPC_PROFILE_BASELINE,
+        .max_request_batch_items = 1,
+        .max_response_payload_bytes = RESPONSE_BUF_SIZE,
+        .auth_token = AUTH_TOKEN,
+    };
+    nipc_cgroups_cache_init(&cache, "/tmp", "bench_lookup", &ccfg);
+    if (!nipc_cgroups_cache_seed_for_tests(&cache, items, 16, 1, 1)) {
+        fprintf(stderr, "lookup: failed to seed cache\n");
+        for (int i = 0; i < 16; i++) {
+            free(items[i].name);
+            free(items[i].path);
+        }
+        nipc_cgroups_cache_close(&cache);
+        return 1;
+    }
 
     uint64_t lookups = 0;
     uint64_t hits = 0;
@@ -1057,12 +1067,16 @@ static int run_lookup_bench(int duration_sec)
 
     while (now_ns() < wall_end) {
         /* Cycle through all 16 items */
-        for (int i = 0; i < 16; i++) {
-            const nipc_cgroups_cache_item_t *found =
-                nipc_cgroups_cache_lookup(&cache, items[i].hash, items[i].name);
-            if (found)
-                hits++;
-            lookups++;
+        nipc_cgroups_cache_read_guard_t guard;
+        if (nipc_cgroups_cache_read_lock(&cache, &guard)) {
+            for (int i = 0; i < 16; i++) {
+                const nipc_cgroups_cache_item_view_t *found =
+                    nipc_cgroups_cache_get(&guard, items[i].hash, items[i].name);
+                if (found)
+                    hits++;
+                lookups++;
+            }
+            nipc_cgroups_cache_read_unlock(&guard);
         }
     }
 
@@ -1088,6 +1102,7 @@ static int run_lookup_bench(int duration_sec)
         free(items[i].name);
         free(items[i].path);
     }
+    nipc_cgroups_cache_close(&cache);
 
     return (hits == lookups) ? 0 : 1;
 }

bench/drivers/c/bench_posix.c

@@ -1453,12 +1453,22 @@ static int run_lookup_bench(int duration_sec)
     }
 
     nipc_cgroups_cache_t cache;
-    memset(&cache, 0, sizeof(cache));
-    cache.items = items;
-    cache.item_count = 16;
-    cache.populated = true;
-    cache.systemd_enabled = 1;
-    cache.generation = 1;
+    nipc_client_config_t ccfg = {
+        .supported_profiles = NIPC_PROFILE_BASELINE,
+        .max_request_batch_items = 1,
+        .max_response_payload_bytes = RESPONSE_BUF_SIZE,
+        .auth_token = AUTH_TOKEN,
+    };
+    nipc_cgroups_cache_init(&cache, ".", "bench_lookup", &ccfg);
+    if (!nipc_cgroups_cache_seed_for_tests(&cache, items, 16, 1, 1)) {
+        fprintf(stderr, "lookup: failed to seed cache\n");
+        for (int i = 0; i < 16; i++) {
+            free(items[i].name);
+            free(items[i].path);
+        }
+        nipc_cgroups_cache_close(&cache);
+        return 1;
+    }
 
     uint64_t lookups = 0;
     uint64_t hits = 0;
@@ -1468,12 +1478,16 @@ static int run_lookup_bench(int duration_sec)
     ULONGLONG tick_deadline = GetTickCount64() + (ULONGLONG)duration_sec * 1000;
 
     while (GetTickCount64() < tick_deadline) {
-        for (int i = 0; i < 16; i++) {
-            const nipc_cgroups_cache_item_t *found =
-                nipc_cgroups_cache_lookup(&cache, items[i].hash, items[i].name);
-            if (found)
-                hits++;
-            lookups++;
+        nipc_cgroups_cache_read_guard_t guard;
+        if (nipc_cgroups_cache_read_lock(&cache, &guard)) {
+            for (int i = 0; i < 16; i++) {
+                const nipc_cgroups_cache_item_view_t *found =
+                    nipc_cgroups_cache_get(&guard, items[i].hash, items[i].name);
+                if (found)
+                    hits++;
+                lookups++;
+            }
+            nipc_cgroups_cache_read_unlock(&guard);
         }
     }
 
@@ -1498,6 +1512,7 @@ static int run_lookup_bench(int duration_sec)
         free(items[i].name);
         free(items[i].path);
     }
+    nipc_cgroups_cache_close(&cache);
 
     return (hits == lookups) ? 0 : 1;
 }

bench/drivers/c/bench_windows.c

@@ -49,19 +49,6 @@ const (
 	lookupMethodBufBytesPerItem = 256
 )
 
-type cacheBucket struct {
-	index int
-	used  bool
-}
-
-func cacheHashName(name string) uint32 {
-	h := uint32(5381)
-	for i := 0; i < len(name); i++ {
-		h = ((h << 5) + h) + uint32(name[i])
-	}
-	return h
-}
-
 // ---------------------------------------------------------------------------
 //  Timing helpers
 // ---------------------------------------------------------------------------
@@ -1216,19 +1203,18 @@ func runLookupBench(durationSec int) int {
 		}
 	}
 
-	lookupIndex := make([]cacheBucket, 32)
-	mask, ok := u32FromNonNegativeInt(len(lookupIndex) - 1)
-	if !ok {
-		fmt.Fprintln(os.Stderr, "cache lookup index too large")
+	cache := rawsvc.NewCache("/tmp", "bench_lookup", posix.ClientConfig{
+		SupportedProfiles:       protocol.ProfileBaseline,
+		PreferredProfiles:       0,
+		MaxRequestBatchItems:    1,
+		MaxResponsePayloadBytes: responseBufSize,
+		MaxResponseBatchItems:   1,
+		AuthToken:               authToken,
+	})
+	if !cache.SeedForTests(cacheItems, 1, 1) {
+		fmt.Fprintln(os.Stderr, "cache seed failed")
 		return 1
 	}
-	for i := range cacheItems {
-		slot := (cacheItems[i].Hash ^ cacheHashName(cacheItems[i].Name)) & mask
-		for lookupIndex[slot].used {
-			slot = (slot + 1) & mask
-		}
-		lookupIndex[slot] = cacheBucket{index: i, used: true}
-	}
 
 	var lookups, hits uint64
 
@@ -1237,22 +1223,14 @@ func runLookupBench(durationSec int) int {
 	deadline := time.Duration(durationSec) * time.Second
 
 	for time.Since(wallStart) < deadline {
+		guard := cache.ReadLock()
 		for _, it := range items {
-			slot := (it.hash ^ cacheHashName(it.name)) & mask
-			found := false
-			for lookupIndex[slot].used {
-				bucketItem := &cacheItems[lookupIndex[slot].index]
-				if bucketItem.Hash == it.hash && bucketItem.Name == it.name {
-					found = true
-					break
-				}
-				slot = (slot + 1) & mask
-			}
-			if found {
+			if guard.Get(it.hash, it.name) != nil {
 				hits++
 			}
 			lookups++
 		}
+		guard.Unlock()
 	}
 
 	wallSec := time.Since(wallStart).Seconds()

bench/drivers/go/main.go

@@ -53,19 +53,6 @@ const (
 	clientShmAttachRetryTimeoutWin  = 5 * time.Second
 )
 
-type cacheBucketWin struct {
-	index int
-	used  bool
-}
-
-func cacheHashNameWin(name string) uint32 {
-	h := uint32(5381)
-	for i := 0; i < len(name); i++ {
-		h = ((h << 5) + h) + uint32(name[i])
-	}
-	return h
-}
-
 type xorshift32Win struct {
 	state uint32
 }
@@ -1840,14 +1827,17 @@ func runLookupBenchWin(durationSec int) int {
 		}
 	}
 
-	lookupIndex := make([]cacheBucketWin, 32)
-	mask := uint32(len(lookupIndex) - 1)
-	for i := range cacheItems {
-		slot := (cacheItems[i].Hash ^ cacheHashNameWin(cacheItems[i].Name)) & mask
-		for lookupIndex[slot].used {
-			slot = (slot + 1) & mask
-		}
-		lookupIndex[slot] = cacheBucketWin{index: i, used: true}
+	cache := rawsvc.NewCache(".", "bench_lookup", windows.ClientConfig{
+		SupportedProfiles:       protocol.ProfileBaseline,
+		PreferredProfiles:       0,
+		MaxRequestBatchItems:    1,
+		MaxResponsePayloadBytes: responseBufSizeWin,
+		MaxResponseBatchItems:   1,
+		AuthToken:               authTokenWin,
+	})
+	if !cache.SeedForTests(cacheItems, 1, 1) {
+		fmt.Fprintln(os.Stderr, "cache seed failed")
+		return 1
 	}
 
 	var lookups, hits uint64
@@ -1857,22 +1847,14 @@ func runLookupBenchWin(durationSec int) int {
 	tickDeadline := tickMSWin() + uint64(durationSec)*1000
 
 	for tickMSWin() < tickDeadline {
+		guard := cache.ReadLock()
 		for _, it := range items {
-			slot := (it.hash ^ cacheHashNameWin(it.name)) & mask
-			found := false
-			for lookupIndex[slot].used {
-				bucketItem := cacheItems[lookupIndex[slot].index]
-				if bucketItem.Hash == it.hash && bucketItem.Name == it.name {
-					found = true
-					break
-				}
-				slot = (slot + 1) & mask
-			}
-			if found {
+			if guard.Get(it.hash, it.name) != nil {
 				hits++
 			}
 			lookups++
 		}
+		guard.Unlock()
 	}
 
 	wallSec := float64(nowNS()-wallStart) / 1e9

bench/drivers/go/main_windows.go

@@ -27,7 +27,8 @@ use netipc::protocol::{
 use netipc::service::cgroups::{CgroupsClient, ClientConfig as TypedClientConfig};
 #[cfg(windows)]
 use netipc::service::raw::{
-    increment_dispatch, snapshot_dispatch, CgroupsCacheItem, DispatchHandler, ManagedServer,
+    increment_dispatch, snapshot_dispatch, CgroupsCache, CgroupsCacheItem, DispatchHandler,
+    ManagedServer,
 };
 #[cfg(windows)]
 use netipc::transport::win_shm::{WinShmContext, PROFILE_HYBRID as WIN_SHM_PROFILE_HYBRID};
@@ -1758,20 +1759,6 @@ fn run_pipeline_batch_client(
 
 #[cfg(windows)]
 fn run_lookup_bench(duration_sec: u32) -> i32 {
-    #[derive(Clone, Copy, Default)]
-    struct Bucket {
-        index: usize,
-        used: bool,
-    }
-
-    fn hash_name(name: &str) -> u32 {
-        let mut h: u32 = 5381;
-        for b in name.as_bytes() {
-            h = ((h << 5).wrapping_add(h)).wrapping_add(*b as u32);
-        }
-        h
-    }
-
     let items: Vec<CgroupsCacheItem> = (0..16)
         .map(|i| CgroupsCacheItem {
             hash: 1000 + i,
@@ -1782,33 +1769,30 @@ fn run_lookup_bench(duration_sec: u32) -> i32 {
         })
         .collect();
 
-    let mut lookup_index = vec![Bucket::default(); 32];
-    let mask = (lookup_index.len() - 1) as u32;
-    for (idx, item) in items.iter().enumerate() {
-        let mut slot = (item.hash ^ hash_name(&item.name)) & mask;
-        while lookup_index[slot as usize].used {
-            slot = (slot + 1) & mask;
-        }
-        lookup_index[slot as usize] = Bucket {
-            index: idx,
-            used: true,
-        };
-    }
+    let cache = CgroupsCache::new(
+        ".",
+        "bench_lookup",
+        ClientConfig {
+            supported_profiles: PROFILE_BASELINE,
+            preferred_profiles: 0,
+            max_request_payload_bytes: 0,
+            max_request_batch_items: 1,
+            max_response_payload_bytes: RESPONSE_BUF_SIZE as u32,
+            max_response_batch_items: 1,
+            auth_token: AUTH_TOKEN,
+            packet_size: 0,
+        },
+    );
+    cache.seed_for_tests(items.clone(), 1, 1);
 
     let mut lookups: u64 = 0;
     let mut hits: u64 = 0;
 
     let warmup_deadline = TickDeadline::new(LOOKUP_WARMUP_SEC);
     while !warmup_deadline.expired() {
+        let guard = cache.read_lock();
         for item in &items {
-            let mut slot = (item.hash ^ hash_name(&item.name)) & mask;
-            while lookup_index[slot as usize].used {
-                let bucket_item = &items[lookup_index[slot as usize].index];
-                if bucket_item.hash == item.hash && bucket_item.name == item.name {
-                    break;
-                }
-                slot = (slot + 1) & mask;
-            }
+            let _ = guard.get(item.hash, &item.name);
         }
     }
 
@@ -1817,18 +1801,9 @@ fn run_lookup_bench(duration_sec: u32) -> i32 {
     let tick_deadline = TickDeadline::new(duration_sec);
 
     while !tick_deadline.expired() {
+        let guard = cache.read_lock();
         for item in &items {
-            let mut slot = (item.hash ^ hash_name(&item.name)) & mask;
-            let mut found = false;
-            while lookup_index[slot as usize].used {
-                let bucket_item = &items[lookup_index[slot as usize].index];
-                if bucket_item.hash == item.hash && bucket_item.name == item.name {
-                    found = true;
-                    break;
-                }
-                slot = (slot + 1) & mask;
-            }
-            if found {
+            if guard.get(item.hash, &item.name).is_some() {
                 hits += 1;
             }
             lookups += 1;