iT邦幫忙

2026 iThome 鐵人賽

DAY 23
0
Software Development

從零打造邊緣運算閘道器:Raspberry Pi 與 Linux 底層軟硬整合實戰系列 第 23

Day 23 - 如何優雅地殺死 Python Thread?世代計數器 (Generation Counter) 的妙用

  • 分享至 

  • xImage
  •  

在前幾天的文章中,我們探討了如何透過「軟即時排程器」與「優先權設計」來管理邊緣閘道器 (Edge Gateway) 上的硬體狀態 (LED 指示燈)。我們實作了一個搶佔式的設計:當發生高優先權事件 (例如致命錯誤) 時,系統必須立刻打斷低優先權任務 (例如待機呼吸燈)。

但這衍生出一個 Python 開發者遲早會遇到的經典問題:我們要怎麼「立刻打斷 (Kill)」一個正在無窮迴圈中執行的 Python Thread?


🚫 問題情境:殺不死的 Thread

在許多高階語言中,強制終止執行緒是非常危險的行為,而 Python 直接在標準函式庫 (threading) 中拔除了這個功能。你找不到 Thread.terminate()Thread.kill() 這類方法。

這在我們的硬體控制中會造成災難。想像一下以下情境:

  1. 閘道器目前處於「待機狀態」,所以系統開了一條 Thread,不斷在迴圈裡執行 綠燈亮 -> 睡 1 秒 -> 綠燈暗 -> 睡 1 秒 的呼吸燈效果。
  2. 突然,系統偵測到 USB 讀寫錯誤 (致命錯誤)。
  3. 主程式立刻開了另一條新的 Thread,準備執行 紅燈快閃

如果我們無法殺死原本的「綠燈 Thread」,兩條 Thread 就會同時競爭 GPIO 腳位,結果 LED 就會變成「綠燈與紅燈不規則狂閃」的靈異現象,也就是我們在 Day 20 提過的 Thread Chaos。


❌ 錯誤嘗試:使用全域 Flag

大一資工系的直覺解法通常是:用一個全域變數 (Global Boolean Flag) 讓 Thread 自己退出。

# 全域變數
is_running = True

def standby_led_task():
    while is_running:
        set_green_led(ON)
        time.sleep(1)
        set_green_led(OFF)
        time.sleep(1)

# 主程式要中斷它時:
is_running = False
time.sleep(0.1) # 等它退出
is_running = True # 再次打開,讓下一個任務執行
start_error_led_task()

為什麼這招在邊緣運算中行不通?

這就是標準的競爭條件 (Race Condition)
如果在主程式將 is_running 設為 False 的瞬間,standby_led_task 剛好卡在 time.sleep(1) 裡面睡覺呢?
主程式等了 0.1 秒覺得它應該死了,於是立刻把 is_running 設回 True 並啟動下一個任務。結果 standby_led_task 醒來一看,發現 is_running 又是 True,於是它繼續開心地閃它的綠燈!

最終,我們還是得到了紅綠燈交替閃爍的悲劇。


🔍 底層原理:為何 Python 不給你 Kill?

Python 之所以不允許外部強制中止 Thread,是因為這會破壞程式的內部狀態一致性
如果 Thread 在持有 Lock (鎖) 時被殺掉,那個 Lock 就永遠不會被釋放,導致整個系統 Deadlock。如果 Thread 正在寫入檔案,強制殺掉會導致檔案毀損。

因此,Python 鼓勵的作法是**「合作式中止 (Cooperative Cancellation)」**。外部只能發送「建議退出的訊號」,Thread 必須自己定期檢查這個訊號,並在安全的地方 return 結束自己。


💡 最終解決方案:世代計數器 (Generation Counter)

為了解決 Global Flag 帶來的 Race Condition,我們在 DataGateway 的任務排程器中引入了一個簡單卻極度優雅的設計模式:世代計數器 (Generation Counter),或是我們在程式碼中稱之為 Job ID 的概念。

核心精神是:我們不告訴 Thread 什麼時候該死,我們只告訴它「你是不是已經過氣了」。

實作步驟:

  1. 宣告一個全域 (或隸屬於排程器) 的計數器。
  2. 每次指派新任務時,計數器遞增 (+1)。
  3. 將當下的計數器數值,當作參數傳入 Thread 裡。
  4. Thread 在每次迴圈時,檢查自己的數值是否等於全域的數值。如果不等,代表已經有新世代接班了,立刻退出!

