I want to create a custom implementation of the asyncio eventloop so that the tasks handled by the eventloop have a priority. The tasks that are available or “ready” can then be ordered by priority and the eventloop selects the one with the highest priority as the next coroutine to execute.

I’m aware I have to use the BaseEventLoop implementation from asyncio and make some adjustments but I don’t know how to approach this problem.

I made a first try using a heapq but it does not seem to be working as intended

import asyncio
import heapq
import random


# Prioritized task class (unchanged)
class PrioritizedTask(asyncio.Task):
    def __init__(self, priority, coro):
        super().__init__(coro)
        self.priority = priority

#asyncio.Task = PrioritizedTask

def create_prioritized_task(coro, priority):
    pt = asyncio.create_task(coro)
    pt.priority = priority
    

# Custom event loop with priority scheduling (unchanged)
class PrioritizedEventLoop(asyncio.BaseEventLoop):
    def __init__(self):
        self._ready = []  # Queue for tasks with priorities
        super().__init__()

    def _schedule_task(self, task):
        heapq.heappush(self._ready, (task.priority, task))  # Push with priority
        print(task)

    def _run_once(self):
        if not self._ready:
            return 0.0

        print(self._ready)
        priority, task = heapq.heappop(self._ready)
        print(task.name)
        return super()._run_task(task)

    def is_running(self):
        return bool(self._ready)  # Check if tasks are pending in the queue



# Usage
async def example_task(priority, name, event):
    print(f'Task {name} started with priority {priority}')
    await event.wait() # Introduce a delay for each task
    print(f'Task {name} finished with priority {priority} ')


async def main():
    tasks = []
    task_names = list(chr(i) for i in range(65, 75))  # Generate task names (A-J)

    event = asyncio.Event()

    # Completion future for task creation
    #completion_future = asyncio.Future()

    # Use random.sample to get 10 unique priorities from 1 to 10
    priorities = random.sample(range(1, 11), 10)

    for i, name in enumerate(task_names):
        #tasks.append(PrioritizedTask(priorities[i], example_task(priorities[i], name, event)))

        #asyncio.create_task(PrioritizedTask(priorities[i], example_task(priorities[i], name, event)))

        create_prioritized_task(example_task(priorities[i], name, event),priorities[i])

    # Set completion future to True after creating all tasks
    #completion_future.set_result(True)

    #await completion_future  # Wait for all tasks to be created

    #await asyncio.gather(*tasks)  # Run tasks concurrently

    await asyncio.sleep(2)

    event.set()



# Uncomment if using custom loop policy (optional)
# asyncio.set_event_loop_policy(PrioritizedEventLoopPolicy())

asyncio.run(main())