bug: core count

Signed-off-by: vsoch <vsoch@users.noreply.github.com>

Author: vsoch

README.md

@@ -39,6 +39,8 @@ Everything in the graph that can be allocated is a node vertex. That includes th
 
 Real nodes have the attribute virtual=false and virtual resources have virtual=true. A full set of attributes might include virtual=true, class=qpu, and region=us-east-1, because RFC 31 only checks whether a key is present. You select a resource by its class, which means any level of the tree can be asked for on its own. Classes are inherited so a nested type is not pruned at its parent on the way down.
 
+A vertex is the unit Fluxion allocates. Anything under a slot is taken whole, so a node modeled as one core vertex of size 64 could hold a single slot of a gang, and two pods of one gang could only land on two nodes. Real nodes are built the way Flux builds them, one vertex per core and per GPU. Memory is chunked, which is how flux-sched's own test graphs and Slurm's memory per CPU work. Each node gets one memory chunk per core, sized as the node's memory split over its cores in MiB, and a pod's request in MiB takes however many whole chunks cover it. The vertex count stays at about twice the cores, and the kubelet's own fit check still admits the pod with the exact bytes.
+
 ## Resources
 
 Here is an example of resources for AWS Braket.

examples/test/e2e/gang/multi-gang-contention.yaml

@@ -1,6 +1,7 @@
-# Two gangs that cannot both place. fluxion allocates one core per slot, so two
-# 2-pod gangs need 4 cores, but the cluster graphs ~3 (3 workers, ~1 core each). One gang places entirely. The loser stays FULLY pending
-# (all-or-nothing), never partial.
+# Two gangs that cannot both place. GANG_CPU is filled in by the test with the
+# free cores on a worker, so one pod fills one worker and a gang of two takes
+# both. One gang places entirely. The loser stays FULLY pending (all-or-nothing),
+# never partial.
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -18,7 +19,7 @@ spec:
         - name: w
           image: busybox
           command: ["sleep", "3600"]
-          resources: {requests: {cpu: "1"}}
+          resources: {requests: {cpu: "${GANG_CPU}"}}
 ---
 apiVersion: apps/v1
 kind: Deployment
@@ -37,4 +38,4 @@ spec:
         - name: w
           image: busybox
           command: ["sleep", "3600"]
-          resources: {requests: {cpu: "1"}}
+          resources: {requests: {cpu: "${GANG_CPU}"}}

examples/test/e2e/gang/multi-gang-requeue.yaml

@@ -4,7 +4,8 @@
 # gang-wait. A 2-pod gang needing the same nodes. Loses the initial race and sits
 #            Unschedulable. When gang-win completes, gang-wait must be re-attempted
 #            (via the shortened unschedulable-recheck timeout) and place atomically.
-# On a 3-worker (~3-core) cluster the two 2-pod gangs (4 cores) cannot co-run.
+# GANG_CPU is filled in by the test with the free cores on a worker, so one pod
+# fills one worker and the two gangs cannot co-run.
 apiVersion: batch/v1
 kind: Job
 metadata:
@@ -24,7 +25,7 @@ spec:
         - name: w
           image: busybox
           command: ["sh","-c","sleep 30"]   # completes, frees nodes
-          resources: {requests: {cpu: "1"}}
+          resources: {requests: {cpu: "${GANG_CPU}"}}
 ---
 apiVersion: batch/v1
 kind: Job
@@ -45,4 +46,4 @@ spec:
         - name: w
           image: busybox
           command: ["sh","-c","sleep 10"]
-          resources: {requests: {cpu: "1"}}
+          resources: {requests: {cpu: "${GANG_CPU}"}}

pkg/cluster/cluster.go

@@ -85,14 +85,23 @@ func BuildGraph(nodes []corev1.Node, opts Options) ([]byte, error) {
 		})
 		parents[n.Name] = nodeV
 
