Session 3: Creating, Running, and Managing Tasks

Synopsis

Covers task creation, task lifecycle, awaiting tasks, background task patterns, and organizing multiple concurrent activities on the Pico 2 W.

Session Content

Session 3: Creating, Running, and Managing Tasks

Session Overview

Duration: ~45 minutes
Audience: Python developers with basic programming knowledge learning MicroPython on Raspberry Pi Pico 2 W
Goal: Understand how to create, run, coordinate, and manage asynchronous tasks using uasyncio in MicroPython for responsive embedded applications.

---

Learning Objectives

By the end of this session, learners will be able to:

• Explain what a task is in asynchronous programming
• Create and schedule multiple uasyncio tasks
• Use async/await correctly in MicroPython
• Run concurrent tasks for LEDs, sensors, and Wi-Fi/IoT operations
• Manage task lifetime, delays, cancellation, and errors
• Apply asynchronous task patterns to real Pico 2 W projects

---

Prerequisites

• Raspberry Pi Pico 2 W running MicroPython
• Thonny IDE installed
• Basic familiarity with:
• Python syntax
• GPIO concepts
• async / await basics
• Optional hardware for exercises:
• LED
• 220Ω resistor
• Breadboard and jumper wires
• Button
• DHT22 or similar sensor

---

Required Development Environment Setup

Thonny Configuration

1. Install Thonny
2. Connect Raspberry Pi Pico 2 W via USB
3. In Thonny:
4. Go to Tools > Options > Interpreter
5. Select MicroPython (Raspberry Pi Pico)
6. Choose the correct serial port
7. Ensure the device is running a recent MicroPython firmware release for RP2

Quick Verification Script

Run this in Thonny to confirm the board is ready:

import sys
print(sys.implementation)

Expected output:

(name='micropython', version=(1, 24, 0), ...)

---

Session Agenda

Time	Topic
0–5 min	Recap: async programming model
5–15 min	Creating and running tasks
15–25 min	Task coordination and scheduling
25–35 min	Managing task lifetime and cancellation
35–45 min	Hands-on lab: LED blinker + button monitor + optional sensor task

---

1. Theory: What Is a Task?

A task is an independently scheduled unit of asynchronous work. In MicroPython, tasks run under the uasyncio event loop and cooperate by yielding control at await points.

Key ideas

• A task is created from a coroutine
• Tasks run concurrently, not in parallel
• await allows the event loop to switch to another task
• Long blocking code prevents other tasks from running

Why tasks matter on a Pico 2 W

Tasks let your application: - blink LEDs while reading sensors - listen for button presses while sending network data - keep the system responsive without threads

---

2. Creating and Running Tasks

Coroutine vs Task

A coroutine is an async function object. A task is the scheduled execution of that coroutine.

Example: Basic task creation

import uasyncio as asyncio

async def hello_task():
    print("Task started")
    await asyncio.sleep(1)
    print("Task finished")

async def main():
    task = asyncio.create_task(hello_task())
    print("Task created")
    await task
    print("Main finished")

asyncio.run(main())

Example output

Task created
Task started
Task finished
Main finished

---

Multiple tasks running together

import uasyncio as asyncio

async def task_one():
    for i in range(3):
        print("Task 1:", i)
        await asyncio.sleep(1)

async def task_two():
    for i in range(5):
        print("Task 2:", i)
        await asyncio.sleep(0.5)

async def main():
    t1 = asyncio.create_task(task_one())
    t2 = asyncio.create_task(task_two())
    await t1
    await t2

asyncio.run(main())

Example output

Task 1: 0
Task 2: 0
Task 2: 1
Task 1: 1
Task 2: 2
Task 2: 3
Task 1: 2
Task 2: 4

---

3. Task Scheduling Patterns

Pattern 1: Fire-and-wait

Create a task, then await it later.

import uasyncio as asyncio

async def background_job():
    await asyncio.sleep(2)
    print("Background job done")

async def main():
    job = asyncio.create_task(background_job())
    print("Doing other work...")
    await asyncio.sleep(1)
    print("Waiting for job...")
    await job
    print("All done")

asyncio.run(main())

Pattern 2: Run tasks forever

Useful for sensors, networking, and control loops.

import uasyncio as asyncio

async def blink():
    while True:
        print("LED ON")
        await asyncio.sleep(0.5)
        print("LED OFF")
        await asyncio.sleep(0.5)

