diff --git a/LICENSE-expection b/LICENSE-expection new file mode 100644 index 0000000..4dced57 --- /dev/null +++ b/LICENSE-expection @@ -0,0 +1,31 @@ +SPDX-Exception-Identifier: Gthulhu-plugin-exception +SPDX-URL: https://github.com/Gthulhu/plugin/blob/main/LICENSE-expection +SPDX-Licenses: Apache-2.0 +Usage-Guide: + This exception is used together with the Apache-2.0 license + to allow this plugin to be linked and used by GPL-licensed projects + (specifically Gthulhu/Gthulhu and Gthulhu/qumun) without + affecting the license terms of those projects. + To use this exception, the plugin may be incorporated into GPL projects + under the terms: + SPDX-License-Identifier: Apache-2.0 WITH Gthulhu-plugin-exception +License-Text: + + NOTE! This exception allows the plugin code licensed under Apache-2.0 + to be used as a dependency or component within GPL-licensed projects + (Gthulhu/Gthulhu and Gthulhu/qumun) without creating license + compatibility issues. + + This plugin is designed specifically to serve as a library/package for + the aforementioned GPL projects. The use of this plugin by those projects + does not affect their GPL licensing terms, nor does it require this plugin + to adopt GPL licensing. + + This exception only applies to the integration and use of this plugin + within the specified GPL projects. Any other use or distribution of this + plugin remains subject to the terms of the Apache-2.0 license. + + The plugin maintains its Apache-2.0 licensing independently and does not + incorporate any GPL-licensed code from the projects that use it. + + Gthulhu Project Maintainers \ No newline at end of file diff --git a/plugin/simple/README.md b/plugin/simple/README.md new file mode 100644 index 0000000..c483ab3 --- /dev/null +++ b/plugin/simple/README.md @@ -0,0 +1,189 @@ +# Simple Scheduler Plugin + +A basic scheduler implementation inspired by the simple BPF scheduler from the Linux kernel's sched_ext framework. This Go implementation provides a clean, efficient task scheduling solution with dual-mode operation. + +## Overview + +The Simple Scheduler Plugin implements a lightweight, high-performance task scheduler that can operate in two distinct modes: + +1. **Weighted Virtual Time Scheduling** (default) - Fair scheduling based on task weights and virtual time +2. **FIFO Scheduling** - First-In-First-Out scheduling for specific workloads + +## Features + +### Core Functionality +- **Dual-mode operation**: Switch between weighted vtime and FIFO scheduling +- **Dynamic slice-based task pool**: Efficient task management with up to 4096 task capacity +- **Virtual time tracking**: Global vtime progression for fair scheduling (never 0) +- **Robust task validation**: Prevents invalid tasks and duplicate dispatch +- **Statistics tracking**: Monitor local and global queue metrics +- **Fixed time slice**: 5ms default time slice for all tasks + +### Scheduling Modes + +#### Weighted Virtual Time Scheduling +- Tasks are scheduled based on their virtual time (vtime) and weight +- Lower vtime tasks get higher priority +- **vtime is never 0** - minimum value is 1 to prevent scheduling anomalies +- Prevents idle tasks from accumulating excessive budget +- Global vtime progresses as tasks execute and starts at 1 +- Execution time is scaled by task weight + +#### FIFO Scheduling +- Simple first-in-first-out task ordering +- No vtime calculations or weight considerations +- Suitable for workloads where order of arrival matters +- Lower overhead compared to weighted vtime mode + +## Architecture + +### Core Components + +```go +type SimplePlugin struct { + fifoMode bool // Scheduling mode flag + sliceDefault uint64 // Default time slice (5ms) + taskPool []Task // Circular buffer for tasks + vtimeNow uint64 // Global virtual time tracker + // Statistics and pool management fields... +} +``` + +### Task Management +- **Task Pool**: Dynamic slice implementation with automatic growth/shrinkage +- **Insertion Logic**: + - FIFO mode: Append to end of slice + - Weighted vtime mode: Insert in vtime-sorted order using slice operations +- **Selection Logic**: Always select from front of slice + +### Interface Compliance +Fully implements the `plugin.CustomScheduler` interface: +- `DrainQueuedTask()` - Drain tasks from system queue to local pool +- `SelectQueuedTask()` - Select next task for execution +- `SelectCPU()` - Always select any CPU (returns 1<<20) +- `DetermineTimeSlice()` - Return fixed 5ms time slice +- `GetPoolCount()` - Return current task pool size + +## Usage + +### Basic Usage + +```go +// Create weighted vtime scheduler (default) +scheduler := NewSimplePlugin(false) + +// Create FIFO scheduler +fifoScheduler := NewSimplePlugin(true) + +// Switch modes at runtime +scheduler.SetMode(true) // Switch to FIFO +scheduler.SetMode(false) // Switch to weighted vtime +``` + +### Integration with Scheduler Framework + +```go +// Drain tasks from system queue +drained := scheduler.DrainQueuedTask(systemScheduler) + +// Select next task for execution +task := scheduler.SelectQueuedTask(systemScheduler) + +// Get CPU assignment +err, cpu := scheduler.SelectCPU(systemScheduler, task) + +// Get time slice +timeSlice := scheduler.DetermineTimeSlice(systemScheduler, task) +``` + +### Statistics Monitoring + +```go +// Get queue statistics +local, global := scheduler.GetStats() +fmt.Printf("Local: %d, Global: %d\n", local, global) + +// Reset statistics +scheduler.ResetStats() + +// Get current pool size +poolSize := scheduler.GetPoolCount() +``` + +## Testing + +The package includes comprehensive tests covering: + +### Unit Tests +- **Instance Isolation**: Multiple scheduler instances maintain independent state +- **Configuration Management**: Default settings and mode switching +- **Task Pool Operations**: Initialization, insertion, and retrieval +- **Statistics Tracking**: Counter updates and reset functionality + +### Integration Tests +- **Runtime Simulation**: End-to-end scheduling workflows +- **Mode-specific Behavior**: FIFO vs weighted vtime scheduling differences +- **Edge Cases**: Empty queues, pool overflow, concurrent operations +- **Virtual Time Updates**: Proper vtime progression and task charging + +### Mock Framework +Includes `MockScheduler` for isolated testing: +- Task queue simulation +- CPU allocation tracking +- Operation counting +- State reset capabilities + +## Performance Characteristics + +### Time Complexity +- **Task Insertion**: + - FIFO mode: O(1) amortized (slice append) + - Weighted vtime mode: O(n) where n is current pool size +- **Task Selection**: O(1) with slice reslicing +- **Pool Management**: O(1) amortized for most operations + +### Memory Usage +- Dynamic memory footprint: grows/shrinks based on actual task count +- Pre-allocated capacity up to 4096 tasks to minimize reallocations +- Automatic garbage collection of completed tasks +- Minimal per-task metadata overhead + +### Scalability +- Shared queue across all CPUs ensures fair distribution +- May benefit interactive workloads in FIFO mode +- Good performance on systems with sufficient CPU count + +## Limitations + +1. **No Preemption**: Tasks run until completion or yield +2. **Fixed Time Slice**: All tasks get the same 5ms time slice +3. **Simple CPU Selection**: Always selects any CPU (returns 1<<20) +4. **FIFO Saturation**: In FIFO mode, CPU-intensive tasks can starve interactive tasks + +## Comparison with Reference Implementation + +This Go implementation closely follows the behavior of the Linux kernel's simple BPF scheduler: + +### Similarities +- Dual-mode operation (weighted vtime vs FIFO) +- Virtual time progression and task charging +- Budget limiting for idle tasks +- Statistics tracking +- Similar algorithmic approach + +### Differences +- **Language**: Go vs C/BPF +- **Environment**: Userspace vs kernel space +- **CPU Selection**: Delegated vs direct DSQ management +- **Memory Management**: Go GC vs manual memory management +- **Task Pool**: Circular buffer vs eBPF maps + +## Contributing + +When contributing to this scheduler: + +1. Maintain interface compatibility with `plugin.CustomScheduler` +2. Add appropriate test coverage for new features +3. Follow Go formatting conventions (`go fmt`) +4. Consider performance implications of changes +5. Update this README for significant changes \ No newline at end of file diff --git a/plugin/simple/simple.go b/plugin/simple/simple.go new file mode 100644 index 0000000..0a4bf92 --- /dev/null +++ b/plugin/simple/simple.go @@ -0,0 +1,257 @@ +package simple + +import ( + "github.com/Gthulhu/plugin/models" + "github.com/Gthulhu/plugin/plugin" +) + +// SimplePlugin implements a basic scheduler that can operate in two modes: +// 1. Weighted vtime scheduling (default) +// 2. FIFO scheduling +type SimplePlugin struct { + // Configuration + fifoMode bool + sliceDefault uint64 + + // Task pool for managing queued tasks + taskPool []Task + + // Global vtime tracking (for weighted vtime scheduling) + vtimeNow uint64 + + // Statistics + localQueueCount uint64 + globalQueueCount uint64 +} + +// Task represents a task in the scheduler pool +type Task struct { + QueuedTask *models.QueuedTask + VTime uint64 + Timestamp uint64 +} + +const ( + sliceDefault = 5000 * 100 // 0.5ms in nanoseconds +) + +// NewSimplePlugin creates a new SimplePlugin instance +func NewSimplePlugin(fifoMode bool) *SimplePlugin { + return &SimplePlugin{ + fifoMode: fifoMode, + sliceDefault: sliceDefault, + taskPool: make([]Task, 0), // Start with empty slice, no pre-allocation + vtimeNow: 1, // Start with 1 to ensure vtime is never 0 + localQueueCount: 0, + globalQueueCount: 0, + } +} + +func (s *SimplePlugin) SetSliceDefault(slice uint64) { + s.sliceDefault = slice +} + +// Verify that SimplePlugin implements the plugin.CustomScheduler interface +var _ plugin.CustomScheduler = (*SimplePlugin)(nil) + +// DrainQueuedTask drains tasks from the scheduler queue into the task pool +func (s *SimplePlugin) DrainQueuedTask(sched plugin.Sched) int { + count := 0 + + // Keep draining until the pool is full or no more tasks available + for { + var queuedTask models.QueuedTask + sched.DequeueTask(&queuedTask) + + // Validate task before processing to prevent corruption + if queuedTask.Pid <= 0 { + // Skip invalid tasks + return count + } + + // Create task and enqueue it + task := s.enqueueTask(&queuedTask) + s.insertTaskToPool(task) + + count++ + s.globalQueueCount++ + } +} + +// SelectQueuedTask selects and returns the next task to be scheduled +func (s *SimplePlugin) SelectQueuedTask(sched plugin.Sched) *models.QueuedTask { + return s.getTaskFromPool() +} + +// SelectCPU selects a CPU for the given task +func (s *SimplePlugin) SelectCPU(sched plugin.Sched, task *models.QueuedTask) (error, int32) { + return nil, 1 << 20 +} + +// DetermineTimeSlice determines the time slice for the given task +func (s *SimplePlugin) DetermineTimeSlice(sched plugin.Sched, task *models.QueuedTask) uint64 { + // Always return default slice + return s.sliceDefault +} + +// GetPoolCount returns the number of tasks in the pool +func (s *SimplePlugin) GetPoolCount() uint64 { + return uint64(len(s.taskPool)) +} + +// enqueueTask processes a task for enqueueing +func (s *SimplePlugin) enqueueTask(queuedTask *models.QueuedTask) Task { + task := Task{ + QueuedTask: queuedTask, + VTime: queuedTask.Vtime, + Timestamp: queuedTask.StartTs, + } + vtime := queuedTask.Vtime + + // Weighted vtime scheduling logic + if !s.fifoMode { + + // Limit the amount of budget that an idling task can accumulate to one slice + if vtime < saturatingSub(s.vtimeNow, s.sliceDefault) { + vtime = saturatingSub(s.vtimeNow, s.sliceDefault) + } + + } + + // Ensure vtime is never 0 - use minimum value of 1 if needed + if vtime == 0 { + vtime = 1 + } + + task.VTime = vtime + + return task +} + +// getTaskFromPool retrieves the next task from the pool +func (s *SimplePlugin) getTaskFromPool() *models.QueuedTask { + if len(s.taskPool) == 0 { + return nil + } + + // Get the first task + task := &s.taskPool[0] + + // Remove the first task from slice + selectedTask := task.QueuedTask + s.taskPool = s.taskPool[1:] + + // Update running task vtime (for weighted vtime scheduling) + if !s.fifoMode { + // Ensure task vtime is never 0 before updating global vtime + if selectedTask.Vtime == 0 { + selectedTask.Vtime = 1 + } + s.updateRunningTask(selectedTask) + } + + return selectedTask +} + +// insertTaskToPool inserts a task into the pool +func (s *SimplePlugin) insertTaskToPool(newTask Task) { + if s.fifoMode { + // FIFO: just append to end + s.taskPool = append(s.taskPool, newTask) + return + } + + // Weighted vtime: insert in sorted order by vtime + insertIdx := len(s.taskPool) // Default to end + for i := 0; i < len(s.taskPool); i++ { + if lessTask(&newTask, &s.taskPool[i]) { + insertIdx = i + break + } + } + + // Insert at the correct position + if insertIdx == len(s.taskPool) { + // Insert at end + s.taskPool = append(s.taskPool, newTask) + } else { + // Insert in middle - grow slice and shift elements + s.taskPool = append(s.taskPool, Task{}) // Add empty element at end + copy(s.taskPool[insertIdx+1:], s.taskPool[insertIdx:]) + s.taskPool[insertIdx] = newTask + } +} + +// lessTask compares two tasks for priority ordering +func lessTask(a, b *Task) bool { + if a.VTime != b.VTime { + return a.VTime < b.VTime + } + if a.Timestamp != b.Timestamp { + return a.Timestamp < b.Timestamp + } + return a.QueuedTask.Pid < b.QueuedTask.Pid +} + +// updateRunningTask updates the vtime when a task starts running +func (s *SimplePlugin) updateRunningTask(task *models.QueuedTask) { + // Global vtime always progresses forward as tasks start executing + if s.vtimeNow < task.Vtime { + s.vtimeNow = task.Vtime + } + + // Ensure global vtime is never 0 + if s.vtimeNow == 0 { + s.vtimeNow = 1 + } +} + +// updateStoppingTask updates the vtime when a task stops running +func (s *SimplePlugin) updateStoppingTask(task *models.QueuedTask, execTime uint64) { + if s.fifoMode { + return + } + + // Scale the execution time by the inverse of the weight and charge + task.Vtime += execTime * 100 / task.Weight + + // Ensure task vtime is never 0 + if task.Vtime == 0 { + task.Vtime = 1 + } +} + +// saturatingSub performs saturating subtraction (returns 0 if b > a) +func saturatingSub(a, b uint64) uint64 { + if a > b { + return a - b + } + return 0 +} + +// GetMode returns whether the scheduler is in FIFO mode +func (s *SimplePlugin) GetMode() bool { + return s.fifoMode +} + +// SetMode sets the scheduling mode +func (s *SimplePlugin) SetMode(fifoMode bool) { + s.fifoMode = fifoMode +} + +// GetStats returns scheduling statistics +func (s *SimplePlugin) GetStats() (uint64, uint64) { + return s.localQueueCount, s.globalQueueCount +} + +// ResetStats resets scheduling statistics +func (s *SimplePlugin) ResetStats() { + s.localQueueCount = 0 + s.globalQueueCount = 0 +} + +// GetPoolStatus returns detailed pool status for debugging +func (s *SimplePlugin) GetPoolStatus() (head, tail, count, capacity int) { + // For slice implementation: head=0, tail=len, count=len, capacity=cap + return 0, len(s.taskPool), len(s.taskPool), cap(s.taskPool) +} diff --git a/plugin/simple/simple_test.go b/plugin/simple/simple_test.go new file mode 100644 index 0000000..8685178 --- /dev/null +++ b/plugin/simple/simple_test.go @@ -0,0 +1,663 @@ +package simple + +import ( + "testing" + + "github.com/Gthulhu/plugin/models" + "github.com/Gthulhu/plugin/plugin" +) + +// TestSimplePluginInstanceIsolation verifies that multiple SimplePlugin instances maintain independent state +func TestSimplePluginInstanceIsolation(t *testing.T) { + // Create two instances with different configurations + plugin1 := NewSimplePlugin(false) // Weighted vtime mode + plugin2 := NewSimplePlugin(true) // FIFO mode + + // Verify that each instance has its own configuration + if plugin1.GetMode() != false { + t.Errorf("Plugin1 mode = %v; want false", plugin1.GetMode()) + } + if plugin2.GetMode() != true { + t.Errorf("Plugin2 mode = %v; want true", plugin2.GetMode()) + } + + // Test configuration changes don't affect other instances + plugin1.SetMode(true) + if plugin2.GetMode() != true { + t.Error("Plugin2 mode should remain unchanged") + } +} + +// TestSimplePluginDefaultConfiguration verifies default configuration is applied +func TestSimplePluginDefaultConfiguration(t *testing.T) { + // Create instance with default configuration + simplePlugin := NewSimplePlugin(false) + + // Verify default slice + if simplePlugin.sliceDefault != sliceDefault { + t.Errorf("Default sliceDefault = %d; want %d", simplePlugin.sliceDefault, sliceDefault) + } + + // Verify initial mode + if simplePlugin.GetMode() != false { + t.Errorf("Default mode = %v; want false", simplePlugin.GetMode()) + } +} + +// TestSimplePluginTaskPoolInitialization verifies task pool is initialized correctly +func TestSimplePluginTaskPoolInitialization(t *testing.T) { + simplePlugin := NewSimplePlugin(false) + + // Verify initial pool count is 0 + if simplePlugin.GetPoolCount() != 0 { + t.Errorf("Initial pool count = %d; want 0", simplePlugin.GetPoolCount()) + } + + // Verify task pool is allocated + if simplePlugin.taskPool == nil { + t.Error("Task pool is nil; expected allocated slice") + } + + // Verify task pool starts empty but has capacity + if len(simplePlugin.taskPool) != 0 { + t.Errorf("Task pool length = %d; want 0 (should start empty)", len(simplePlugin.taskPool)) + } + + // Verify initial statistics + local, global := simplePlugin.GetStats() + if local != 0 || global != 0 { + t.Errorf("Initial stats = (%d, %d); want (0, 0)", local, global) + } +} + +// TestSimplePluginModeSwitch verifies mode switching works correctly +func TestSimplePluginModeSwitch(t *testing.T) { + simplePlugin := NewSimplePlugin(false) + + // Initially weighted vtime mode + if simplePlugin.GetMode() != false { + t.Error("Initial mode should be weighted vtime (false)") + } + + // Switch to FIFO mode + simplePlugin.SetMode(true) + if simplePlugin.GetMode() != true { + t.Error("Mode should be FIFO (true) after switching") + } + + // Switch back to weighted vtime mode + simplePlugin.SetMode(false) + if simplePlugin.GetMode() != false { + t.Error("Mode should be weighted vtime (false) after switching back") + } +} + +// TestSimplePluginStatistics verifies statistics tracking +func TestSimplePluginStatistics(t *testing.T) { + simplePlugin := NewSimplePlugin(false) + + // Initial stats should be zero + local, global := simplePlugin.GetStats() + if local != 0 || global != 0 { + t.Errorf("Initial stats = (%d, %d); want (0, 0)", local, global) + } + + // Reset should maintain zero stats + simplePlugin.ResetStats() + local, global = simplePlugin.GetStats() + if local != 0 || global != 0 { + t.Errorf("Stats after reset = (%d, %d); want (0, 0)", local, global) + } +} + +// MockScheduler implements the plugin.Sched interface for testing +type MockScheduler struct { + taskQueue []*models.QueuedTask + queueIndex int + cpuAllocated map[int32]int32 // PID -> CPU mapping + defaultCPU int32 + dequeueCount int + selectCPUCall int +} + +// Compile-time check that MockScheduler implements plugin.Sched +var _ plugin.Sched = (*MockScheduler)(nil) + +// NewMockScheduler creates a new mock scheduler for testing +func NewMockScheduler() *MockScheduler { + return &MockScheduler{ + taskQueue: make([]*models.QueuedTask, 0), + queueIndex: 0, + cpuAllocated: make(map[int32]int32), + defaultCPU: 0, + } +} + +// EnqueueTask adds a task to the mock scheduler's queue +func (m *MockScheduler) EnqueueTask(task *models.QueuedTask) { + m.taskQueue = append(m.taskQueue, task) +} + +// DequeueTask implements plugin.Sched.DequeueTask +func (m *MockScheduler) DequeueTask(task *models.QueuedTask) { + m.dequeueCount++ + if m.queueIndex >= len(m.taskQueue) { + // No more tasks, return sentinel value + task.Pid = -1 + return + } + + // Copy the task from queue + qt := m.taskQueue[m.queueIndex] + *task = *qt + m.queueIndex++ +} + +// DefaultSelectCPU implements plugin.Sched.DefaultSelectCPU +func (m *MockScheduler) DefaultSelectCPU(t *models.QueuedTask) (error, int32) { + m.selectCPUCall++ + // Simple round-robin CPU selection + cpu := m.defaultCPU + m.defaultCPU = (m.defaultCPU + 1) % 4 // Assume 4 CPUs + m.cpuAllocated[t.Pid] = cpu + return nil, cpu +} + +// Reset resets the mock scheduler state +func (m *MockScheduler) Reset() { + m.taskQueue = make([]*models.QueuedTask, 0) + m.queueIndex = 0 + m.cpuAllocated = make(map[int32]int32) + m.defaultCPU = 0 + m.dequeueCount = 0 + m.selectCPUCall = 0 +} + +// GetCPUMapping returns the current CPU allocation for testing +func (m *MockScheduler) GetCPUMapping() map[int32]int32 { + return m.cpuAllocated +} + +// GetDequeueCount returns the number of dequeue operations +func (m *MockScheduler) GetDequeueCount() int { + return m.dequeueCount +} + +// GetSelectCPUCount returns the number of SelectCPU calls +func (m *MockScheduler) GetSelectCPUCount() int { + return m.selectCPUCall +} + +// TestSimplePluginRuntimeSimulation provides comprehensive runtime testing +func TestSimplePluginRuntimeSimulation(t *testing.T) { + t.Run("EmptyQueue", func(t *testing.T) { + simplePlugin := NewSimplePlugin(false) + mockSched := NewMockScheduler() + + // Drain tasks from empty queue + drained := simplePlugin.DrainQueuedTask(mockSched) + if drained != 0 { + t.Errorf("DrainQueuedTask on empty queue = %d; want 0", drained) + } + + // Try to select a task + task := simplePlugin.SelectQueuedTask(mockSched) + if task != nil { + t.Error("SelectQueuedTask on empty pool should return nil") + } + + // Pool count should be 0 + if simplePlugin.GetPoolCount() != 0 { + t.Errorf("GetPoolCount = %d; want 0", simplePlugin.GetPoolCount()) + } + }) + + t.Run("SingleTaskWorkflow", func(t *testing.T) { + simplePlugin := NewSimplePlugin(false) + mockSched := NewMockScheduler() + + // Create a task + task1 := &models.QueuedTask{ + Pid: 100, + Cpu: -1, + NrCpusAllowed: 4, + Flags: 0, + StartTs: 1000, + StopTs: 2000, + SumExecRuntime: 1000, + Weight: 100, + Vtime: 0, + Tgid: 100, + } + + // Enqueue task + mockSched.EnqueueTask(task1) + + // Drain tasks + drained := simplePlugin.DrainQueuedTask(mockSched) + if drained != 1 { + t.Errorf("DrainQueuedTask = %d; want 1", drained) + } + + // Verify pool count + if simplePlugin.GetPoolCount() != 1 { + t.Errorf("GetPoolCount after drain = %d; want 1", simplePlugin.GetPoolCount()) + } + + // Verify statistics + _, global := simplePlugin.GetStats() + if global != 1 { + t.Errorf("Global queue count = %d; want 1", global) + } + + // Select task from pool + selectedTask := simplePlugin.SelectQueuedTask(mockSched) + if selectedTask == nil { + t.Fatal("SelectQueuedTask returned nil") + } + if selectedTask.Pid != 100 { + t.Errorf("Selected task PID = %d; want 100", selectedTask.Pid) + } + + // Pool count should decrease + if simplePlugin.GetPoolCount() != 0 { + t.Errorf("GetPoolCount after select = %d; want 0", simplePlugin.GetPoolCount()) + } + + // Select CPU for task + err, cpu := simplePlugin.SelectCPU(mockSched, selectedTask) + if err != nil { + t.Errorf("SelectCPU returned error: %v", err) + } + if (cpu < 0 || cpu >= 4) && cpu != 1<<20 { + t.Errorf("SelectCPU returned invalid CPU: %d", cpu) + } + + // Determine time slice + timeSlice := simplePlugin.DetermineTimeSlice(mockSched, selectedTask) + if timeSlice != sliceDefault { + t.Errorf("DetermineTimeSlice = %d; want %d", timeSlice, sliceDefault) + } + }) + + t.Run("MultipleTasksWeightedVtime", func(t *testing.T) { + simplePlugin := NewSimplePlugin(false) // Weighted vtime mode + mockSched := NewMockScheduler() + + // Create multiple tasks with different weights and vtimes + tasks := []*models.QueuedTask{ + {Pid: 100, Weight: 100, Vtime: 5000, Tgid: 100, StartTs: 1000, StopTs: 2000}, + {Pid: 200, Weight: 150, Vtime: 3000, Tgid: 200, StartTs: 1500, StopTs: 2500}, + {Pid: 300, Weight: 80, Vtime: 7000, Tgid: 300, StartTs: 2000, StopTs: 3000}, + } + + // Enqueue all tasks + for _, task := range tasks { + mockSched.EnqueueTask(task) + } + + // Drain all tasks + drained := simplePlugin.DrainQueuedTask(mockSched) + if drained != 3 { + t.Errorf("DrainQueuedTask = %d; want 3", drained) + } + + // Verify pool count + if simplePlugin.GetPoolCount() != 3 { + t.Errorf("GetPoolCount = %d; want 3", simplePlugin.GetPoolCount()) + } + + // Process all tasks - should be ordered by vtime + processedTasks := make([]*models.QueuedTask, 0) + for simplePlugin.GetPoolCount() > 0 { + task := simplePlugin.SelectQueuedTask(mockSched) + if task == nil { + t.Fatal("SelectQueuedTask returned nil while pool count > 0") + } + + // Select CPU and determine time slice + err, cpu := simplePlugin.SelectCPU(mockSched, task) + if err != nil { + t.Errorf("SelectCPU error: %v", err) + } + if cpu < 0 { + t.Errorf("Invalid CPU selected: %d", cpu) + } + + timeSlice := simplePlugin.DetermineTimeSlice(mockSched, task) + if timeSlice != sliceDefault { + t.Errorf("DetermineTimeSlice = %d; want %d", timeSlice, sliceDefault) + } + + processedTasks = append(processedTasks, task) + } + + // Verify all tasks were processed + if len(processedTasks) != 3 { + t.Errorf("Processed tasks = %d; want 3", len(processedTasks)) + } + + // In weighted vtime mode, tasks should be processed in vtime order + // Task with PID 200 should be first (lowest vtime: 3000) + if processedTasks[0].Pid != 200 { + t.Errorf("First processed task PID = %d; want 200", processedTasks[0].Pid) + } + }) + + t.Run("MultipleTasksFIFO", func(t *testing.T) { + simplePlugin := NewSimplePlugin(true) // FIFO mode + mockSched := NewMockScheduler() + + // Create multiple tasks + tasks := []*models.QueuedTask{ + {Pid: 100, Weight: 100, Vtime: 7000, Tgid: 100, StartTs: 1000, StopTs: 2000}, + {Pid: 200, Weight: 150, Vtime: 3000, Tgid: 200, StartTs: 1500, StopTs: 2500}, + {Pid: 300, Weight: 80, Vtime: 5000, Tgid: 300, StartTs: 2000, StopTs: 3000}, + } + + // Enqueue all tasks + for _, task := range tasks { + mockSched.EnqueueTask(task) + } + + // Drain all tasks + drained := simplePlugin.DrainQueuedTask(mockSched) + if drained != 3 { + t.Errorf("DrainQueuedTask = %d; want 3", drained) + } + + // Process all tasks - should be in FIFO order + processedTasks := make([]*models.QueuedTask, 0) + for simplePlugin.GetPoolCount() > 0 { + task := simplePlugin.SelectQueuedTask(mockSched) + if task == nil { + t.Fatal("SelectQueuedTask returned nil while pool count > 0") + } + processedTasks = append(processedTasks, task) + } + + // In FIFO mode, tasks should be processed in enqueue order + expectedOrder := []int32{100, 200, 300} + for i, expectedPID := range expectedOrder { + if processedTasks[i].Pid != expectedPID { + t.Errorf("Task %d PID = %d; want %d", i, processedTasks[i].Pid, expectedPID) + } + } + }) + + t.Run("PoolOverflowHandling", func(t *testing.T) { + simplePlugin := NewSimplePlugin(false) + mockSched := NewMockScheduler() + + // Try to enqueue more tasks than pool can hold + maxTasks := 4095 // Leave room for pool management + + for i := 0; i < maxTasks; i++ { + task := &models.QueuedTask{ + Pid: int32(1000 + i), + Weight: 100, + Vtime: 0, + Tgid: int32(1000 + i), + } + mockSched.EnqueueTask(task) + } + + // Drain tasks - should stop when pool is full + drained := simplePlugin.DrainQueuedTask(mockSched) + + // Should have drained all tasks or stopped when pool is full + if drained > maxTasks { + t.Errorf("Drained more tasks than enqueued: %d > %d", drained, maxTasks) + } + + // Pool count should not exceed capacity + poolCount := simplePlugin.GetPoolCount() + if poolCount > 4095 { + t.Errorf("Pool count exceeds capacity: %d", poolCount) + } + }) + + t.Run("ConcurrentInstanceIsolation", func(t *testing.T) { + // Create two independent plugin instances + plugin1 := NewSimplePlugin(false) // Weighted vtime + plugin2 := NewSimplePlugin(true) // FIFO + + // Create separate mock schedulers + sched1 := NewMockScheduler() + sched2 := NewMockScheduler() + + // Add tasks to each scheduler + sched1.EnqueueTask(&models.QueuedTask{Pid: 100, Weight: 100, Vtime: 5000, Tgid: 100}) + sched1.EnqueueTask(&models.QueuedTask{Pid: 101, Weight: 100, Vtime: 3000, Tgid: 101}) + + sched2.EnqueueTask(&models.QueuedTask{Pid: 200, Weight: 100, Vtime: 7000, Tgid: 200}) + + // Drain tasks in both plugins + drained1 := plugin1.DrainQueuedTask(sched1) + drained2 := plugin2.DrainQueuedTask(sched2) + + if drained1 != 2 { + t.Errorf("Plugin1 drained = %d; want 2", drained1) + } + if drained2 != 1 { + t.Errorf("Plugin2 drained = %d; want 1", drained2) + } + + // Verify pool counts are independent + if plugin1.GetPoolCount() != 2 { + t.Errorf("Plugin1 pool count = %d; want 2", plugin1.GetPoolCount()) + } + if plugin2.GetPoolCount() != 1 { + t.Errorf("Plugin2 pool count = %d; want 1", plugin2.GetPoolCount()) + } + + // Process a task from plugin1 - should get task with lower vtime first + task1 := plugin1.SelectQueuedTask(sched1) + if task1 == nil { + t.Fatal("Plugin1 SelectQueuedTask returned nil") + } + if task1.Pid != 101 { // PID 101 has lower vtime (3000) + t.Errorf("Plugin1 first task PID = %d; want 101", task1.Pid) + } + + // Plugin2's pool should be unaffected + if plugin2.GetPoolCount() != 1 { + t.Errorf("Plugin2 pool count after plugin1 select = %d; want 1", plugin2.GetPoolCount()) + } + + // Plugin1's pool should decrease + if plugin1.GetPoolCount() != 1 { + t.Errorf("Plugin1 pool count after select = %d; want 1", plugin1.GetPoolCount()) + } + }) +} + +// TestSimplePluginVtimeUpdates tests vtime tracking functionality +func TestSimplePluginVtimeUpdates(t *testing.T) { + simplePlugin := NewSimplePlugin(false) // Weighted vtime mode + mockSched := NewMockScheduler() + + // Test with task that has existing vtime + task := &models.QueuedTask{ + Pid: 100, + Weight: 100, + Vtime: 10000, + Tgid: 100, + } + + mockSched.EnqueueTask(task) + simplePlugin.DrainQueuedTask(mockSched) + + // Select the task to trigger vtime update + selectedTask := simplePlugin.SelectQueuedTask(mockSched) + if selectedTask == nil { + t.Fatal("SelectQueuedTask returned nil") + } + + // The global vtime should be updated + if simplePlugin.vtimeNow != 10000 { + t.Errorf("Global vtime = %d; want 10000", simplePlugin.vtimeNow) + } + + // Test stopping task vtime update + simplePlugin.updateStoppingTask(selectedTask, 1000000) // 1ms execution time + expectedVtime := uint64(10000 + 1000000*100/100) // 10000 + 1000000 + if selectedTask.Vtime != expectedVtime { + t.Errorf("Task vtime after stopping = %d; want %d", selectedTask.Vtime, expectedVtime) + } +} + +// TestSimplePluginStatisticsUpdate tests statistics tracking during operations +func TestSimplePluginStatisticsUpdate(t *testing.T) { + simplePlugin := NewSimplePlugin(false) + mockSched := NewMockScheduler() + + // Add some tasks + for i := 0; i < 5; i++ { + task := &models.QueuedTask{ + Pid: int32(100 + i), + Weight: 100, + Vtime: uint64(1000 * i), + Tgid: int32(100 + i), + } + mockSched.EnqueueTask(task) + } + + // Drain tasks + drained := simplePlugin.DrainQueuedTask(mockSched) + if drained != 5 { + t.Errorf("Drained tasks = %d; want 5", drained) + } + + // Check statistics + _, global := simplePlugin.GetStats() + if global != 5 { + t.Errorf("Global queue count = %d; want 5", global) + } + + // Reset and verify + simplePlugin.ResetStats() + _, global = simplePlugin.GetStats() + if global != 0 { + t.Errorf("Global queue count after reset = %d; want 0", global) + } +} + +// TestSimplePluginVtimeNeverZero tests that vtime is never 0 +func TestSimplePluginVtimeNeverZero(t *testing.T) { + simplePlugin := NewSimplePlugin(false) + mockSched := NewMockScheduler() + + // Test 1: Initial vtimeNow should not be 0 + if simplePlugin.vtimeNow == 0 { + t.Error("Initial vtimeNow should not be 0") + } + + // Test 2: Task with vtime 0 should be adjusted + taskWithZeroVtime := &models.QueuedTask{ + Pid: 100, + Weight: 100, + Vtime: 0, // This should be adjusted to 1 + Tgid: 100, + } + mockSched.EnqueueTask(taskWithZeroVtime) + + drained := simplePlugin.DrainQueuedTask(mockSched) + if drained != 1 { + t.Errorf("DrainQueuedTask = %d; want 1", drained) + } + + // Select the task and verify its vtime is not 0 + selectedTask := simplePlugin.SelectQueuedTask(mockSched) + if selectedTask == nil { + t.Fatal("SelectQueuedTask returned nil") + } + if selectedTask.Vtime == 0 { + t.Error("Task vtime should not be 0 after processing") + } + + // Test 3: Global vtime should never be 0 after updateRunningTask + if simplePlugin.vtimeNow == 0 { + t.Error("Global vtimeNow should not be 0 after task processing") + } + + // Test 4: updateStoppingTask should not result in 0 vtime + originalVtime := selectedTask.Vtime + simplePlugin.updateStoppingTask(selectedTask, 0) // Even with 0 exec time + if selectedTask.Vtime == 0 { + t.Error("Task vtime should not be 0 after updateStoppingTask") + } + if selectedTask.Vtime < originalVtime { + t.Error("Task vtime should not decrease after updateStoppingTask") + } +} + +// TestSimplePluginInvalidTaskHandling tests handling of invalid tasks +func TestSimplePluginInvalidTaskHandling(t *testing.T) { + simplePlugin := NewSimplePlugin(false) + mockSched := NewMockScheduler() + + // Add a valid task + validTask := &models.QueuedTask{ + Pid: 100, + Weight: 100, + Vtime: 1000, + Tgid: 100, + } + mockSched.EnqueueTask(validTask) + + // Add an invalid task (will be created internally during test) + invalidTask := &models.QueuedTask{ + Pid: 0, // This will be caught by validation (changed from -1 to 0) + Weight: 100, + Vtime: 1000, + Tgid: 0, + } + mockSched.EnqueueTask(invalidTask) + + // Drain tasks - should only get the valid one + drained := simplePlugin.DrainQueuedTask(mockSched) + if drained != 1 { + t.Errorf("DrainQueuedTask = %d; want 1 (invalid task should be skipped)", drained) + } + + // Pool should only contain 1 valid task + if simplePlugin.GetPoolCount() != 1 { + t.Errorf("Pool count = %d; want 1", simplePlugin.GetPoolCount()) + } +} + +// TestSimplePluginPoolStatusAndCleanup tests pool status and cleanup functionality +func TestSimplePluginPoolStatusAndCleanup(t *testing.T) { + simplePlugin := NewSimplePlugin(false) + + // Check initial pool status + head, tail, count, _ := simplePlugin.GetPoolStatus() + if head != 0 || tail != 0 || count != 0 { + t.Errorf("Initial pool status = (%d, %d, %d); want (0, 0, 0)", + head, tail, count) + } + + // Add some tasks manually to test pool status + for i := 0; i < 3; i++ { + task := &models.QueuedTask{ + Pid: int32(100 + i), + Weight: 100, + Vtime: uint64(1000 * i), + Tgid: int32(100 + i), + } + taskWrapper := simplePlugin.enqueueTask(task) + simplePlugin.insertTaskToPool(taskWrapper) + } + + // Check pool status after adding tasks + _, _, count, _ = simplePlugin.GetPoolStatus() + if count != 3 { + t.Errorf("Pool count after adding tasks = %d; want 3", count) + } + + // Pool count should remain the same + if simplePlugin.GetPoolCount() != 3 { + t.Errorf("Pool count after cleanup = %d; want 3", simplePlugin.GetPoolCount()) + } +}