在前幾天的文章中,我們探討了如何透過「軟即時排程器」與「優先權設計」來管理邊緣閘道器 (Edge Gateway) 上的硬體狀態 (LED 指示燈)。我們實作了一個搶佔式的設計:當發生高優先權事件 (例如致命錯誤) 時,系統必須立刻打斷低優先權任務 (例如待機呼吸燈)。
但這衍生出一個 Python 開發者遲早會遇到的經典問題:我們要怎麼「立刻打斷 (Kill)」一個正在無窮迴圈中執行的 Python Thread?
在許多高階語言中,強制終止執行緒是非常危險的行為,而 Python 直接在標準函式庫 (threading) 中拔除了這個功能。你找不到 Thread.terminate() 或 Thread.kill() 這類方法。
這在我們的硬體控制中會造成災難。想像一下以下情境:
綠燈亮 -> 睡 1 秒 -> 綠燈暗 -> 睡 1 秒 的呼吸燈效果。紅燈快閃。如果我們無法殺死原本的「綠燈 Thread」,兩條 Thread 就會同時競爭 GPIO 腳位,結果 LED 就會變成「綠燈與紅燈不規則狂閃」的靈異現象,也就是我們在 Day 20 提過的 Thread Chaos。
大一資工系的直覺解法通常是:用一個全域變數 (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 之所以不允許外部強制中止 Thread,是因為這會破壞程式的內部狀態一致性。
如果 Thread 在持有 Lock (鎖) 時被殺掉,那個 Lock 就永遠不會被釋放,導致整個系統 Deadlock。如果 Thread 正在寫入檔案,強制殺掉會導致檔案毀損。
因此,Python 鼓勵的作法是**「合作式中止 (Cooperative Cancellation)」**。外部只能發送「建議退出的訊號」,Thread 必須自己定期檢查這個訊號,並在安全的地方 return 結束自己。
為了解決 Global Flag 帶來的 Race Condition,我們在 DataGateway 的任務排程器中引入了一個簡單卻極度優雅的設計模式:世代計數器 (Generation Counter),或是我們在程式碼中稱之為 Job ID 的概念。
核心精神是:我們不告訴 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 退出。
世代計數器 (Generation Counter) 是一個在併發編程 (Concurrent Programming) 中極度強大卻常被忽略的技巧。它把「終止 Thread 的主動權」從主程式交還給了 Thread 自己,完美解決了硬體控制上的狀態殘留問題。
然而,眼尖的讀者可能會發現一個致命傷:「如果舊的 Thread 卡在 time.sleep(10) 裡面睡上 10 秒,那它豈不是要等 10 秒後才會醒來發現自己過氣了?」
沒錯!這在要求反應速度的邊緣閘道器上是不可接受的,如果發生致命錯誤,紅燈應該要「立刻」亮起,而不是等綠燈睡醒。
在明天的 Day 24 中,我們將要徹底消滅硬體開發者最愛用的 time.sleep(),介紹如何透過迴圈切片與 Debounce (防彈跳) 技巧,打造真正零延遲的任務切換!