日韩欧美亚洲在线,不卡国产女人

雙核介紹

BPI-Pico-RP2040官方介紹如下：

其核心是RP2040，采用的是ARM Cortex M0+ CPU內(nèi)核，運行頻率高達 133 MHz。

比一般使用Cortex M0+的MCU更強大的是，RP2040使用了雙核ARM Cortex M0+，既然是雙核的，那么我們就可以在BPI-Pico-RP2040運行多線程程序了，更好的挖掘出其潛力來。

多線程了解

關于什么是多線程，本文不講，大家可以自行查找資料詳細了解。

為了更方便的進行測試，本次所有的實例，都是在python環(huán)境中進行的。

經(jīng)過了解，circuitpython還不支持多線程，而micropython則已經(jīng)提供支持。

不過micropython中的多線程還是實驗性質(zhì)的支持，這從官方文檔中可以了解：MicroPython libraries ? _thread – multithreading support

micropython官方為RP2040提供的最新固件為v1.19.1，其已提供對多線程的支持。

因為micropython的多線程基于CPython中的_thread模塊，所以可以從Python官方文檔了解其具體用法：_thread --- 底層多線程 API

如果是開始使用多線程，那么先關注如下的調(diào)用，等熟悉了以后，再深入學習其他的：

_thread.start_new_thread(function, args[, kwargs])：開啟一個新線程
_thread.allocate_lock()：返回一個新的鎖對象
lock.acquire(blocking=True, timeout=- 1)：申請獲得鎖
lock.release()：釋放鎖

本文中所有的實例代碼，都可以從以下地址獲?。?/p>

Pico(RP2040)上的MicroPython環(huán)境中多線程編程https://gitee.com/honestqiao/multithread_in_micropython_on_pico

基礎多線程

首先，用一個簡單的micropython程序，來控制板載的LED不同時間點亮和熄滅

# file: multicore_test01.py
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()

（左右移動查看全部內(nèi)容）

在RP2040的micropython環(huán)境中，程序默認在core0上運行，使用_thread.start_new_thread()啟動新的線程后，將會在core1上運行。

上面的程序運行后，具體輸出結果如下：

在run_on_core1中，先延時300ms，然后點亮led，再延時700ms，然后繼續(xù)循環(huán)

在run_on_core0中，先延時700ms，然后熄滅led，再延時300ms，然后繼續(xù)循環(huán)

從以上的輸出可以看到，點亮和熄滅led，都對應到了對應的時間點。

也許有人會說，這有啥用，我不用多線程，也完全可以在對應的時間點點亮和熄滅LED，用多線程豈不是多此一舉。

上面的例子，是一個基礎的多線程演示，其只是在兩個線程中，控制同一個LED，所以會覺得意義不大。如果我們的程序要同時做兩件不同的事情，那么每件事情在一個core上運行，互不干擾，就很重要的，在后面會有這樣的實例展示。

確認雙線程

在不同的開發(fā)板上，對多線程的支持，也是有差異的。

RP2040上的micropython，只能跑兩個線程，每個線程占用1個core，多了就會出錯。

我們可以用下面的程序進行驗證：

# file: multicore_test02.py


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)

（左右移動查看全部內(nèi)容）

運行上面的程序后，將會出現(xiàn)如下的錯誤信息：

其原因在于，主程序本身，使用了core0，而使用_thread.start_new_thread()創(chuàng)建一個線程時，會自動的使用core1，第二次調(diào)用_thread.start_new_thread()再次創(chuàng)建一個線程時，無法再使用core1，所以就會出錯。

在core1上運行的子線程，需要使用_thread.start_new_thread()創(chuàng)建，所以其運行的需要使用一個函數(shù)進行調(diào)用作為入口。

而程序的主線程，運行在core0上，可以直接在程序主流程中寫運行邏輯，也可以寫一個函數(shù)調(diào)用，效果是一樣的。

后續(xù)的實例中，我們將使用run_on_core0()和run_on_core1()來區(qū)分在core0、core1的所運行的線程。

線程間交互

全局變量

通常時候，讓兩個線程，分別做各自獨立的事情，可以運行的很好。

但有的時候，我們可能還需要兩個之間，能夠有一些交流。

最簡單的方法，就是使用一個全局變量，然后兩個線程之間，都調(diào)用這個全局變量即可。

下面用一個簡單的程序進行演示：