async def main():
    asyncio.create_task(blink())
    await asyncio.sleep(5)

asyncio.run(main())

Pattern 3: Grouped concurrent work

Run multiple infinite tasks together.

import uasyncio as asyncio

async def task_a():
    while True:
        print("A")
        await asyncio.sleep(1)

async def task_b():
    while True:
        print("B")
        await asyncio.sleep(0.7)

async def main():
    asyncio.create_task(task_a())
    asyncio.create_task(task_b())
    while True:
        await asyncio.sleep(10)

asyncio.run(main())

---

4. Managing Task Lifetime

Infinite tasks

Many embedded programs use tasks that run forever. Keep in mind: - they must include await calls - they should handle exceptions - they should be cancellable if needed

Cooperative design

A well-behaved task: - runs briefly - yields often - avoids blocking calls like time.sleep() - handles cleanup when cancelled

---

Delays: Use await asyncio.sleep()

Do not use blocking sleeps in async tasks.

Good

await asyncio.sleep(0.2)

Avoid inside tasks

import time
time.sleep(0.2)

The blocking sleep prevents other tasks from running.

---

5. Task Cancellation

Sometimes you need to stop a task, for example: - when a button is pressed - when Wi-Fi disconnects - when a new mode is selected

Cancelling a task safely

import uasyncio as asyncio

async def worker():
    try:
        while True:
            print("Working...")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Worker cancelled, cleaning up...")
        raise

async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(3)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task cancellation confirmed")

asyncio.run(main())

Example output

Working...
Working...
Working...
Worker cancelled, cleaning up...
Task cancellation confirmed

---

6. Handling Exceptions in Tasks

If a task fails, you should know about it.

import uasyncio as asyncio

async def faulty_task():
    await asyncio.sleep(1)
    raise ValueError("Something went wrong")

async def main():
    task = asyncio.create_task(faulty_task())
    try:
        await task
    except Exception as e:
        print("Task error:", e)

asyncio.run(main())

Example output

Task error: Something went wrong

---

7. Hands-On Exercise 1: Concurrent LED Blinker and Status Reporter

Goal

Run two tasks at the same time: - one toggles an LED - one prints a status message

Hardware

• Built-in LED or external LED on GP15
• 220Ω resistor if using external LED

Wiring for external LED

• LED anode to GP15 through a resistor
• LED cathode to GND

Code

from machine import Pin
import uasyncio as asyncio

led = Pin("LED", Pin.OUT)  # Use built-in LED on Pico 2 W

async def blink_led():
    while True:
        led.on()
        print("LED ON")
        await asyncio.sleep(0.5)
        led.off()
        print("LED OFF")
        await asyncio.sleep(0.5)

async def status_reporter():
    count = 0
    while True:
        print("System running... tick =", count)
        count += 1
        await asyncio.sleep(1)

async def main():
    asyncio.create_task(blink_led())
    asyncio.create_task(status_reporter())
    while True:
        await asyncio.sleep(10)

asyncio.run(main())

Expected output

LED ON
System running... tick = 0
LED OFF
LED ON
System running... tick = 1
LED OFF
...

Task

1. Change the blink interval to 200 ms
2. Change the status interval to 2 seconds
3. Observe how the two tasks interleave

---

8. Hands-On Exercise 2: Button-Controlled Task Cancellation

Goal

Start a blinking task and stop it with a button press.

Hardware

• Built-in LED
• Pushbutton
• Jumper wires

Wiring

• One side of the button to GP14
• Other side of the button to GND

Use the internal pull-up resistor.

Code

from machine import Pin
import uasyncio as asyncio

led = Pin("LED", Pin.OUT)
button = Pin(14, Pin.IN, Pin.PULL_UP)

async def blink_led():
    try:
        while True:
            led.on()
            await asyncio.sleep(0.25)
            led.off()
            await asyncio.sleep(0.25)
    except asyncio.CancelledError:
        led.off()
        print("Blink task stopped")
        raise

async def monitor_button(blink_task):
    while True:
        if button.value() == 0:
            print("Button pressed, cancelling blink task...")
            blink_task.cancel()
            return
        await asyncio.sleep(0.05)

async def main():
    blink_task = asyncio.create_task(blink_led())
    button_task = asyncio.create_task(monitor_button(blink_task))

    try:
        await button_task
        await blink_task
    except asyncio.CancelledError:
        pass