-		if cpu := count(n, corev1.ResourceCPU); cpu > 0 {
-			b.AddChild(nodeV, "core", "core", jgf.Options{Size: cpu})
+		// A vertex is the unit Fluxion allocates. Anything under a slot is taken
+		// whole, so a node built as one sized vertex per type can host one slot
+		// of a gang. Cores and GPUs are one vertex each, as in Flux. Memory is
+		// chunked the way flux-sched's own graphs and Slurm's memory per CPU
+		// do it, one chunk per core sized in MiB, so a request takes whole
+		// chunks and the vertex count stays at twice the cores.
+		cores := count(n, corev1.ResourceCPU)
+		for i := int64(0); i < cores; i++ {
+			b.AddChild(nodeV, "core", "core", jgf.Options{})
 		}
-		if memMB := memoryMB(n); memMB > 0 {
-			b.AddChild(nodeV, "memory", "memory", jgf.Options{Size: memMB, Unit: "MB"})
+		if chunk := memoryChunkMiB(n, cores); chunk > 0 {
+			for i := int64(0); i < cores; i++ {
+				b.AddChild(nodeV, "memory", "memory", jgf.Options{Size: chunk, Unit: "MiB"})
+			}
 		}
-		if gpu := count(n, gpuName); gpu > 0 {
-			b.AddChild(nodeV, "gpu", "gpu", jgf.Options{Size: gpu})
+		for i := int64(0); i < count(n, gpuName); i++ {
+			b.AddChild(nodeV, "gpu", "gpu", jgf.Options{})
 		}
 	}
 
@@ -131,14 +140,19 @@ func count(n *corev1.Node, name corev1.ResourceName) int64 {
 	return 0
 }
 
-// memoryMB returns node memory in mebibytes.
-func memoryMB(n *corev1.Node) int64 {
+// memoryChunkMiB is the size of one memory chunk, the node's memory split
+// evenly over its cores, in whole MiB rounded down. Zero when there is no
+// memory or no cores to split it over.
+func memoryChunkMiB(n *corev1.Node, cores int64) int64 {
+	if cores <= 0 {
+		return 0
+	}
 	q, ok := n.Status.Allocatable[corev1.ResourceMemory]
 	if !ok {
 		q, ok = n.Status.Capacity[corev1.ResourceMemory]
 	}
 	if !ok {
 		return 0
 	}
-	return q.Value() / (1024 * 1024)
+	return q.Value() / (1024 * 1024) / cores
 }

pkg/cluster/cluster_test.go

@@ -1,9 +1,11 @@
 package cluster
 
 import (
+	"encoding/json"
 	"testing"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 )
 
 // isControlPlane detects control-plane nodes by taint, independent of node name
@@ -38,3 +40,56 @@ func TestIsControlPlane(t *testing.T) {
 		t.Error("node with no taints should not be control-plane")
 	}
 }
+
+// A real node is built with one vertex per core and per GPU, and memory as one
+// chunk per core, each chunk the node's memory split over its cores in whole
+// MiB. Fluxion hands a vertex to a slot and never splits it, so a node built
+// as one sized vertex per type could host one slot of a gang.
+func TestBuildGraphOneVertexPerUnit(t *testing.T) {
+	n := corev1.Node{}
+	n.Name = "worker"
+	n.Status.Allocatable = corev1.ResourceList{
+		corev1.ResourceCPU:                      resource.MustParse("4"),
+		corev1.ResourceMemory:                   resource.MustParse("8704Mi"), // 8.5 GiB
+		corev1.ResourceName(DefaultGPUResource): resource.MustParse("2"),
+	}
+	out, err := BuildGraph([]corev1.Node{n}, Options{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var doc struct {
+		Graph struct {
+			Nodes []struct {
+				Metadata struct {
+					Type string `json:"type"`
+					Size int64  `json:"size"`
+					Unit string `json:"unit"`
+				} `json:"metadata"`
+			} `json:"nodes"`
+		} `json:"graph"`
+	}
+	if err := json.Unmarshal(out, &doc); err != nil {
+		t.Fatal(err)
+	}
+	count := map[string]int{}
+	for _, v := range doc.Graph.Nodes {
+		count[v.Metadata.Type]++
+		switch v.Metadata.Type {
+		case "memory":
+			// 8704 MiB over 4 cores
+			if v.Metadata.Size != 2176 || v.Metadata.Unit != "MiB" {
+				t.Errorf("memory chunk = %d %s, want 2176 MiB", v.Metadata.Size, v.Metadata.Unit)
+			}
+		case "core", "gpu":
+			if v.Metadata.Size != 1 {
+				t.Errorf("%s vertex has size %d, want 1", v.Metadata.Type, v.Metadata.Size)
+			}
+		}
+	}
+	want := map[string]int{"cluster": 1, "node": 1, "core": 4, "memory": 4, "gpu": 2}
+	for typ, w := range want {
+		if count[typ] != w {
+			t.Errorf("%s vertices = %d, want %d", typ, count[typ], w)
+		}
+	}
+}

pkg/placement/placement.go

@@ -144,6 +144,10 @@ func requireConstraintsFromPods(pods []corev1.Pod) []string {
 // fluxion.flux-framework.org/<type> prefix is a virtual device type instead.
 var computeTypes = map[string]bool{"core": true, "memory": true, "gpu": true}
 
+// mib is the unit memory is requested in. The graph sizes its memory chunks
+// in MiB, so a request in MiB takes however many whole chunks cover it.
+const mib = 1024 * 1024
+
 // podResources distills a pod's container requests into Fluxion resource counts
 // keyed by Fluxion graph type. Native Kubernetes resources (cpu, memory,
 // nvidia.com/gpu) map to compute types. Every fluxion.flux-framework.org/<type>
@@ -156,7 +160,8 @@ func podResources(p *corev1.Pod) map[string]int {
 			case name == corev1.ResourceCPU:
 				counts["core"] += int(q.Value()) // rounds millicores up to whole cores
 			case name == corev1.ResourceMemory:
-				counts["memory"] += int(q.Value() / (1000 * 1000)) // bytes -> MB
+				// whole MiB rounded up, the unit the memory chunks are sized in
+				counts["memory"] += int((q.Value() + mib - 1) / mib)
 			case name == "nvidia.com/gpu":
 				counts["gpu"] += int(q.Value())
 			case strings.HasPrefix(string(name), FluxionResourcePrefix):

pkg/placement/placement_test.go

@@ -439,3 +439,26 @@ func computeConstraints(t *testing.T, spec *jobspec.Jobspec) map[string]interfac
 	}
 	return cons
 }
+
+// Memory requests are counted in whole MiB rounded up, the unit the graph
+// sizes its memory chunks in, so Fluxion takes however many chunks cover it.
+func TestMemoryRequestsCountInMiB(t *testing.T) {
+	cases := []struct {
+		req  string
+		want int
+	}{
+		{"128Mi", 128},
+		{"1Gi", 1024},
+		{"1536Ki", 2},
+		{"100M", 96},
+	}
+	for _, tc := range cases {
+		p := podWith("p", corev1.ResourceList{
+			corev1.ResourceCPU:    qty(1),
+			corev1.ResourceMemory: resource.MustParse(tc.req),
+		})
+		if got := podResources(&p)["memory"]; got != tc.want {
+			t.Errorf("memory %s = %d vertices, want %d", tc.req, got, tc.want)
+		}
+	}
+}

test/e2e/gang/03-multi-gang.sh

@@ -37,7 +37,8 @@ kubectl wait --for=delete pod -l app=gang3 --timeout=60s 2>/dev/null || true
 
 # ---- B) contention: the gang that can't fully fit stays ENTIRELY pending --------
 log "TEST 6B: contention, a gang that cannot fully fit must NOT partially place"
-kubectl apply -f examples/test/e2e/gang/multi-gang-contention.yaml
+log "each pod asks for $(free_cores_per_worker) cores, the free cores on a worker, so a gang of two takes both workers"
+render_gang_manifest examples/test/e2e/gang/multi-gang-contention.yaml | kubectl apply -f -
 
 # wait until the cluster settles. Three possible outcomes.
 #   - one gang fully Running, other fully Pending  -> contention; assert no partial
@@ -66,7 +67,7 @@ else
   log "PASS 6B: $loser stayed entirely pending, no partial placement under contention"
 fi
 
-kubectl delete -f examples/test/e2e/gang/multi-gang-contention.yaml --wait=false || true
+render_gang_manifest examples/test/e2e/gang/multi-gang-contention.yaml | kubectl delete -f - --wait=false || true
 for g in gang-a gang-b; do
   kubectl patch podgroup $g --type=merge -p '{"metadata":{"finalizers":null}}' 2>/dev/null || true
 done

test/e2e/gang/04-requeue-on-capacity.sh

@@ -10,8 +10,9 @@
 #      re-attempted on its own (no manual nudge) and then ALSO places atomically
 #      (2/2), driven by the shortened --pod-max-in-unschedulable-pods-duration.
 #
-# SCOPE / LIMITATION. This is a 3-node kind cluster with small (1-core) pods. It
-# verifies the INVARIANTS on a minimal contention case. It does NOT reproduce the
+# SCOPE / LIMITATION. This is a kind cluster with two workers, and each pod's cpu
+# request is raised to fill a worker, so one gang fits at a time. It verifies
+# the INVARIANTS on a minimal contention case. It does NOT reproduce the
 # GKE-scale dynamics where the bug was first seen, one-pod-per-node (~80-core)
 # saturation and ~20 simultaneous mixed-size gangs draining in sequence. That
 # scale behavior is validated on the real cluster, not in CI. A pass here means
@@ -25,7 +26,8 @@ running() { kubectl get pods -l job-name="$1" --field-selector=status.phase=Runn
 on_nodes() { kubectl get pods -l job-name="$1" -o jsonpath='{range .items[*]}{.spec.nodeName}{"\n"}{end}' 2>/dev/null | grep -c . || true; }
 
 log "TEST 9: contended gangs stay all-or-nothing, loser requeues when capacity frees"
-kubectl apply -f examples/test/e2e/gang/multi-gang-requeue.yaml
+log "each pod asks for $(free_cores_per_worker) cores, the free cores on a worker, so one gang fits at a time"
+render_gang_manifest examples/test/e2e/gang/multi-gang-requeue.yaml | kubectl apply -f -
 
 # ---- 1. one gang wins CLEANLY (2/2); the other places NOTHING (0/2) ------------
 log "waiting for a clean 2/0 split (one whole gang runs, the other entirely waits)"
@@ -79,7 +81,7 @@ either the shortened --pod-max-in-unschedulable-pods-duration is not taking effe
 (gang stuck) or the requeued gang did not assemble"
 log "PASS 9: $loser requeued and placed atomically (2/2) after $winner freed capacity"
 
-kubectl delete -f examples/test/e2e/gang/multi-gang-requeue.yaml --wait=false || true
+render_gang_manifest examples/test/e2e/gang/multi-gang-requeue.yaml | kubectl delete -f - --wait=false || true
 for g in gang-win gang-wait; do
   kubectl patch podgroup $g --type=merge -p '{"metadata":{"finalizers":null}}' 2>/dev/null || true
 done

test/e2e/lib.sh

@@ -85,3 +85,33 @@ assert_stays_pending() {
   done
   return 0
 }
+
+# cpu quantities to millicores, so "4" and "250m" add up
+millicores() { awk '{ if ($0 ~ /m$/) { sub(/m$/, ""); t += $0 } else if ($0 != "") { t += $0 * 1000 } } END { print t + 0 }'; }
+
+# Whole cores free on the fullest worker. Allocatable less what the pods already
+# on it request, which is the arithmetic the kube-scheduler's fit check does. A
+# pod that asks for this fills a worker, so a two pod gang takes two workers and
+# two gangs contend, on any runner, with nothing assumed about the system pods.
+free_cores_per_worker() {
+  local min="" node alloc used free
+  for node in $(kubectl get nodes -l '!node-role.kubernetes.io/control-plane' -o jsonpath='{.items[*].metadata.name}'); do
+    alloc="$(kubectl get node "$node" -o jsonpath='{.status.allocatable.cpu}' | millicores)"
+    used="$(kubectl get pods -A --field-selector "spec.nodeName=$node,status.phase!=Succeeded,status.phase!=Failed" \
+      -o jsonpath='{range .items[*].spec.containers[*]}{.resources.requests.cpu}{"\n"}{end}' | millicores)"
+    free=$(( (alloc - used) / 1000 ))
+    if [ -z "$min" ] || [ "$free" -lt "$min" ]; then min=$free; fi
+  done
+  if [ "${min:-0}" -lt 1 ]; then min=1; fi
+  echo "$min"
+}
+
+# Render a manifest whose cpu request is ${GANG_CPU} with the free cores on a
+# worker. Pipe it to kubectl apply or delete.
+render_gang_manifest() {
+  local cpu line
+  cpu="$(free_cores_per_worker)"
+  while IFS= read -r line || [ -n "$line" ]; do
+    printf '%s\n' "${line//\$\{GANG_CPU\}/$cpu}"
+  done < "$1"
+}