# file: multicore_test03.py
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()

（左右移動查看全部內(nèi)容）

在上面的程序中，core0上的程序，每隔1秒，將status取反一次。core1上的程序，則根據(jù)status的值，來點亮或者熄滅LED。

線程鎖

上面這個程序比較簡單，處理起來的速度很快，所以這么實用，不會有什么問題。

如果我們有一個程序需要兩個線程進行配合，例如主線程進行數(shù)據(jù)采集分析，而子線程進行數(shù)據(jù)的呈現(xiàn)，就有可能會出現(xiàn)問題了。

我們看一看下面的程序：

# file: multicore_test04.py
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()

（左右移動查看全部內(nèi)容）

在core0的主線程中，根據(jù)status的值，將data設置為100個0或者1；而在core1的子線程中，則將其值合并為字符串輸出出來，輸出的同時，顯示字符串的長度。

運行上面的程序后，實際輸出結果如下：

按說，其長度，要么是空，要么是100，可是實際結果卻會出現(xiàn)不為100的情況呢？

這是因為，core0上的主線程在操作data，core1的子線程也在操作data，兩者都是在同時進行的，而多個控制線程之間是共享全局數(shù)據(jù)空間，那么就會出現(xiàn)，core0上的主線程處理數(shù)據(jù)處理到到一半了，core1的子線程已經(jīng)開始操作了，這樣就會出現(xiàn)問題，數(shù)據(jù)不完整了。

顯然，這種情況，是我們所不期望的。那要解決這種情況，可以用一個全局變量作為標志，主線程告訴子線程是否處理完成了，一旦處理完成了，子線程就可以開始處理了。

但線程調(diào)用庫本身，有更好的辦法，那就是鎖。

我們先看下面的程序：

# file: multicore_test05.py
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()

（左右移動查看全部內(nèi)容）

在上面的程序中，啟動線程之前，使用 _thread.allocate_lock() 來獲取一個新的鎖，然后在core0的主線程中，處理數(shù)據(jù)前，使用 lock.acquire() 獲得鎖，處理完成后，再使用lock.release()釋放鎖。

一但一個線程獲得鎖，那么其他線程想要獲得該鎖時，只能等待直到這個鎖被釋放，也就是不能同時獲得，這在python中叫做互斥鎖。

因而，在core1的子線程，要輸出數(shù)據(jù)的時候，也使用同樣的機制來獲得和釋放鎖。

最終，data改變時，其他地方需要等待改變完成。data輸出時，其他地方也需要等待輸出完成。從而確保了任何時刻，對只有一個地方操作改數(shù)據(jù)。

運行上面的程序，就能得到理想的輸出了：

運行中啟動線程

前面演示的程序，都是在主線程中，啟動了子線程，然后并行運行。

在實際使用中，還可以在主線程中，按需啟動子線程。

我們先看下面的程序：

# file: multicore_test06.py
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()

（左右移動查看全部內(nèi)容）

在上面的程序中，core0上運行的主線程，會每過1秒啟動一個子線程。子線程在core1上運行完以后，會自動退出。

運行后，輸出如下：

需要特別注意的是，如果子線程還沒有退出，那么再次啟動，將會出現(xiàn)錯誤。

例如我們修改上面的程序的延時如下：

# file: multicore_test07.py
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()

（左右移動查看全部內(nèi)容）

運行后，就會出錯：

[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

其原因就在于，子線程還沒有結束，主線程又再次啟動主線程了。

這在多線程編程中，是需要特別注意的問題。

要解決這個問題，可以使用前面主線程和子線程交互中的方法，例如使用一個全局變量表示子線程是否運行完成，或者使用鎖。

下面是一個使用鎖的程序：

# file: multicore_test08.py
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()

（左右移動查看全部內(nèi)容）

運行后，輸出如下：

[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

下面，再用一段稍微復雜一點點的程序，演示多線程的使用。

# file: multicore_test09.py
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()

（左右移動查看全部內(nèi)容）

在上面的這段程序中，我們會在core0上運行的主線程中，控制GPIO25的輸出占空比，從而讓板載LED產(chǎn)生類似呼吸燈的效果。同時，還會在core1上運行的子線程中，控制板載WS2812B燈珠變色。

雙線程播放Bad Apple

最后，我們再用經(jīng)典的Bad Apple，作為這篇文章的結尾。

# file: multicore_test10.py
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()