asyncio.run(main())

Expected behavior

• LED blinks continuously
• Press the button
• Blink task stops cleanly

Example output

Button pressed, cancelling blink task...
Blink task stopped

Task

Modify the program so that: - first press starts blinking - second press stops blinking

---

9. Hands-On Exercise 3: Multiple Cooperative Tasks with a Sensor

Goal

Add a sensor-reading task to the event loop.

Optional hardware

• DHT22 temperature/humidity sensor
• 10k pull-up resistor if required by your module

Example wiring for DHT22

• VCC to 3.3V
• GND to GND
• DATA to GP16

Code

from machine import Pin
import dht
import uasyncio as asyncio

sensor = dht.DHT22(Pin(16))

async def read_sensor():
    while True:
        try:
            sensor.measure()
            temp = sensor.temperature()
            hum = sensor.humidity()
            print("Temperature:", temp, "C")
            print("Humidity:", hum, "%")
        except Exception as e:
            print("Sensor read error:", e)
        await asyncio.sleep(2)

async def heartbeat():
    led = Pin("LED", Pin.OUT)
    while True:
        led.toggle()
        await asyncio.sleep(0.5)

async def main():
    asyncio.create_task(read_sensor())
    asyncio.create_task(heartbeat())
    while True:
        await asyncio.sleep(10)

asyncio.run(main())

Expected output

Temperature: 24 C
Humidity: 58 %
Temperature: 24 C
Humidity: 57 %

---

10. Common Mistakes and Best Practices

Common mistakes

• Using time.sleep() inside async code
• Forgetting to await a coroutine
• Creating tasks without keeping a reference when they must be managed later
• Blocking inside loops with long computations
• Ignoring exceptions in tasks

Best practices

• Keep tasks small and focused
• Use await asyncio.sleep() to yield control
• Name tasks clearly
• Wrap task bodies in try/except when appropriate
• Clean up hardware state on cancellation
• Use one task per responsibility

---

11. Mini Challenge: Task Dashboard

Goal

Create a simple dashboard using serial output: - Task 1: blink the built-in LED - Task 2: print uptime every second - Task 3: read a button and report presses

Requirements

• All tasks must run together
• No blocking calls
• Button press should not interfere with LED blinking

Starter code

from machine import Pin
import uasyncio as asyncio

led = Pin("LED", Pin.OUT)
button = Pin(14, Pin.IN, Pin.PULL_UP)

async def blink_task():
    while True:
        led.on()
        await asyncio.sleep(0.2)
        led.off()
        await asyncio.sleep(0.2)

async def uptime_task():
    seconds = 0
    while True:
        print("Uptime:", seconds, "s")
        seconds += 1
        await asyncio.sleep(1)

async def button_task():
    while True:
        if button.value() == 0:
            print("Button pressed")
            while button.value() == 0:
                await asyncio.sleep(0.05)
        await asyncio.sleep(0.05)

async def main():
    asyncio.create_task(blink_task())
    asyncio.create_task(uptime_task())
    asyncio.create_task(button_task())
    while True:
        await asyncio.sleep(10)

asyncio.run(main())

Example output

Uptime: 0 s
Uptime: 1 s
Button pressed
Uptime: 2 s
Uptime: 3 s

---

12. Review Questions

1. What is the difference between a coroutine and a task?
2. Why should time.sleep() be avoided in async code?
3. How do tasks cooperate in uasyncio?
4. How do you cancel a task safely?
5. What happens if a task raises an exception and it is not handled?
6. Why is task-based design useful on the Pico 2 W?

---

13. Summary

Key takeaways

• Tasks are scheduled coroutines managed by the event loop
• asyncio.create_task() starts concurrent work
• await is how tasks yield control
• Tasks must be cooperative and non-blocking
• Cancellation and exception handling are essential for robust embedded systems
• Multiple tasks allow responsive, multitasking Pico applications

---

14. Suggested Further Practice

• Add a Wi-Fi status task that periodically checks connectivity
• Create a task that sends sensor data to an MQTT broker
• Build a task manager that starts/stops tasks based on button input
• Add debouncing logic to the button task
• Combine LED, sensor, and network tasks into one IoT application

---

Back to Chapter | Back to Master Plan | Previous Session | Next Session