Single-Threaded CPU

Author: volodimyr

1834.single_threaded_cpu/cpu.py

@@ -0,0 +1,88 @@
+# 1834. Single-Threaded CPU
+# Topics: 'Sorting', 'Array', 'Heap (Priority Queue)'
+# Level: 'Medium'
+
+# You are given n​​​​​​ tasks labeled from 0 to n - 1 represented by a 2D integer array tasks, where tasks[i] = [enqueueTimei, processingTimei] means that the i​​​​​​th​​​​ task will be available to process at enqueueTimei and will take processingTimei to finish processing.
+
+# You have a single-threaded CPU that can process at most one task at a time and will act in the following way:
+
+#     If the CPU is idle and there are no available tasks to process, the CPU remains idle.
+#     If the CPU is idle and there are available tasks, the CPU will choose the one with the shortest processing time. If multiple tasks have the same shortest processing time, it will choose the task with the smallest index.
+#     Once a task is started, the CPU will process the entire task without stopping.
+#     The CPU can finish a task then start a new one instantly.
+
+# Return the order in which the CPU will process the tasks.
+
+ 
+
+# Example 1:
+
+# Input: tasks = [[1,2],[2,4],[3,2],[4,1]]
+# Output: [0,2,3,1]
+# Explanation: The events go as follows: 
+# - At time = 1, task 0 is available to process. Available tasks = {0}.
+# - Also at time = 1, the idle CPU starts processing task 0. Available tasks = {}.
+# - At time = 2, task 1 is available to process. Available tasks = {1}.
+# - At time = 3, task 2 is available to process. Available tasks = {1, 2}.
+# - Also at time = 3, the CPU finishes task 0 and starts processing task 2 as it is the shortest. Available tasks = {1}.
+# - At time = 4, task 3 is available to process. Available tasks = {1, 3}.
+# - At time = 5, the CPU finishes task 2 and starts processing task 3 as it is the shortest. Available tasks = {1}.
+# - At time = 6, the CPU finishes task 3 and starts processing task 1. Available tasks = {}.
+# - At time = 10, the CPU finishes task 1 and becomes idle.
+
+# Example 2:
+
+# Input: tasks = [[7,10],[7,12],[7,5],[7,4],[7,2]]
+# Output: [4,3,2,0,1]
+# Explanation: The events go as follows:
+# - At time = 7, all the tasks become available. Available tasks = {0,1,2,3,4}.
+# - Also at time = 7, the idle CPU starts processing task 4. Available tasks = {0,1,2,3}.
+# - At time = 9, the CPU finishes task 4 and starts processing task 3. Available tasks = {0,1,2}.
+# - At time = 13, the CPU finishes task 3 and starts processing task 2. Available tasks = {0,1}.
+# - At time = 18, the CPU finishes task 2 and starts processing task 0. Available tasks = {1}.
+# - At time = 28, the CPU finishes task 0 and starts processing task 1. Available tasks = {}.
+# - At time = 40, the CPU finishes task 1 and becomes idle.
+
+ 
+
+# Constraints:
+
+#     tasks.length == n
+#     1 <= n <= 105
+#     1 <= enqueueTimei, processingTimei <= 109
+
+import heapq
+from typing import List
+
+class Solution:
+    def getOrder(self, tasks: List[List[int]]) -> List[int]:
+        my_list =[]
+        for i in range (len(tasks)):
+            my_list.append(Task(i,tasks[i][0], tasks[i][1]))
+        sorted_tasks = sorted(my_list, key=lambda p: p.enqueue, reverse=True)
+
+        min_heap = []
+        cur_time = sorted_tasks[-1].enqueue
+        processed = []
+        while sorted_tasks or len(min_heap) > 0:             
+            while sorted_tasks and sorted_tasks[-1].enqueue <= cur_time:
+                heapq.heappush(min_heap, sorted_tasks.pop())
+            if len(min_heap) < 1:
+                cur_time = sorted_tasks[-1].enqueue
+            else:
+                task = heapq.heappop(min_heap)
+                cur_time = cur_time+task.processing
+                processed.append(task.index)
+
+        return processed
+
+class Task:
+    def __init__(self, index, enqueue, processing):
+        self.index = index
+        self.enqueue = enqueue
+        self.processing = processing
+
+    def __lt__(self, other):
+        if self.processing == other.processing:
+            return self.index < other.index
+        return self.processing < other.processing

1834.single_threaded_cpu/test_cpu.py