程式碼範例:

import threading
import time

class LEDScheduler:
    def __init__(self):
        self.current_generation = 0
        self.lock = threading.Lock()
        
    def start_task(self, task_func):
        with self.lock:
            # 產生新的世代 (Job ID)
            self.current_generation += 1
            assigned_gen = self.current_generation
            
        print(f"啟動新任務,世代編號: {assigned_gen}")
        
        # 將專屬的世代編號傳給 Thread
        t = threading.Thread(target=task_func, args=(assigned_gen,))
        t.daemon = True
        t.start()
        
    def is_current_generation(self, my_gen):
        # 讓 Thread 隨時可以檢查自己是不是最新的
        with self.lock:
            return self.current_generation == my_gen

scheduler = LEDScheduler()

# 定義一個硬體控制任務
def blink_green(my_gen):
    while True:
        # 【關鍵檢查點】
        if not scheduler.is_current_generation(my_gen):
            print(f"[Thread {my_gen}] 我已經過氣了,優雅退出!")
            break
            
        print(f"[Thread {my_gen}] 綠燈亮...")
        time.sleep(1) # 這裡稍微有阻塞風險,我們下一篇 Day 24 會解決
        print(f"[Thread {my_gen}] 綠燈暗...")
        time.sleep(1)

# 測試情境
scheduler.start_task(blink_green) # 啟動世代 1
time.sleep(1.5)

# 觸發高優先權錯誤!直接指派新任務,不用手動 kill
scheduler.start_task(blink_green) # 啟動世代 2
time.sleep(3)

執行的結果會是:

啟動新任務,世代編號: 1
[Thread 1] 綠燈亮...
[Thread 1] 綠燈暗...
啟動新任務,世代編號: 2
[Thread 2] 綠燈亮...
[Thread 1] 我已經過氣了,優雅退出!
[Thread 2] 綠燈暗...
[Thread 2] 綠燈亮...

看見了嗎?我們根本不需要寫出 stop_task() 或是把 is_running 設為 False。當「世代 2」被啟動的瞬間,全域的 current_generation 變成了 2。
此時還在睡覺的「世代 1」醒來後,呼叫 is_current_generation(1) 發現結果是 1 == 2 (False),它就知道自己已經被取代了,便心甘情願地 break 退出。

設計的優勢

  1. 絕對無競爭條件 (No Race Condition):不管舊的 Thread 睡了多久才醒來,只要它的號碼牌比全域的號碼牌舊,它就只有死路一條。我們徹底消滅了「新舊任務同時亮燈」的硬體混亂。
  2. 擴展性極佳:如果你一口氣狂發 100 個新任務,前 99 個 Thread 在醒來時都會乖乖自我了斷,最後只會留下第 100 個在執行。
  3. 符合 Python 哲學:Thread 依然是「合作式中止」,它們會在迴圈的起點檢查並安全釋放資源,沒有強制刪除帶來的 Deadlock 風險。

結語

世代計數器 (Generation Counter) 是一個在併發編程 (Concurrent Programming) 中極度強大卻常被忽略的技巧。它把「終止 Thread 的主動權」從主程式交還給了 Thread 自己,完美解決了硬體控制上的狀態殘留問題。

然而,眼尖的讀者可能會發現一個致命傷:「如果舊的 Thread 卡在 time.sleep(10) 裡面睡上 10 秒,那它豈不是要等 10 秒後才會醒來發現自己過氣了?」
沒錯!這在要求反應速度的邊緣閘道器上是不可接受的,如果發生致命錯誤,紅燈應該要「立刻」亮起,而不是等綠燈睡醒。

在明天的 Day 24 中,我們將要徹底消滅硬體開發者最愛用的 time.sleep(),介紹如何透過迴圈切片與 Debounce (防彈跳) 技巧,打造真正零延遲的任務切換!


上一篇
Day 22 - 優先權設計:如何確保「致命錯誤」的紅燈永遠不會被「待機」的綠燈覆蓋?
下一篇
[Day 24] 硬體通訊的眉角:time.sleep() 帶來的災難與 Debounce (防彈跳) 實戰
系列文
從零打造邊緣運算閘道器:Raspberry Pi 與 Linux 底層軟硬整合實戰26
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言