// SPDX-License-Identifier: GPL-2.0/* * Copyright (c) 2025 NVIDIA Corporation. */#define _GNU_SOURCE#include <stdio.h>#include <unistd.h>#include <signal.h>#include <time.h>#include <bpf/bpf.h>#include <scx/common.h>#include <sys/wait.h>#include <sched.h>#include <pthread.h>#include "scx_test.h"#include "dequeue.bpf.skel.h" #define NUM_WORKERS 8#define AFFINITY_HAMMER_MS 500 /* * Worker function that creates enqueue/dequeue events via CPU work and * sleep. */static void worker_fn(int id){	int i;	volatile int sum = 0; 	for (i = 0; i < 1000; i++) {		volatile int j; 		/* Do some work to trigger scheduling events */		for (j = 0; j < 10000; j++)			sum += j;		asm volatile("" : : "r"(sum)); 		/* Sleep to trigger dequeue */		usleep(1000 + (id * 100));	} 	exit(0);} /* * This thread changes workers' affinity from outside so that some changes * hit tasks while they are still in the scheduler's queue and trigger * property-change dequeues. */static void *affinity_hammer_fn(void *arg){	pid_t *pids = arg;	cpu_set_t cpuset;	int i = 0, n = NUM_WORKERS;	struct timespec start, now; 	clock_gettime(CLOCK_MONOTONIC, &start);	while (1) {		int w = i % n;		int cpu = (i / n) % 4; 		CPU_ZERO(&cpuset);		CPU_SET(cpu, &cpuset);		sched_setaffinity(pids[w], sizeof(cpuset), &cpuset);		i++; 		/* Check elapsed time every 256 iterations to limit gettime cost */		if ((i & 255) == 0) {			long long elapsed_ms; 			clock_gettime(CLOCK_MONOTONIC, &now);			elapsed_ms = (now.tv_sec - start.tv_sec) * 1000LL +				     (now.tv_nsec - start.tv_nsec) / 1000000;			if (elapsed_ms >= AFFINITY_HAMMER_MS)				break;		}	}	return NULL;} static enum scx_test_status run_scenario(struct dequeue *skel, u32 scenario,					 const char *scenario_name){	struct bpf_link *link;	pid_t pids[NUM_WORKERS];	pthread_t hammer; 	int i, status;	u64 enq_start, deq_start,	    dispatch_deq_start, change_deq_start, bpf_queue_full_start;	u64 enq_delta, deq_delta,	    dispatch_deq_delta, change_deq_delta, bpf_queue_full_delta; 	/* Set the test scenario */	skel->bss->test_scenario = scenario; 	/* Record starting counts */	enq_start = skel->bss->enqueue_cnt;	deq_start = skel->bss->dequeue_cnt;	dispatch_deq_start = skel->bss->dispatch_dequeue_cnt;	change_deq_start = skel->bss->change_dequeue_cnt;	bpf_queue_full_start = skel->bss->bpf_queue_full; 	link = bpf_map__attach_struct_ops(skel->maps.dequeue_ops);	SCX_FAIL_IF(!link, "Failed to attach struct_ops for scenario %s", scenario_name); 	/* Fork worker processes to generate enqueue/dequeue events */	for (i = 0; i < NUM_WORKERS; i++) {		pids[i] = fork();		SCX_FAIL_IF(pids[i] < 0, "Failed to fork worker %d", i); 		if (pids[i] == 0) {			worker_fn(i);			/* Should not reach here */			exit(1);		}	} 	/*	 * Run an "affinity hammer" so that some property changes hit tasks	 * while they are still in BPF custody (e.g., in user DSQ or BPF	 * queue), triggering SCX_DEQ_SCHED_CHANGE dequeues.	 */	SCX_FAIL_IF(pthread_create(&hammer, NULL, affinity_hammer_fn, pids) != 0,		    "Failed to create affinity hammer thread");	pthread_join(hammer, NULL); 	/* Wait for all workers to complete */	for (i = 0; i < NUM_WORKERS; i++) {		SCX_FAIL_IF(waitpid(pids[i], &status, 0) != pids[i],			    "Failed to wait for worker %d", i);		SCX_FAIL_IF(status != 0, "Worker %d exited with status %d", i, status);	} 	bpf_link__destroy(link); 	SCX_EQ(skel->data->uei.kind, EXIT_KIND(SCX_EXIT_UNREG)); 	/* Calculate deltas */	enq_delta = skel->bss->enqueue_cnt - enq_start;	deq_delta = skel->bss->dequeue_cnt - deq_start;	dispatch_deq_delta = skel->bss->dispatch_dequeue_cnt - dispatch_deq_start;	change_deq_delta = skel->bss->change_dequeue_cnt - change_deq_start;	bpf_queue_full_delta = skel->bss->bpf_queue_full - bpf_queue_full_start; 	printf("%s:\n", scenario_name);	printf("  enqueues: %lu\n", (unsigned long)enq_delta);	printf("  dequeues: %lu (dispatch: %lu, property_change: %lu)\n",	       (unsigned long)deq_delta,	       (unsigned long)dispatch_deq_delta,	       (unsigned long)change_deq_delta);	printf("  BPF queue full: %lu\n", (unsigned long)bpf_queue_full_delta); 	/*	 * Validate enqueue/dequeue lifecycle tracking.	 *	 * For scenarios 0, 1, 3, 4 (local and global DSQs from	 * ops.select_cpu() and ops.enqueue()), both enqueues and dequeues	 * should be 0 because tasks bypass the BPF scheduler entirely:	 * tasks never enter BPF scheduler's custody.	 *	 * For scenarios 2, 5, 6 (user DSQ or BPF internal queue) we expect	 * both enqueues and dequeues.	 *	 * The BPF code does strict state machine validation with	 * scx_bpf_error() to ensure the workflow semantics are correct.	 *	 * If we reach this point without errors, the semantics are	 * validated correctly.	 */	if (scenario == 0 || scenario == 1 ||	    scenario == 3 || scenario == 4) {		/* Tasks bypass BPF scheduler completely */		SCX_EQ(enq_delta, 0);		SCX_EQ(deq_delta, 0);		SCX_EQ(dispatch_deq_delta, 0);		SCX_EQ(change_deq_delta, 0);	} else {		/*		 * User DSQ from ops.enqueue() or ops.select_cpu(): tasks		 * enter BPF scheduler's custody.		 *		 * Also validate 1:1 enqueue/dequeue pairing.		 */		SCX_GT(enq_delta, 0);		SCX_GT(deq_delta, 0);		SCX_EQ(enq_delta, deq_delta);	} 	return SCX_TEST_PASS;} static enum scx_test_status setup(void **ctx){	struct dequeue *skel; 	skel = dequeue__open();	SCX_FAIL_IF(!skel, "Failed to open skel");	SCX_ENUM_INIT(skel);	SCX_FAIL_IF(dequeue__load(skel), "Failed to load skel"); 	*ctx = skel; 	return SCX_TEST_PASS;} static enum scx_test_status run(void *ctx){	struct dequeue *skel = ctx;	enum scx_test_status status; 	status = run_scenario(skel, 0, "Scenario 0: Local DSQ from ops.select_cpu()");	if (status != SCX_TEST_PASS)		return status; 	status = run_scenario(skel, 1, "Scenario 1: Global DSQ from ops.select_cpu()");	if (status != SCX_TEST_PASS)		return status; 	status = run_scenario(skel, 2, "Scenario 2: User DSQ from ops.select_cpu()");	if (status != SCX_TEST_PASS)		return status; 	status = run_scenario(skel, 3, "Scenario 3: Local DSQ from ops.enqueue()");	if (status != SCX_TEST_PASS)		return status; 	status = run_scenario(skel, 4, "Scenario 4: Global DSQ from ops.enqueue()");	if (status != SCX_TEST_PASS)		return status; 	status = run_scenario(skel, 5, "Scenario 5: User DSQ from ops.enqueue()");	if (status != SCX_TEST_PASS)		return status; 	status = run_scenario(skel, 6, "Scenario 6: BPF queue from ops.enqueue()");	if (status != SCX_TEST_PASS)		return status; 	printf("\n=== Summary ===\n");	printf("Total enqueues: %lu\n", (unsigned long)skel->bss->enqueue_cnt);	printf("Total dequeues: %lu\n", (unsigned long)skel->bss->dequeue_cnt);	printf("  Dispatch dequeues: %lu (no flag, normal workflow)\n",	       (unsigned long)skel->bss->dispatch_dequeue_cnt);	printf("  Property change dequeues: %lu (SCX_DEQ_SCHED_CHANGE flag)\n",	       (unsigned long)skel->bss->change_dequeue_cnt);	printf("  BPF queue full: %lu\n",	       (unsigned long)skel->bss->bpf_queue_full);	printf("\nAll scenarios passed - no state machine violations detected\n");	printf("-> Validated: Local DSQ dispatch bypasses BPF scheduler\n");	printf("-> Validated: Global DSQ dispatch bypasses BPF scheduler\n");	printf("-> Validated: User DSQ dispatch triggers ops.dequeue() callbacks\n");	printf("-> Validated: Dispatch dequeues have no flags (normal workflow)\n");	printf("-> Validated: Property change dequeues have SCX_DEQ_SCHED_CHANGE flag\n");	printf("-> Validated: No duplicate enqueues or invalid state transitions\n"); 	return SCX_TEST_PASS;} static void cleanup(void *ctx){	struct dequeue *skel = ctx; 	dequeue__destroy(skel);} struct scx_test dequeue_test = {	.name = "dequeue",	.description = "Verify ops.dequeue() semantics",	.setup = setup,	.run = run,	.cleanup = cleanup,}; REGISTER_SCX_TEST(&dequeue_test)