@@ -0,0 +1,133 @@
+import heapq
+from typing import List
+from cpu import Solution
+
+# Test Suite
+def run_tests():
+    sol = Solution()
+    
+    # Test 1: Example 1 from problem
+    tasks = [[1,2],[2,4],[3,2],[4,1]]
+    result = sol.getOrder(tasks)
+    expected = [0,2,3,1]
+    assert result == expected, f"Test 1 failed: expected {expected}, got {result}"
+    print("✓ Test 1 passed: Example 1")
+    
+    # Test 2: Example 2 from problem
+    tasks = [[7,10],[7,12],[7,5],[7,4],[7,2]]
+    result = sol.getOrder(tasks)
+    expected = [4,3,2,0,1]
+    assert result == expected, f"Test 2 failed: expected {expected}, got {result}"
+    print("✓ Test 2 passed: Example 2 (all tasks at same time)")
+    
+    # Test 3: Single task
+    tasks = [[1,5]]
+    result = sol.getOrder(tasks)
+    expected = [0]
+    assert result == expected, f"Test 3 failed: expected {expected}, got {result}"
+    print("✓ Test 3 passed: Single task")
+    
+    # Test 4: Tasks with same processing time (tie-breaker by index)
+    tasks = [[0,3],[0,3],[0,3]]
+    result = sol.getOrder(tasks)
+    expected = [0,1,2]
+    assert result == expected, f"Test 4 failed: expected {expected}, got {result}"
+    print("✓ Test 4 passed: Same processing time, sorted by index")
+    
+    # Test 5: Sequential tasks (no overlap)
+    tasks = [[1,2],[5,3],[10,1]]
+    result = sol.getOrder(tasks)
+    expected = [0,1,2]
+    assert result == expected, f"Test 5 failed: expected {expected}, got {result}"
+    print("✓ Test 5 passed: Sequential non-overlapping tasks")
+    
+    # Test 6: Tasks with gaps (CPU idle periods)
+    tasks = [[1,2],[10,3],[20,1]]
+    result = sol.getOrder(tasks)
+    expected = [0,1,2]
+    assert result == expected, f"Test 6 failed: expected {expected}, got {result}"
+    print("✓ Test 6 passed: Tasks with idle gaps")
+    
+    # Test 7: All tasks start at time 0
+    tasks = [[0,5],[0,3],[0,4]]
+    result = sol.getOrder(tasks)
+    expected = [1,2,0]
+    assert result == expected, f"Test 7 failed: expected {expected}, got {result}"
+    print("✓ Test 7 passed: All tasks at time 0")
+    
+    # Test 8: Tasks arrive while processing
+    tasks = [[0,10],[1,1],[2,1],[3,1]]
+    result = sol.getOrder(tasks)
+    expected = [0,1,2,3]
+    assert result == expected, f"Test 8 failed: expected {expected}, got {result}"
+    print("✓ Test 8 passed: Tasks arrive during long processing")
+    
+    # Test 9: Later task has shorter processing time
+    tasks = [[0,5],[1,2]]
+    result = sol.getOrder(tasks)
+    expected = [0,1]
+    assert result == expected, f"Test 9 failed: expected {expected}, got {result}"
+    print("✓ Test 9 passed: Later task can't interrupt")
+    
+    # Test 10: Multiple tasks become available, choose shortest
+    tasks = [[1,5],[2,2],[2,3],[2,1]]
+    result = sol.getOrder(tasks)
+    expected = [0,3,1,2]
+    assert result == expected, f"Test 10 failed: expected {expected}, got {result}"
+    print("✓ Test 10 passed: Choose shortest among available")
+    
+    # Test 11: Large enqueue times (testing time jump)
+    tasks = [[1000000000,1],[1000000001,1]]
+    result = sol.getOrder(tasks)
+    expected = [0,1]
+    assert result == expected, f"Test 11 failed: expected {expected}, got {result}"
+    print("✓ Test 11 passed: Large enqueue times")
+    
+    # Test 12: Same enqueue and processing, different indices
+    tasks = [[5,5],[5,5],[5,5]]
+    result = sol.getOrder(tasks)
+    expected = [0,1,2]
+    assert result == expected, f"Test 12 failed: expected {expected}, got {result}"
+    print("✓ Test 12 passed: Identical tasks, sorted by index")
+    
+    # Test 13: Reverse order arrival
+    tasks = [[10,1],[5,1],[1,1]]
+    result = sol.getOrder(tasks)
+    expected = [2,1,0]
+    assert result == expected, f"Test 13 failed: expected {expected}, got {result}"
+    print("✓ Test 13 passed: Tasks arrive in reverse order")
+    
+    # Test 14: Mix of short and long tasks
+    tasks = [[1,100],[2,1],[3,1],[4,1]]
+    result = sol.getOrder(tasks)
+    expected = [0,1,2,3]
+    assert result == expected, f"Test 14 failed: expected {expected}, got {result}"
+    print("✓ Test 14 passed: Long task followed by short tasks")
+    
+    # Test 15: Tasks that pile up during processing
+    tasks = [[0,50],[10,1],[20,1],[30,1],[40,1]]
+    result = sol.getOrder(tasks)
+    expected = [0,1,2,3,4]
+    assert result == expected, f"Test 15 failed: expected {expected}, got {result}"
+    print("✓ Test 15 passed: Tasks pile up during long processing")
+    
+    # Test 16: Edge case - two tasks, second has same enqueue as first finishes
+    tasks = [[1,2],[3,1]]
+    result = sol.getOrder(tasks)
+    expected = [0,1]
+    assert result == expected, f"Test 16 failed: expected {expected}, got {result}"
+    print("✓ Test 16 passed: Task arrives exactly when CPU becomes free")
+    
+    # Test 17: Multiple tasks with complex tie-breaking
+    tasks = [[0,5],[0,5],[0,3],[0,3]]
+    result = sol.getOrder(tasks)
+    expected = [2,3,0,1]
+    assert result == expected, f"Test 17 failed: expected {expected}, got {result}"
+    print("✓ Test 17 passed: Complex tie-breaking scenario")
+    
+    print("\n" + "="*50)
+    print("All 17 tests passed! ✓")
+    print("="*50)
+
+if __name__ == "__main__":
+    run_tests()