【試用報告】RP2040上的MicroPython環境中多線程編程
雙核介紹
BPI-Pico-RP2040官方介紹如下：
其核心是RP2040，采用的是ARM Cortex M0+ CPU內核，運行頻率高達 133 MHz。
比一般使用Cortex M0+的MCU更強大的是，RP2040使用了雙核ARM Cortex M0+，既然是雙核的，那么我們就可以在BPI-Pico-RP2040運行多線程程序了，更好的挖掘出其潛力來。
多線程了解
關于什么是多線程，本文不講，大家可以自行查找資料詳細了解。
為了更方便的進行測試，本次所有的實例，都是在python環境中進行的。
經過了解，circuitpython還不支持多線程，而micropython則已經提供支持。
不過micropython中的多線程還是實驗性質的支持，這從官方文檔中可以了解：MicroPython libraries ? _thread – multithreading support
micropython官方為RP2040提供的最新固件為v1.19.1，其已提供對多線程的支持。
因為micropython的多線程基于CPython中的_thread模塊，所以可以從Python官方文檔了解其具體用法：_thread --- 底層多線程 API
如果是開始使用多線程，那么先關注如下的調用，等熟悉了以后，再深入學習其他的：
_thread.start_new_thread(function, args[, kwargs])：開啟一個新線程
_thread.allocate_lock()：返回一個新的鎖對象
lock.acquire(blocking=True, timeout=- 1)：申請獲得鎖
lock.release()：釋放鎖
本文中所有的實例代碼，都可以從以下地址獲取：
Pico(RP2040)上的MicroPython環境中多線程編程https://gitee.com/honestqiao/multithread_in_micropython_on_pico
基礎多線程
首先，用一個簡單的micropython程序，來控制板載的LED不同時間點亮和熄滅
import machine
import _thread
import utime
led = machine.Pin(25, machine.Pin.OUT)
led.off()
key = 0
start_time = 0
def run_on_core1():
global start_time
while start_time == 0:
pass
while True:
utime.sleep_ms(300)
print((utime.ticks_us()-start_time)//100000, "led on")
led.on()
utime.sleep_ms(700)
def run_on_core0():
global start_time
start_time = utime.ticks_us()
while True:
utime.sleep_ms(700)
print((utime.ticks_us()-start_time)//100000, "led off")
led.off()
utime.sleep_ms(300)
_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()（左右移動查看全部內容）
在RP2040的micropython環境中，程序默認在core0上運行，使用_thread.start_new_thread()啟動新的線程后，將會在core1上運行。
上面的程序運行后，具體輸出結果如下：
在run_on_core1中，先延時300ms，然后點亮led，再延時700ms，然后繼續循環
在run_on_core0中，先延時700ms，然后熄滅led，再延時300ms，然后繼續循環
從以上的輸出可以看到，點亮和熄滅led，都對應到了對應的時間點。
也許有人會說，這有啥用，我不用多線程，也完全可以在對應的時間點點亮和熄滅LED，用多線程豈不是多此一舉。
上面的例子，是一個基礎的多線程演示，其只是在兩個線程中，控制同一個LED，所以會覺得意義不大。如果我們的程序要同時做兩件不同的事情，那么每件事情在一個core上運行，互不干擾，就很重要的，在后面會有這樣的實例展示。
確認雙線程
在不同的開發板上，對多線程的支持，也是有差異的。
RP2040上的micropython，只能跑兩個線程，每個線程占用1個core，多了就會出錯。
我們可以用下面的程序進行驗證：
import machine
import _thread
import utime
def thread_1():
while True:
print("thread_1")
utime.sleep_ms(1000)
def thread_2():
while True:
print("thread_2")
utime.sleep_ms(1000)
_thread.start_new_thread(thread_1, ( ))
_thread.start_new_thread(thread_2, ( ))
while True:
print("main")
utime.sleep_ms(1000)（左右移動查看全部內容）
運行上面的程序后，將會出現如下的錯誤信息：
其原因在于，主程序本身，使用了core0，而使用_thread.start_new_thread()創建一個線程時，會自動的使用core1，第二次調用_thread.start_new_thread()再次創建一個線程時，無法再使用core1，所以就會出錯。
在core1上運行的子線程，需要使用_thread.start_new_thread()創建，所以其運行的需要使用一個函數進行調用作為入口。
而程序的主線程，運行在core0上，可以直接在程序主流程中寫運行邏輯，也可以寫一個函數調用，效果是一樣的。
后續的實例中，我們將使用run_on_core0()和run_on_core1()來區分在core0、core1的所運行的線程。
線程間交互
全局變量
通常時候，讓兩個線程，分別做各自獨立的事情，可以運行的很好。
但有的時候，我們可能還需要兩個之間，能夠有一些交流。
最簡單的方法，就是使用一個全局變量，然后兩個線程之間，都調用這個全局變量即可。
下面用一個簡單的程序進行演示：
import machine
import _thread
import utime
led = machine.Pin(25, machine.Pin.OUT)
led.off()
status = 0
def run_on_core1():
global status
while True:
if status:
led.on()
else:
led.off()
utime.sleep_ms(100)
def run_on_core0():
global status
while True:
status = 1 if not status else 0
utime.sleep_ms(1000)
_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()（左右移動查看全部內容）
在上面的程序中，core0上的程序，每隔1秒，將status取反一次。core1上的程序，則根據status的值，來點亮或者熄滅LED。
線程鎖
上面這個程序比較簡單，處理起來的速度很快，所以這么實用，不會有什么問題。
如果我們有一個程序需要兩個線程進行配合，例如主線程進行數據采集分析，而子線程進行數據的呈現，就有可能會出現問題了。
我們看一看下面的程序：
import machine
import _thread
import utime
led = machine.Pin(25, machine.Pin.OUT)
led.off()
status = 0
data = []
def run_on_core1():
global status, data
while True:
if status:
led.on()
else:
led.off()
str_data = ''.join(data)
print("str_data: len=%d content=%s" % (len(str_data), str_data))
utime.sleep_ms(1000)
def run_on_core0():
global status, data
while True:
status = 1 if not status else 0
data = []
for i in range(100):
data.append(str(status))
utime.sleep_ms(10)
utime.sleep_ms(1000)
_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()（左右移動查看全部內容）
在core0的主線程中，根據status的值，將data設置為100個0或者1；而在core1的子線程中，則將其值合并為字符串輸出出來，輸出的同時，顯示字符串的長度。
運行上面的程序后，實際輸出結果如下：
按說，其長度，要么是空，要么是100，可是實際結果卻會出現不為100的情況呢？
這是因為，core0上的主線程在操作data，core1的子線程也在操作data，兩者都是在同時進行的，而多個控制線程之間是共享全局數據空間，那么就會出現，core0上的主線程處理數據處理到到一半了，core1的子線程已經開始操作了，這樣就會出現問題，數據不完整了。
顯然，這種情況，是我們所不期望的。那要解決這種情況，可以用一個全局變量作為標志，主線程告訴子線程是否處理完成了，一旦處理完成了，子線程就可以開始處理了。
但線程調用庫本身，有更好的辦法，那就是鎖。
我們先看下面的程序：
import machine
import _thread
import utime
led = machine.Pin(25, machine.Pin.OUT)
led.off()
status = 0
data = []
def run_on_core1():
global status, data
while True:
if status:
led.on()
else:
led.off()
lock.acquire()
str_data = ''.join(data)
print("str_data: len=%d content=%s" % (len(str_data), str_data))
lock.release()
utime.sleep_ms(1000)
def run_on_core0():
global status, data
while True:
status = 1 if not status else 0
lock.acquire()
data = []
for i in range(100):
data.append(str(status))
utime.sleep_ms(10)
lock.release()
utime.sleep_ms(1000)
lock = _thread.allocate_lock()
_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()（左右移動查看全部內容）
在上面的程序中，啟動線程之前，使用 _thread.allocate_lock() 來獲取一個新的鎖，然后在core0的主線程中，處理數據前，使用 lock.acquire() 獲得鎖，處理完成后，再使用lock.release()釋放鎖。
一但一個線程獲得鎖，那么其他線程想要獲得該鎖時，只能等待直到這個鎖被釋放，也就是不能同時獲得，這在python中叫做互斥鎖。
因而，在core1的子線程，要輸出數據的時候，也使用同樣的機制來獲得和釋放鎖。
最終，data改變時，其他地方需要等待改變完成。data輸出時，其他地方也需要等待輸出完成。從而確保了任何時刻，對只有一個地方操作改數據。
運行上面的程序，就能得到理想的輸出了：
運行中啟動線程
前面演示的程序，都是在主線程中，啟動了子線程，然后并行運行。
在實際使用中，還可以在主線程中，按需啟動子線程。
我們先看下面的程序：
import machine
import _thread
import utime
def run_on_core1():
print("[core1] run thread")
utime.sleep_ms(100)
def run_on_core0():
while True:
print("[core0] start thread:")
_thread.start_new_thread(run_on_core1, ( ))
utime.sleep_ms(1000)
run_on_core0()（左右移動查看全部內容）
在上面的程序中，core0上運行的主線程，會每過1秒啟動一個子線程。子線程在core1上運行完以后，會自動退出。
運行后，輸出如下：
需要特別注意的是，如果子線程還沒有退出，那么再次啟動，將會出現錯誤。
例如我們修改上面的程序的延時如下：
import machine
import _thread
import utime
def run_on_core1():
print("[core1] run thread")
utime.sleep_ms(1000)
def run_on_core0():
while True:
print("[core0] start thread:")
_thread.start_new_thread(run_on_core1, ( ))
utime.sleep_ms(100)
run_on_core0()（左右移動查看全部內容）
運行后，就會出錯：
[core0] start thread:
[core1] run thread
[core0] start thread:
Traceback (most recent call last):
File "", line 17, in
File "", line 14, in run_on_core0
OSError: core1 in use
其原因就在于，子線程還沒有結束，主線程又再次啟動主線程了。
這在多線程編程中，是需要特別注意的問題。
要解決這個問題，可以使用前面主線程和子線程交互中的方法，例如使用一個全局變量表示子線程是否運行完成，或者使用鎖。
下面是一個使用鎖的程序：
import machine
import _thread
import utime
def run_on_core1():
lock.acquire()
print("[core1] run thread")
utime.sleep_ms(1000)
lock.release()
def run_on_core0():
while True:
print("[core0] start thread:")
lock.acquire()
_thread.start_new_thread(run_on_core1, ( ))
lock.release()
utime.sleep_ms(100)
lock = _thread.allocate_lock()
run_on_core0()（左右移動查看全部內容）
運行后，輸出如下：
[core0] start thread:
[core1] run thread
[core0] start thread:
[core1] run thread
[core0] start thread:
[core1] run thread
[core0] start thread:
[core1] run thread
多線程的實例
雙線程做pwm和ws2812b
下面，再用一段稍微復雜一點點的程序，演示多線程的使用。
import machine
import _thread
import utime
from ws2812 import WS2812
led = machine.Pin(25, machine.Pin.OUT)
led.off()
BLACK = (0, 0, 0)
RED = (255, 0, 0)
YELLOW = (255, 150, 0)
GREEN = (0, 255, 0)
CYAN = (0, 255, 255)
BLUE = (0, 0, 255)
PURPLE = (180, 0, 255)
WHITE = (255, 255, 255)
COLORS = (BLACK, RED, YELLOW, GREEN, CYAN, BLUE, PURPLE, WHITE)
ws = WS2812(3, 1) #WS2812(pin_num,led_count)
ws.pixels_fill(BLACK)
ws.pixels_show()
def run_on_core1():
while True:
for color in COLORS:
ws.pixels_fill(color)
ws.pixels_show()
utime.sleep_ms(200)
def run_on_core0():
duty = 0
step = 1
count = 0
while True:
led.on()
utime.sleep_ms(duty)
led.off()
utime.sleep_ms(10-duty)
count = count + 1
if count>10:
count = 0
duty = duty + step
if duty >= 10:
step = -1
if duty <= 0 :
step = 1
_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()（左右移動查看全部內容）
在上面的這段程序中，我們會在core0上運行的主線程中，控制GPIO25的輸出占空比，從而讓板載LED產生類似呼吸燈的效果。同時，還會在core1上運行的子線程中，控制板載WS2812B燈珠變色。
雙線程播放Bad Apple
最后，我們再用經典的Bad Apple，作為這篇文章的結尾。
from machine import SPI,Pin
from ssd1306 import SSD1306_SPI
import framebuf
import _thread
import utime
spi = SPI(1, 100000, mosi=Pin(11), sck=Pin(10))
display = SSD1306_SPI(128, 64, spi, Pin(9),Pin(8), Pin(1))
def run_on_core1():
global fbuf
while True:
if not fbuf == None:
display.fill(0)
lock.acquire()
display.blit(fbuf,19,0)
fbuf = None
lock.release()
display.show()
utime.sleep_ms(100)
def run_on_core0():
global fbuf
while True:
for i in range(1,139):
dirt = 'BAD_APPLE/' + str(i) + '.pbm'
print(i, dirt)
with open(dirt,'rb') as f :
f.readline()
f.readline()
data = bytearray(f.read())
lock.acquire()
fbuf = framebuf.FrameBuffer(data,88,64,framebuf.MONO_HLSB)
lock.release()
utime.sleep_ms(100)
fbuf = None
lock = _thread.allocate_lock()
_thread.start_new_thread(run_on_core1, ( ))
run_on_core0()（左右移動查看全部內容）
上面的代碼，使用core0上運行的主線程，來從pbm文件中讀取需要呈現的圖片數據，而在core1上運行的子線程中，則使用讀取到的數據輸出到OLED進行顯示。
因為受限于Pico內置存儲的限制，并沒有存儲完整的Bad Apple數據，所以只播放了部分。如果感興趣，可以將數據放置到SD卡上，主線程讀取數據，子線程顯示數據，一樣絲滑流暢。
后記
多線程是個讓人有愛又恨的東西，用好了能有大作用，但是用不好可能會出現莫名其妙的問題，需要好好鉆研。本文只是一些較為基礎的研究，還比較淺顯，對于gc等方面，都尚未涉及，感興趣的讀者可以進一步深入了解。
在鉆研的過程中，參考了不少資料，對所有資料的貢獻者表示感謝。以下為參考到的部分資料列表：
更多熱點文章閱讀
原文標題：【試用報告】RP2040上的MicroPython環境中多線程編程
文章出處：【微信公眾號：電子發燒友論壇】歡迎添加關注！文章轉載請注明出處。
-
電子技術
+關注
關注
18文章
890瀏覽量
55979 -
電子發燒友論壇
+關注
關注
4文章
197瀏覽量
1085
原文標題：【試用報告】RP2040上的MicroPython環境中多線程編程
文章出處：【微信號：gh_9b9470648b3c，微信公眾號：電子發燒友論壇】歡迎添加關注！文章轉載請注明出處。
發布評論請先 登錄
相關推薦
【社區工程師專題系列第九期】一個狂熱的開源愛好者和傳播者——喬楚
socket 多線程編程實現方法
Python中多線程和多進程的區別
我用了3分鐘，從零實現了單片機的點燈開發！
ubuntu上交叉編譯rp2040
樹莓派Pico 2發布，搭載RP2350雙核RISC-V和Arm Cortex-M33微控制器！
pico-ice：RP2040 plus Lattice iCE40UP5K FPGA 開發板 介紹
從多線程設計模式到對 CompletableFuture 的應用
基于樹莓派RP2040的解魔方機器人，7秒還原三階魔方
基于樹莓派RP2040單片機設計的三階魔方還原機器人
工商網監
評論