Raspberry Pi Pico Task Scheduler: Timer‑Based MicroPython Multitasking
MicroPython task scheduler built upon software‑timer, featuring Task/Scheduler classes with task add/delete/pause/resume, idle‑time GC and exception handling.
COMPONENTS
PROJECT DESCRIPTION
【Preliminary Note】The original hardware example in this article was written based on the RP2040. The actual hardware used in this hands-on demonstration features the W55RP20 as the main controller chip. The circuit logic and UF2 flashing operation principles are universally applicable, with only the main controller model differing. The original chip model mentioned in the circuit descriptions below is provided for reference purposes only.
1. Basic Concepts and Characteristics of Task Scheduling
The biggest convenience of using timers is that they allow us to perform certain tasks at specific points in time or at regular intervals, which we can regard as an execution flow different from the while loop in the main program:
We can use the machine. Timer library to create timer objects, run timers in single-shot or periodic mode, execute callback functions and stop timers, which is extremely convenient.
Consider the following scenario:
Task 1 is executed once every 500ms on a scheduled basis, and will no longer run 3000ms after the program starts.
Task 2 is executed once every 1000ms on a scheduled basis, suspended 3000ms after the program starts running, and resumes execution 7000ms after the program starts running
Execute Task 3 periodically every 1000ms starting from 3000ms after the program starts running.
Perform garbage collection when there are currently no tasks to execute
If multiple timer objects are used, a large number of additional global variables and judgment conditions will be introduced, which will make the program complex and difficult to expand, for example:
After Task 1 finishes running, we need to decide whether to start Task 3 based on the data and execution status obtained from Task 1
Task 2 needs to have its scheduled run time modified at 7500 ms after the program starts running, and needs to be paused again at 10000 ms after the program starts running.
In this case, we can use a single timer to implement a task scheduler, which is a common programming pattern in embedded systems or real-time operating systems for managing and executing scheduled tasks.
Its main functions include:
Task Management:
Define the data structure of the task, including task callback function, execution cycle, parameters, etc.
Provides methods for adding, deleting, pausing and resuming tasks
Maintain the task list and track the status and execution progress of tasks
Task Scheduling:
Determine which tasks need to be executed according to the execution cycle of the tasks and the current time
Execute the callback function of the task and handle any possible exceptions that may arise
Simply put, a task scheduler uses timers (such as hardware timers or software timers on microcontrollers) to periodically check the task list, determine which tasks need to be executed and which need to be suspended, and then perform corresponding operations.
2. Implementation of Custom Task Scheduling Class
Here, the routines reference code from the MicroPython Chinese Community, with some comments and minor modifications added on that basis:
Task class represents a scheduled task, which includes attributes such as task callback function, parameters, execution interval and task status. It also provides methods for pausing the task, resuming the task and executing the callback function. The task status is represented by two constants: TASK_RUN and TASK_STOP.
Here, the core implementation principle of the pause method and resume method is to modify the status of the task instance. When the scheduler iterates through the status of each task instance in the task list, it will not execute the task if the task status is Task. TASK_STOP; otherwise, it will execute the task.
Then, we defined the task scheduling Scheduler class, and the Scheduler class is responsible for managing and executing all registered tasks.
A timer instance (machine. Timer), timer interval, optional task idle callback function and task error callback function need to be passed in during initialization;_tmrirq () method is the timer interrupt callback function, which is used to iterate through the task list and determine whether a task is running; if the task is still in the running state, the task's execution interval will be updated:
def _tmrirq(self, t: Timer) -> None:
"""
定时器回调中断函数，用于处理定时器中断。
Args:
t (machine.Timer): 定时器实例，触发中断。
Returns:
None
"""
# 遍历任务列表
for i in range(len(self._tasks)):
# 判断任务状态
if self._tasks[i]._state == Task.TASK_RUN:
# 任务_tasks[i]的执行时间间隔+1
self._tasks[i]._rt += 1
_run () method is used to execute the callback function of a single task and handle any exceptions that may occur during task execution.
The key method of the Scheduler class is scheduler (), which is an infinite loop responsible for continuously checking the status of tasks in the task list and performing corresponding operations according to the task status. Specifically, it will:
Iterate through the task list and for each task:
If the task status is Task. TASK_RUN, increase the task execution time interval;
If the execution time interval of the task reaches or exceeds the time interval required for the task to be executed, the callback function of the task will be executed, and the execution time interval will be reset to 0;
If all tasks are in the suspended state and there is an idle processing function, call this function;
If any exception occurs, print the exception information and continue the execution.
def scheduler(self) -> None:
"""
调度器的主循环，负责循环执行所有已注册的任务。
Args:
None
Returns:
None
"""
# 轮询检测任务状态，判断是否执行任务
while True:
try:
# 执行到达执行时间间隔的任务
for i in range(len(self._tasks)):
task = self._tasks[i]
self._run(task)
# 若空闲，则执行任务空闲回调函数
if self._task_idle:
self._task_idle()
except KeyboardInterrupt:
return
# 发生异常时，抛出异常位置和类型
except Exception as e:
print('except {}'.format(e))
In addition, the Scheduler class also provides other methods, such as find () for finding the task index, clear () for clearing the task list, add () for adding new tasks, delete () for deleting tasks, pause () and resume () for pausing and resuming tasks, as well as run () for executing tasks immediately.
All the code for the following experiments is open-source and can be found in the material package we provide, in the elegance-devkit v1\Demo\23 TIMER_Scheduler folder.
Next, we will main. py to complete our sample program. First, import the relevant modules:
Then define some code that initializes the global variable time_start to record the task start time, and RunCnt to record the execution count of the callback function:
Next, define a timing decorator function to measure the running time of the program:
# 计时装饰器，用于计算函数运行时间
def timed_function(f: callable, *args: tuple, **kwargs: dict) -> callable:
"""
计时装饰器，用于计算并打印函数/方法运行时间。
Args:
f (callable): 需要传入的函数/方法
args (tuple): 函数/方法 f 传入的任意数量的位置参数
kwargs (dict): 函数/方法 f 传入的任意数量的关键字参数
Returns:
callable: 返回计时后的函数
"""
myname = str(f).split(' ')[1]
def new_func(*args: tuple, **kwargs: dict) -> any:
t: int = time.ticks_us()
result = f(*args, **kwargs)
delta: int = time.ticks_diff(time.ticks_us(), t)
print('Function {} Time = {:6.3f}ms'.format(myname, delta / 1000))
return result
return new_func
Then we define task_callback task callback function, which is used to perform specific operations when each task is executed, and this function accepts one parameter task_id that represents the ID of the current task. The main execution logic of this function is as follows:
Calculate the time difference from the task start time to the current time, and convert it into milliseconds
Output the number of task runs, task running time and task ID
Increment the task run count by 1
Perform different operations according to the number of task runs:
If the task has run 5 times, pause Task 2, create a new Task 3 (executed once every 1000 milliseconds), add it to the scheduler, and delete Task 1
Then two callback functions are defined: task_idle_callback () and task_err_callback ():
Idle Task Callback Function: The task_idle_callback function is invoked when there are no tasks to be executed in the scheduler. Inside this function, it first checks whether the number of bytes of currently available heap RAM is less than 230000. If so, it indicates that the memory may be insufficient, so the garbage collection function (gc. collect ()) is manually triggered to release memory that is no longer in use. The function of this function is to help the system perform self-optimization when memory is tight.
Exception callback function: The task_err_callback function is invoked when an exception occurs during task execution. Within this function, sys. print_exception (e) is used to print the detailed information of the exception. After that, the function enters an infinite loop, continuously printing "task run error" so that issues can be detected and handled in a timely manner when exceptions arise. The function serves to provide error handling and debugging information when a task execution error occurs.
