用 Ada 打造即時系統程式設計 ── 優先權、週期、執行時間控制實戰
· 小村 豪 · Ada, RealTime, Ravenscar, CeilingLocking, Tasking, Scheduling, PriorityInversion, ProgrammingLanguage, 即時系統, 高可靠性
1. 前言 ── Ada 與即時系統的深厚淵源
上一篇文章「Ada 的安全並行處理」介紹了 Ada 透過任務(task)與保護物件(protected object)進行安全並行處理的基礎。這次要進一步深入延伸領域──限制更嚴格的領域──即時系統。
在即時系統中,「正確」不僅代表邏輯計算結果正確,也包含在期限內得到結果。慢了一毫秒的正確答案,跟錯誤答案一樣危險。
Ada 針對這項需求,以語言規範的 Annex D(Real-Time Systems) 標準化提供了一整套即時功能。這並非「用函式庫後補上去」,而是直接內建於語言執行環境本身的即時保證。
Ada 的即時功能(Annex D):
- 任務優先權與搶佔(FIFO_Within_Priorities)
- Ceiling_Locking 協定(防止優先權反轉)
- 以 delay until 進行絕對時刻週期執行
- Ravenscar 設定檔(安全關鍵子集合)
- 計時事件(不需輪詢的時刻喚醒)
- 執行時間監控(Ada.Execution_Time)
- 多週期排程
本文將以 8 個實戰程式碼範例逐步說明這些機制。每個片段都可以當作獨立範例使用,但包含多個編譯單元的 04/05 範例,需要先用 gnatchop 拆分後再用 gnatmake 編譯。
此外,本文出現的程式碼片段已依章節整理成檔案,作為參考程式碼集合公開在 GitHub 上。
ada-real-time-systems - komurasoft-blog-samples (GitHub)
2. 什麼是即時系統
首先從整理術語開始。
| 概念 | 說明 |
|---|---|
| 硬即時 | 超過期限代表系統發生致命失敗(飛行控制、安全氣囊、心律調節器) |
| 軟即時 | 超過期限並不理想,但偶爾超過仍可接受(影音串流、遊戲) |
| 期限 (deadline) | 任務必須完成的絕對時刻 |
| 週期 (period) | 任務重複啟動的時間間隔 |
| WCET (Worst-Case Execution Time) | 任務的最壞情況執行時間 |
| 抖動 (jitter) | 週期執行的變動幅度 |
在即時系統設計中,每個任務必須滿足「WCET <= 期限」,這是重要的必要條件。但光靠這一點並不能保證整個系統都能如期完成。還需要另外進行涵蓋阻塞時間、優先權分配、抖動、中斷,以及執行環境與 OS 行為的回應時間分析。實務上為了保留餘裕,通常會以 WCET < 期限 為目標。Ada 的即時功能,正是在語言層級提供一套容易進行這類分析的可預測執行模型。
flowchart LR
HRT[硬即時] -->|超過期限 = 致命失敗| Examples[飛行控制<br/>安全氣囊<br/>心律調節器]
SRT[軟即時] -->|偶爾超過可接受| Examples2[影音傳輸<br/>遊戲<br/>UI]
Ada[Ada Annex D<br/>支撐可預測性的機制] --> Mechanism[FIFO_Within_Priorities<br/>Ceiling_Locking<br/>delay until]
subgraph Requirements[即時性要求]
D[期限<br/>應完成的絕對時刻]
P[週期<br/>重複間隔]
W[WCET<br/>最壞執行時間]
J[抖動<br/>週期的變動]
end
D --> Analysis[可排程性分析]
P --> Analysis
W --> Analysis
J --> Analysis
HRT --> Analysis
SRT --> Analysis
Mechanism --> Analysis
Analysis --> Constraint[必要條件: WCET <= 期限<br/>充分性須以回應時間分析確認]
即時系統中最危險的現象之一是優先權反轉。這個問題曾在 1997 年的 Mars Pathfinder 上實際發生,造成探測器不斷重新啟動。
sequenceDiagram
participant S as 排程器
participant L as 低優先權任務
participant H as 高優先權任務
participant M as 中優先權任務
participant R as 共享資源
L->>R: 取得鎖
activate L
Note over L: 執行關鍵區段中
Note over S,L: H 喚醒,排程器中斷 L
deactivate L
activate H
H->>R: 嘗試取得鎖
Note over H: 因等待鎖而阻塞!(L 持有中)
deactivate H
Note over S,L: H 在等待鎖,因此恢復執行 L
activate L
Note over L: 繼續朝釋放鎖前進...
Note over S,L: M 喚醒,排程器中斷 L
deactivate L
activate M
Note over L: L 無法釋放鎖
Note over M: M 持續執行(H 與 L 都動不了)
Note over H: 【優先權反轉】高優先權被無限期阻塞
deactivate M
低優先權任務持有鎖時被中優先權任務搶佔,導致高優先權任務被無限期阻塞。Mars Pathfinder 實際採取的對策,是啟用 VxWorks 的 priority inheritance,而 Ada 針對同類問題則以另一種方式──Ceiling_Locking──作為語言功能提供。
3. 任務優先權的基礎 ── FIFO_Within_Priorities
FIFO_Within_Priorities 是 Ada Annex D 中可以指定的標準優先權式派發策略。未明確指定策略時的預設行為屬於實作定義,但 GNAT 在許多目標平台上都採用這一類策略。同一優先權內以 FIFO(先進先出)方式執行,優先權較高的任務會搶佔(preempt)優先權較低的任務。
-- 01_task_priority.ada
-- 任務優先權與 FIFO_Within_Priorities 的基本形式
-- 組態 pragma 要放在 context 子句之前
pragma Task_Dispatching_Policy (FIFO_Within_Priorities);
with Ada.Text_IO; use Ada.Text_IO;
with System; use System;
with Ada.Real_Time; use Ada.Real_Time;
procedure Task_Priority_Demo is
task High_Priority_Task is
pragma Priority (Priority'Last);
pragma Storage_Size (4 * 1024);
end High_Priority_Task;
task Low_Priority_Task is
pragma Priority (Priority'First);
pragma Storage_Size (4 * 1024);
end Low_Priority_Task;
task body High_Priority_Task is
begin
Put_Line ("[T=0.0s] High priority task started");
delay until Clock + Milliseconds (100);
Put_Line ("[T=0.1s] High priority task completed");
end High_Priority_Task;
task body Low_Priority_Task is
begin
Put_Line ("[T=0.0s] Low priority task started");
delay until Clock + Milliseconds (500);
Put_Line ("[T=0.5s] Low priority task completed");
end Low_Priority_Task;
begin
Put_Line ("=== Task Priority Demo (FIFO_Within_Priorities) ===");
Put_Line ("Main: waiting for tasks to complete...");
delay until Clock + Milliseconds (800);
Put_Line ("Main: done");
end Task_Priority_Demo;
重點:
- 用
pragma Priority為各任務指定靜態優先權。Priority'Last最高,Priority'First最低。 - 這個示範中的任務主體不是繁重運算,而是用
delay until等到指定時刻。這裡想確認的是,當兩者同時變為可執行時,高優先權任務會先取得執行機會。 - 這張圖呈現的是
FIFO_Within_Priorities中不同優先權之間的搶佔面向。若要確認同一優先權內的 FIFO 順序,需要另外用同優先權的多個任務來示範。 - 實際系統中一般會以
System.Default_Priority為基準來設計相對優先權。
sequenceDiagram
participant S as 排程器
participant Main as 主任務
participant HP as 高優先權任務<br/>(Priority=Last)
participant LP as 低優先權任務<br/>(Priority=First)
Main->>HP: 建立任務
Main->>LP: 建立任務
Note over HP,LP: T=0ms: 兩個任務都可執行
S->>HP: 選擇優先權最高的 HP
activate HP
Note over HP: 輸出開始訊息
HP->>S: 因 delay until T+100ms 而阻塞
deactivate HP
S->>LP: 接著執行 LP
activate LP
Note over LP: 輸出開始訊息
LP->>S: 因 delay until T+500ms 而阻塞
deactivate LP
Note over S: T=100ms: HP 喚醒
S->>HP: 執行 HP
activate HP
Note over HP: 輸出完成訊息
deactivate HP
Note over S: T=500ms: LP 喚醒
S->>LP: 執行 LP
activate LP
Note over LP: 輸出完成訊息
deactivate LP
Note over Main: (T=800ms) 主任務結束
Ada 優先權的範圍(GNAT 預設值):
Priority'First = 0 (最低)
Priority'Last = 30 (最高,但依 OS 而異)
4. Ceiling_Locking ── 由語言防止優先權反轉
即時系統中最麻煩的問題之一就是優先權反轉 (priority inversion)。高優先權任務等待低優先權任務持有的鎖,而該低優先權任務又被中優先權任務搶佔,導致高優先權任務被無限期阻塞的現象。
Ada 針對這個問題,將 Ceiling_Locking 協定 直接內建於保護物件中。
-- 02_ceiling_locking.ada
-- 以 Ceiling_Locking 協定防止優先權反轉
-- 組態 pragma 要放在 context 子句之前
pragma Locking_Policy (Ceiling_Locking);
with Ada.Text_IO; use Ada.Text_IO;
with System; use System;
with Ada.Real_Time; use Ada.Real_Time;
procedure Ceiling_Locking_Demo is
Ceiling : constant System.Any_Priority := System.Any_Priority'Last;
protected Shared_Data is
pragma Priority (Ceiling);
procedure Write (V : Integer);
function Read return Integer;
private
Value : Integer := 0;
end Shared_Data;
protected body Shared_Data is
procedure Write (V : Integer) is
begin
Value := V;
end Write;
function Read return Integer is
begin
return Value;
end Read;
end Shared_Data;
task Producer is
pragma Priority (Priority'Last);
pragma Storage_Size (4 * 1024);
end Producer;
task Consumer is
pragma Priority (Priority'First);
pragma Storage_Size (4 * 1024);
end Consumer;
task body Producer is
begin
Put_Line ("[T=0.0s] Producer (high prio): about to write");
Shared_Data.Write (42);
Put_Line ("[T=0.0s] Producer (high prio): write done");
delay until Clock + Milliseconds (100);
end Producer;
task body Consumer is
begin
delay until Clock + Milliseconds (10);
Put_Line ("[T=0.01s] Consumer (low prio): about to read");
declare
V : Integer;
begin
V := Shared_Data.Read;
Put_Line ("[T=0.01s] Consumer (low prio): read done, got" &
Integer'Image (V));
end;
delay until Clock + Milliseconds (100);
end Consumer;
begin
Put_Line ("=== Ceiling_Locking Demo ===");
Put_Line ("Main: producer priority = Last, consumer priority = First");
Put_Line ("Ceiling = Any_Priority'Last, locking = Ceiling_Locking");
delay until Clock + Milliseconds (300);
Put_Line ("Main: done");
end Ceiling_Locking_Demo;
Ceiling_Locking 的運作方式:
- 在保護物件上以
pragma Priority (Ceiling)設定天花板優先權。 - 無論哪個任務進入保護物件,進入時都會自動提升到天花板優先權。
- 因此,使用中的保護物件不會被中優先權任務搶佔。
- 離開保護物件後恢復原本的優先權。
下圖不是前一段範例程式碼的精確時間追蹤,而是概念圖,說明圖 2 的優先權反轉模式在 Ceiling_Locking 下如何被抑制。
sequenceDiagram
participant S as 排程器
participant L as 低優先權任務<br/>(優先權=10)
participant M as 中優先權任務<br/>(優先權=20)
participant H as 高優先權任務<br/>(優先權=30)
participant PO as 保護物件<br/>(天花板優先權=30)
Note over PO: 呼叫端的活動優先權若高於天花板優先權則會 Program_Error<br/>圖中 H(30) 等於天花板(30),因此可以進入
L->>PO: 進入保護操作
activate L
Note over L,PO: 執行優先權提升為 30
Note over S: M 喚醒
Note over S,L: L 以天花板優先權 30 執行中<br/>M(20) 無法搶佔
Note over S: H 喚醒
Note over S,H: H(30) 通過天花板檢查<br/>但因 L 正在使用 PO 而等待
L->>PO: 執行操作
L->>PO: 離開保護操作
deactivate L
Note over L: 優先權恢復為 10
Note over S,H: PO 釋放後執行 H
activate H
H->>PO: 進入保護操作
Note over H,PO: H(30) = 天花板(30),競爭解除後可進入
H->>PO: 離開保護操作
deactivate H
設計原則: 保護物件的天花板優先權,應設定為使用該保護物件的所有任務中最高優先權以上。若違反此原則,讓活動優先權高於天花板優先權的任務呼叫保護操作,Ada 會透過
Program_Error偵測出這個設計錯誤。
在 C 語言中,要用 pthread mutex 達到相同效果,必須明確設定 PTHREAD_PRIO_PROTECT 屬性,但在 Ada 中這是語言的標準功能。
5. delay until ── 讓週期性任務執行時不產生漂移
即時系統的基本模式是週期性任務。在固定間隔重複執行的任務中,防止累積性的計時誤差(漂移)極為重要。
Ada 的 delay until 能優雅地解決這個問題。
-- 03_periodic_task.ada
-- 以 delay until 實現週期性任務 ── 防止累積漂移
with Ada.Text_IO; use Ada.Text_IO;
with System; use System;
with Ada.Real_Time; use Ada.Real_Time;
procedure Periodic_Task_Demo is
Period_MS : constant Time_Span := Milliseconds (200);
Cycles : constant Positive := 5;
task Sensor_Reader is
pragma Priority (Priority'Last - 2);
pragma Storage_Size (4 * 1024);
end Sensor_Reader;
task body Sensor_Reader is
Start_Time : constant Time := Clock;
Next_Release : Time := Start_Time + Period_MS;
Cycle_Count : Natural := 0;
begin
Put_Line ("[Sensor] Periodic task starts, period=" &
To_Duration (Period_MS)'Image & "s, cycles=" &
Natural'Image (Cycles));
for I in 1 .. Cycles loop
delay until Next_Release;
Cycle_Count := Cycle_Count + 1;
Put_Line ("[Sensor] Cycle" & Natural'Image (Cycle_Count) &
" at" & Duration'Image (To_Duration (Clock - Start_Time)) & "s");
Next_Release := Next_Release + Period_MS;
end loop;
Put_Line ("[Sensor] Periodic task finished. Actual elapsed:" &
Duration'Image (To_Duration (Clock - Start_Time)) & "s");
end Sensor_Reader;
begin
Put_Line ("=== Periodic Task Demo (delay until) ===");
Put_Line ("Main: waiting for" & Natural'Image (Cycles) & " cycles...");
delay until Clock + Milliseconds (1500);
Put_Line ("Main: done");
end Periodic_Task_Demo;
為什麼要用 delay until:
| 方法 | 問題 |
|---|---|
delay Period; |
每次迴圈的處理時間會累加,導致週期逐漸偏移(累積漂移) |
delay until Next_Release; Next_Release := Next_Release + Period; |
以絕對時刻為基準,即使某一次處理延遲,下一次的啟動時刻依然正確 |
不過,delay until 並不會自動保證處理時間必定落在週期以內。如果處理時間超過下一次應該喚醒的時刻,該次 delay until 幾乎會立即返回,系統就進入應視為 deadline miss 的狀態。
delay 的情況:
T=0ms → 處理(15ms) → delay 100ms → T=115ms → 處理(10ms) → ...
實際間隔:115ms, 110ms, ...(處理時間累積)
delay until 的情況:
Next_Release: 100ms, 200ms, 300ms, ...(絕對時刻)
T=0ms → 處理(15ms) → delay until 100ms → T=100ms → 處理(10ms) → delay until 200ms
實際間隔:100ms, 100ms, ...(不受處理時間影響)
這種 delay until 模式,之後所有週期性任務都會沿用。
flowchart TB
subgraph Bad["delay Period - 累積漂移"]
B1[T=0ms: 計算15ms] --> B2[delay 100ms → 於115ms喚醒]
B2 --> B3[計算10ms → 125ms]
B3 --> B4[delay 100ms → 於225ms喚醒]
B4 --> B5[實際間隔: 115ms, 110ms...]
end
subgraph Good["delay until - 以絕對時刻為基準"]
G1[下一次 = T+100ms] --> G2[計算15ms]
G2 --> G3[delay until T+100ms → 於100ms喚醒]
G3 --> G4[計算10ms]
G4 --> G5[下一次 = T+200ms → 於200ms喚醒]
G5 --> G6[實際間隔: 100ms, 100ms...]
end
subgraph Overrun["週期超時 - deadline miss"]
O1[下一次 = T+100ms] --> O2[計算130ms]
O2 --> O3[delay until T+100ms 立即返回]
O3 --> O4[偵測延遲並視為過載]
end
Bad --> Drift[誤差隨時間累積]
Good --> Stable[防止累積漂移]
Good --> Overrun
6. Ravenscar 設定檔 ── 可驗證的即時子集合
Ada 的任務功能雖然強大,但在安全性極為重要的系統中,「太過強大」反而成為問題。動態任務建立、select 陳述式、abort 陳述式等,都會讓最壞執行時間的靜態分析變得困難。
Ravenscar 設定檔 正是 Ada 針對這類問題提出的解答,它將任務功能限制在可靜態分析、具決定性的子集合內。
-- 04_ravenscar_profile.ada
-- Ravenscar 設定檔的基本形式
-- 編譯時需在 gnat.adc 中指定 pragma Profile (Ravenscar);
with Ada.Text_IO; use Ada.Text_IO;
with System; use System;
with Ada.Real_Time; use Ada.Real_Time;
package Ravenscar_State is
protected Signal is
pragma Priority (System.Default_Priority + 5);
entry Wait_For_Release;
procedure Release;
private
Released : Boolean := False;
end Signal;
task Periodic_Worker is
pragma Priority (System.Default_Priority + 1);
pragma Storage_Size (4 * 1024);
end Periodic_Worker;
task Monitor is
pragma Priority (System.Default_Priority);
pragma Storage_Size (4 * 1024);
end Monitor;
end Ravenscar_State;
with Ada.Text_IO; use Ada.Text_IO;
with System; use System;
with Ada.Real_Time; use Ada.Real_Time;
package body Ravenscar_State is
protected body Signal is
entry Wait_For_Release when Released is
begin
Released := False;
end Wait_For_Release;
procedure Release is
begin
Released := True;
end Release;
end Signal;
task body Periodic_Worker is
Start_Time : constant Time := Clock;
Next_Release : Time := Start_Time + Milliseconds (100);
Period : constant Time_Span := Milliseconds (100);
Cycle_Count : Natural := 0;
begin
Put_Line ("[Worker] Ravenscar periodic task starts");
for I in 1 .. 4 loop
delay until Next_Release;
Cycle_Count := Cycle_Count + 1;
Put_Line ("[Worker] Cycle" & Natural'Image (Cycle_Count) &
" at" & Duration'Image (To_Duration (Clock - Start_Time)) & "s");
Signal.Release;
Next_Release := Next_Release + Period;
end loop;
Put_Line ("[Worker] Finished demo, waiting (Ravenscar: No_Task_Termination)");
loop
delay until Clock + Seconds (1);
end loop;
end Periodic_Worker;
task body Monitor is
begin
Put_Line ("[Monitor] Waiting for signals...");
for I in 1 .. 4 loop
Signal.Wait_For_Release;
Put_Line ("[Monitor] Received signal" & Natural'Image (I));
end loop;
Put_Line ("[Monitor] All signals received, waiting (Ravenscar: No_Task_Termination)");
loop
delay until Clock + Seconds (1);
end loop;
end Monitor;
end Ravenscar_State;
with Ravenscar_State; use Ravenscar_State;
with Ada.Text_IO; use Ada.Text_IO;
with Ada.Real_Time; use Ada.Real_Time;
procedure Ravenscar_Demo is
begin
Put_Line ("=== Ravenscar Profile Demo ===");
Put_Line ("(compile with: gnatmake -gnatec=gnat.adc ravenscar_demo)");
Put_Line ("Main: waiting for Ravenscar tasks...");
delay until Clock + Milliseconds (800);
Put_Line ("Main: demo window elapsed; waiting forever (Ravenscar: No_Task_Termination)");
loop
delay until Clock + Seconds (1);
end loop;
end Ravenscar_Demo;
Ravenscar 設定檔的限制:
| 被禁止的功能 | 原因 |
|---|---|
動態任務建立(new 或存取型別) |
執行期的記憶體配置不具決定性 |
select 陳述式 |
不只多重替代方案,整個 select 陳述式都會讓控制流程分析變困難 |
abort 陳述式 |
非同步中斷會造成狀態不可預測 |
Ada.Task_Attributes |
執行期動態行為 |
| 動態變更優先權 | 排程分析的前提會在執行期改變 |
相對 delay(delay) |
容易產生累積漂移,因此使用絕對時刻的 delay until |
| 每個保護物件多個入口(entry) | 會增加阻塞條件與分析對象 |
| 任務終止 | 在 Ravenscar 中所有任務都視為不會終止 |
requeue 陳述式 |
使控制流程追蹤更複雜 |
透過這些限制,符合 Ravenscar 規範的程式會變成容易進行靜態時序分析的形式。這正是 DO-178C(航空軟體)、ISO 26262(汽車功能安全)等安全標準所要求的特性。下方列表只是主要限制的摘要,實際的設定檔還包含 No_Task_Hierarchy、Detect_Blocking 等與執行環境及可分析性相關的額外規則。
flowchart TB
Full[完整的 Ada 任務功能] --> Profile[Ravenscar 設定檔]
Profile --> Restrict[限制]
Profile --> Policy[必要政策]
Restrict --> R1[禁止動態任務建立]
Restrict --> R2[禁止 select 陳述式]
Restrict --> R3[禁止 abort 陳述式]
Restrict --> R4[禁止 Task_Attributes]
Restrict --> R5[限制每個保護物件僅 1 個入口]
Restrict --> R6[禁止 requeue 陳述式]
Restrict --> R7[禁止相對delay<br/>使用delay until]
Restrict --> R8[禁止動態變更優先權]
Restrict --> R9[禁止任務終止<br/>所有任務不終止]
Policy --> P1[FIFO_Within_Priorities]
Policy --> P2[Ceiling_Locking]
Restrict --> Benefit[變得容易進行:<br/>靜態時序分析]
Policy --> Benefit
Benefit --> DO178[DO-178C<br/>航空軟體]
Benefit --> ISO26262[ISO 26262<br/>汽車功能安全]
Benefit --> IEC62304[IEC 62304<br/>醫療器材軟體]
要啟用 Ravenscar 設定檔,需要在 gnat.adc 檔案中寫入以下內容:
pragma Profile (Ravenscar);
7. 計時事件 ── 不需輪詢的時刻驅動喚醒
許多即時系統經常需要「到了指定時刻就喚醒高優先權任務」這類需求。單純的做法會用輪詢計時器,但 Ada 提供了更精緻的機制──計時事件。
-- 05_timing_events.ada
-- 計時事件 (Ada.Real_Time.Timing_Events)
-- 不需輪詢即可喚醒高優先權任務的機制
pragma Locking_Policy (Ceiling_Locking);
with Ada.Text_IO; use Ada.Text_IO;
with System; use System;
with Ada.Real_Time; use Ada.Real_Time;
with Ada.Real_Time.Timing_Events; use Ada.Real_Time.Timing_Events;
package Signal_Pkg is
protected type Signal_Type is
pragma Priority (System.Interrupt_Priority'Last);
entry Wait_For_Event;
procedure Fire (Event : in out Timing_Event);
private
Fired : Boolean := False;
end Signal_Type;
S : Signal_Type;
end Signal_Pkg;
with Ada.Text_IO; use Ada.Text_IO;
with System; use System;
with Ada.Real_Time; use Ada.Real_Time;
with Ada.Real_Time.Timing_Events; use Ada.Real_Time.Timing_Events;
package body Signal_Pkg is
protected body Signal_Type is
entry Wait_For_Event when Fired is
begin
Fired := False;
end Wait_For_Event;
procedure Fire (Event : in out Timing_Event) is
begin
Fired := True;
end Fire;
end Signal_Type;
end Signal_Pkg;
with Signal_Pkg; use Signal_Pkg;
with Ada.Text_IO; use Ada.Text_IO;
with System; use System;
with Ada.Real_Time; use Ada.Real_Time;
with Ada.Real_Time.Timing_Events; use Ada.Real_Time.Timing_Events;
procedure Timing_Events_Demo is
pragma Priority (29);
Timer_1 : Timing_Event;
Timer_2 : Timing_Event;
task Reactor is
pragma Priority (System.Default_Priority + 5);
pragma Storage_Size (4 * 1024);
end Reactor;
task body Reactor is
begin
Put_Line ("[Reactor] Waiting for timing events...");
S.Wait_For_Event;
Put_Line ("[Reactor] Got event #1");
S.Wait_For_Event;
Put_Line ("[Reactor] Got event #2");
Put_Line ("[Reactor] Done");
end Reactor;
begin
Put_Line ("=== Timing Events Demo ===");
Put_Line ("Scheduling two timers at +100ms and +250ms...");
Set_Handler (Timer_1, Clock + Milliseconds (100), S.Fire'Access);
Set_Handler (Timer_2, Clock + Milliseconds (250), S.Fire'Access);
delay until Clock + Milliseconds (500);
Put_Line ("Main: done");
end Timing_Events_Demo;
計時事件的運作方式:
1. Set_Handler(Timer_1, T+100ms, S.Fire'Access) ── 在絕對時刻登錄處理器
2. 經過 T+100ms ── 執行環境以**天花板優先權**呼叫 S.Fire
3. Fire 將 Fired 旗標設為 True ── 屏障開啟
4. Reactor 任務從 Wait_For_Event 中喚醒
重點在於,這個範例明確指定了 Ceiling_Locking,且 Fire 處理器是保護物件的程序,因此以天花板優先權執行。作為計時事件處理器使用的保護程序,應放在具有中斷層級天花板優先權(此處為 System.Interrupt_Priority'Last)的保護物件中。這樣一來,處理計時事件時就不會發生優先權反轉。
8. 以保護物件實作即時佇列
即時系統中常見的模式是生產者消費者。感測器產生資料,控制任務消費資料──這時需要有效地處理緩衝區的互斥控制與阻塞。
只要使用 Ada 的保護物件與入口屏障,就能實作以屏障為基礎的同步機制。內部由執行環境管理互斥,因此應用程式碼不需要直接撰寫 mutex 或條件變數。
-- 06_protected_queue.ada
-- 以保護物件實現即時資料共享
-- 管線: Producer -> Bounded_Buffer -> Consumer
-- 編譯時需在 gnat.adc 中指定 pragma Locking_Policy (Ceiling_Locking);
pragma Locking_Policy (Ceiling_Locking);
with Ada.Text_IO; use Ada.Text_IO;
with System; use System;
with Ada.Real_Time; use Ada.Real_Time;
procedure Protected_Queue_Demo is
Buffer_Size : constant := 4;
type Buf_Array is array (1 .. Buffer_Size) of Integer;
protected Bounded_Buffer is
pragma Priority (System.Any_Priority'Last);
entry Put (Item : Integer);
entry Get (Item : out Integer);
private
Buf : Buf_Array;
Count : Natural := 0;
Head : Positive := 1;
Tail : Positive := 1;
end Bounded_Buffer;
protected body Bounded_Buffer is
entry Put (Item : Integer) when Count < Buffer_Size is
begin
Buf (Tail) := Item;
Tail := (Tail mod Buffer_Size) + 1;
Count := Count + 1;
end Put;
entry Get (Item : out Integer) when Count > 0 is
begin
Item := Buf (Head);
Head := (Head mod Buffer_Size) + 1;
Count := Count - 1;
end Get;
end Bounded_Buffer;
task Producer is
pragma Priority (System.Default_Priority + 2);
pragma Storage_Size (4 * 1024);
end Producer;
task Consumer is
pragma Priority (System.Default_Priority + 1);
pragma Storage_Size (4 * 1024);
end Consumer;
task body Producer is
Next_Release : Time := Clock + Milliseconds (50);
Period : constant Time_Span := Milliseconds (50);
begin
for I in 1 .. 6 loop
Bounded_Buffer.Put (I);
Put_Line ("[Producer] Put" & Integer'Image (I));
delay until Next_Release;
Next_Release := Next_Release + Period;
end loop;
Put_Line ("[Producer] Done");
end Producer;
task body Consumer is
Item : Integer;
Next_Release : Time := Clock + Milliseconds (80);
Period : constant Time_Span := Milliseconds (80);
begin
delay until Clock + Milliseconds (30);
for I in 1 .. 6 loop
Bounded_Buffer.Get (Item);
Put_Line ("[Consumer] Got" & Integer'Image (Item));
delay until Next_Release;
Next_Release := Next_Release + Period;
end loop;
Put_Line ("[Consumer] Done");
end Consumer;
begin
Put_Line ("=== Protected Queue Demo (Ceiling_Locking) ===");
Put_Line ("Buffer size = 4; Producer every 50ms, Consumer every 80ms");
delay until Clock + Milliseconds (800);
Put_Line ("Main: done");
end Protected_Queue_Demo;
設計要點:
entry Put when Count < Buffer_Size── 緩衝區滿時,Producer 會自動阻塞。entry Get when Count > 0── 緩衝區空時,Consumer 會自動阻塞。pragma Priority (System.Any_Priority'Last)── 藉由天花板鎖定,Producer 與 Consumer 之間不會發生優先權反轉。- 屏障條件由保護物件的內部狀態(
Count)定義,並在鎖釋放時自動重新評估。
這段程式碼中完全沒有出現應用層的 mutex、semaphore 或條件變數。所有需要的等待,都以保護物件的入口屏障來表示。
stateDiagram-v2
Empty: 空 / Count=0
Partial: 部分 / Count=1..Buffer_Size-1
Full: 滿 / Count=Buffer_Size
[*] --> Empty: 初始狀態
Empty --> Partial: Put (新增1個元素)
Partial --> Partial: Put / Get
Partial --> Empty: Get (取出最後一個元素)
Partial --> Full: Put (填滿最後的空位)
Full --> Partial: Get (產生空位)
Empty --> Empty: Get 會阻塞 (屏障 Count=0)
Full --> Full: Put 會阻塞 (屏障 Count=Buffer_Size)
Put 成功時會重新評估 Get 的等待屏障,Get 成功時則重新評估 Put 的等待屏障。無論圖中處於哪個狀態,這都會在保護操作完成時執行。
9. 執行時間量測 ── 執行時間監控的第一步
要評估即時系統的可排程性,必須準確掌握每個任務的執行時間(CPU 時間)。Ada 的 Ada.Execution_Time 套件提供了以任務為單位的 CPU 消耗時間。
-- 07_execution_time.ada
-- 執行時間控制 (Execution_Time)
-- 量測每個任務的 CPU 消耗時間
with Ada.Text_IO; use Ada.Text_IO;
with System; use System;
with Ada.Real_Time; use Ada.Real_Time;
with Ada.Execution_Time;
use type Ada.Execution_Time.CPU_Time;
procedure Execution_Time_Demo is
package ET renames Ada.Execution_Time;
task Busy_Worker is
pragma Priority (System.Default_Priority + 1);
pragma Storage_Size (4 * 1024);
end Busy_Worker;
task body Busy_Worker is
Wall_Start : Time;
Cpu_Start : ET.CPU_Time;
Dummy : Integer := 0;
pragma Volatile (Dummy);
begin
Wall_Start := Clock;
Cpu_Start := ET.Clock;
Put_Line ("[Worker] Starting compute-bound work...");
for I in 1 .. 20_000_000 loop
Dummy := Dummy + 1;
end loop;
Put_Line ("[Worker] Dummy =" & Integer'Image (Dummy));
declare
Wall_Elapsed : constant Duration :=
To_Duration (Clock - Wall_Start);
Cpu_Span : constant Time_Span :=
ET.Clock - Cpu_Start;
begin
Put_Line ("[Worker] Done, wall time:" &
Duration'Image (Wall_Elapsed) & "s");
Put_Line ("[Worker] CPU time consumed:" &
Duration'Image (To_Duration (Cpu_Span)) & "s");
end;
end Busy_Worker;
Cpu_Start_Main : constant ET.CPU_Time := ET.Clock;
begin
Put_Line ("=== Execution Time Demo ===");
delay until Clock + Milliseconds (500);
declare
Cpu_Span : constant Time_Span := ET.Clock - Cpu_Start_Main;
begin
Put_Line ("Main: CPU time consumed after 500ms:" &
Duration'Image (To_Duration (Cpu_Span)) & "s");
end;
Put_Line ("Main: done");
end Execution_Time_Demo;
牆上時間 vs CPU 時間:
牆上時間 (Wall Clock): Ada.Real_Time.Clock
→ 實際經過的時間,包含阻塞中與被搶佔期間。
CPU 時間 (Execution Time): Ada.Execution_Time.Clock
→ 該任務實際在 CPU 上執行的時間。
→ 阻塞中、被搶佔期間不計入。
這個區分,是執行時間監控與 WCET 驗證的出發點。Busy_Worker 在 delay until 等待期間 CPU 時間不會增加,只有在實際計算處理中才會增加。主任務的 delay until Clock + Milliseconds(500) 期間,CPU 時間應該也幾乎為零。不過,CPU 時間的實測並不能保證真正的 WCET。包含快取、管線化、記憶體競爭等因素的 WCET,需要另外用靜態分析或在目標環境上進行驗證。
flowchart LR
subgraph Wall[牆上時間]
W1[經過時間總計: 500ms] --> W2[內容: 計算 + 等待 + 阻塞 + 被搶佔]
end
subgraph CPU[CPU時間]
C1[CPU時間總計: 120ms] --> C2[內容: 僅實際計算]
end
Wall --> Diff[差值 = 等待・阻塞・被搶佔時間]
CPU --> Diff
Diff --> Insight[CPU時間可觀察實際計算成本<br/>有助於WCET驗證與監控<br/>排除等待・阻塞・被搶佔時間]
Insight --> Caveat[注意<br/>實測不保證真正的WCET<br/>需要靜態分析或目標環境驗證]
10. 綜合示範 ── 多週期即時系統
整合前面學到的所有元素──優先權、Ceiling_Locking、delay until、保護物件──建構一個典型的多週期即時系統。
-- 08_multiperiodic.ada
-- 多週期即時系統整合示範
-- 快週期(100ms)的感測器讀取任務
-- 慢週期(400ms)的控制任務
-- 以 Ceiling_Locking 共享資料
pragma Locking_Policy (Ceiling_Locking);
with Ada.Text_IO; use Ada.Text_IO;
with System; use System;
with Ada.Real_Time; use Ada.Real_Time;
procedure Multiperiodic_Demo is
package Int_IO is new Ada.Text_IO.Integer_IO (Integer);
protected Shared_Sensor is
pragma Priority (System.Any_Priority'Last);
procedure Write (V : Integer);
function Read return Integer;
private
Value : Integer := 0;
end Shared_Sensor;
protected body Shared_Sensor is
procedure Write (V : Integer) is
begin
Value := V;
end Write;
function Read return Integer is
begin
return Value;
end Read;
end Shared_Sensor;
task Fast_Sensor is
pragma Priority (System.Default_Priority + 3);
pragma Storage_Size (4 * 1024);
end Fast_Sensor;
task body Fast_Sensor is
Next_Release : Time := Clock + Milliseconds (100);
Period : constant Time_Span := Milliseconds (100);
Cycle : Natural := 0;
begin
Put_Line ("[Fast] Sensor reader starts (100ms period)");
for I in 1 .. 12 loop
delay until Next_Release;
Cycle := Cycle + 1;
Shared_Sensor.Write (Cycle * 10);
Next_Release := Next_Release + Period;
end loop;
Put_Line ("[Fast] Done");
end Fast_Sensor;
task Slow_Controller is
pragma Priority (System.Default_Priority + 2);
pragma Storage_Size (4 * 1024);
end Slow_Controller;
task body Slow_Controller is
Next_Release : Time := Clock + Milliseconds (150);
Period : constant Time_Span := Milliseconds (400);
Cycle : Natural := 0;
Raw : Integer;
begin
Put_Line ("[Slow] Controller starts (400ms period)");
for I in 1 .. 3 loop
delay until Next_Release;
Cycle := Cycle + 1;
Raw := Shared_Sensor.Read;
Put_Line ("[Slow] Cycle" & Natural'Image (Cycle) &
" reads sensor =" & Integer'Image (Raw));
Next_Release := Next_Release + Period;
end loop;
Put_Line ("[Slow] Done");
end Slow_Controller;
begin
Put_Line ("=== Multiperiodic Real-Time System Demo ===");
Put_Line ("Fast sensor (100ms) x 12 + Slow controller (400ms) x 3");
Put_Line ("Ceiling_Locking prevents priority inversion on shared data");
delay until Clock + Milliseconds (2000);
Put_Line ("Main: done");
end Multiperiodic_Demo;
系統架構:
下圖是以範例程式碼的 release 時刻(快速感測器為 100ms 週期,慢速控制為偏移 150ms 的 400ms 週期)為基礎,並假設一組示範用執行時間所畫出的排程範例。由於程式碼本身並沒有 80ms 的控制運算,因此這不是實測圖。當快速感測器的優先權較高時,若快速感測器的 release 出現在慢速控制執行期間,慢速控制就會暫時被中斷。為了方便閱讀,圖中每個慢速控制週期只畫出一次代表性的中斷,但實際上快速感測器會在每個 100ms 邊界被 release。
flowchart TB
Assumption["示範用假設<br/>快速感測器: 處理10ms<br/>慢速控制: 處理80ms"]
subgraph Cycle1["慢速控制 週期1(release=150ms)"]
direction LR
C1F1["100-110ms<br/>快速感測器 #1"] --> C1S1["150-200ms<br/>慢速控制 #1 前半"]
C1S1 --> C1F2["200-210ms<br/>快速感測器 #2<br/>因P+3而中斷"]
C1F2 --> C1S2["210-240ms<br/>慢速控制 #1 後半"]
end
subgraph Cycle2["慢速控制 週期2(release=550ms)"]
direction LR
C2S1["550-600ms<br/>慢速控制 #2 前半"] --> C2F6["600-610ms<br/>快速感測器 #6<br/>因P+3而中斷"]
C2F6 --> C2S2["610-640ms<br/>慢速控制 #2 後半"]
end
subgraph Cycle3["慢速控制 週期3(release=950ms)"]
direction LR
C3S1["950-1000ms<br/>慢速控制 #3 前半"] --> C3F10["1000-1010ms<br/>快速感測器 #10<br/>因P+3而中斷"]
C3F10 --> C3S2["1010-1040ms<br/>慢速控制 #3 後半"]
end
Assumption --> C1F1
C1S2 --> C2S1
C2S2 --> C3S1
這種模式,是工業控制系統或機器人控制中常見的「快速感測器收集 + 慢速控制迴路」典型結構。
11. Ada 即時功能大放異彩的場景
Ada 的即時功能特別能在以下領域發揮價值。
flowchart TB
Ada[Ada Annex D<br/>即時功能] --> Aero[航太<br/>DO-178C]
Ada --> Rail[鐵路<br/>EN 50128系列]
Ada --> Auto[汽車<br/>ISO 26262]
Ada --> Medical[醫療器材<br/>IEC 62304]
Ada --> Industrial[工業控制<br/>IEC 61508系列]
Ada --> Defense[國防・高可靠性系統]
Aero --> A1[飛行控制<br/>應用實績豐富的領域]
Aero --> A2[衛星・太空載具控制]
Rail --> R1[號誌系統]
Rail --> R2[自動列車控制]
Auto --> Au1[安全相關ECU的候選方案]
Auto --> Au2[在以C / MISRA-C為主流的領域中<br/>有限且選擇性的應用]
Medical --> M1[心律調節器]
Medical --> M2[輸液泵]
Industrial --> I1[機器人控制]
Industrial --> I2[數控機床]
Defense --> D1[任務電腦]
Defense --> D2[長期運作系統]
12. 注意事項與限制
Ada 的即時功能雖然強大,但並非萬能。
1. 平台相依性:
pragma Priority實際對應到的行為取決於執行環境(OS + GNAT 執行環境)。在 Linux 上會對應到SCHED_FIFO,但在 Windows 上,完全的搶佔式排程未必能得到保證。
2. Ravenscar 的限制:
- 由於禁止動態任務建立,所有任務必須在系統啟動時就靜態宣告完成。這會限制設計上的自由度。
3. WCET 量測的限制:
Ada.Execution_Time提供的是量測,而不是保證。包含快取未中、管線化風險等因素的真正 WCET,需要另外用靜態分析工具驗證。
4. 額外負擔:
- 保護物件的屏障評估,會在入口完成或取消時,以及離開保護物件時自動執行。若保護物件被高頻率呼叫,就需要考慮這部分的額外負擔。
5. 工具鏈的門檻:
- 要完全發揮 Ada 即時功能的價值,需要合適的跨平台編譯器與執行環境。尤其在嵌入式目標平台上,往往要依賴廠商提供的執行環境。
13. 總結
本文透過 8 個程式碼範例,逐步檢視了 Ada 的 Annex D 所提供的即時功能。
| 功能 | 提供的價值 |
|---|---|
| 任務優先權 | 搶佔式的優先權基礎排程 |
| Ceiling_Locking | 語言內建的優先權反轉防止機制 |
delay until |
防止累積漂移的週期執行 |
| Ravenscar 設定檔 | 容易進行靜態分析的任務子集合 |
| 計時事件 | 不需輪詢的時刻驅動喚醒 |
| 保護佇列 | 保護物件的屏障式同步 |
| 執行時間量測 | 以任務為單位的 CPU 時間監控 |
| 多週期整合 | 讓不同週期的任務安全共存的設計 |
Ada 即時功能的本質在於「不是事後添加的」。防止優先權反轉的鎖定規則、用於週期執行的時刻指定、執行時間監控等,都作為語言規範的一部分提供。當然,能否真正達成期限,仍要靠設計與分析來確認,但語言執行環境已經替你準備好了所需的前提條件,這正是最大的優勢所在。
mindmap
root((Ada Annex D<br/>即時系統))
排程
FIFO_Within_Priorities
任務優先權
搶佔
優先權反轉防止
Ceiling_Locking 協定
自動優先權提升
天花板優先權規則
週期執行
delay until
防止累積漂移
以絕對時刻為基準
Ravenscar 設定檔
靜態任務集合
決定性分析
禁止相對delay
DO-178C / ISO 26262
計時事件
不需輪詢的喚醒
登錄保護處理器
處理器以天花板優先權執行
保護物件/佇列
入口屏障
空/部分/滿的狀態管理
屏障式同步
執行時間監控
各任務CPU時間
牆上時間 vs CPU時間
有助於WCET驗證・監控
整合設計
多週期設計
以屏障保護
語言層級的安全性
接下來,如果想實際嘗試用 Ada 開發即時系統,可以到 Alire 安裝 GNAT 工具鏈,並用 gnatchop + gnatmake 建置本文的範例程式碼。
另外,關於 Ada 並行處理的基礎(任務、會合、保護物件),請參考上一篇文章「Ada 的安全並行處理」。
14. 參考資料
相關文章
共用相同標籤的最新文章。能以相近的主題延伸理解。
Ada 的安全並行處理 ── 任務與保護物件實戰指南
本文是 Ada 語言內建並行處理機制──任務(task)與保護物件(protected object)的實戰入門文章。內容涵蓋會合(entry/accept)、選擇性接受(select)、生產者消費者模式、保護物件實現的排他控制、帶屏障的保護入口、附逾時的呼叫,以及任務優先...
Ada 語言的魅力 —— 用型別闡述設計理念,支撐運作數十年軟體的語言
介紹 Ada 語言的魅力。從強型別、範圍限制、以套件分離規格與實作、契約式設計、內建於語言的工作(task)、SPARK 形式驗證,到 GNAT 與 Alire 開發環境,整理支撐高可靠軟體的設計理念。
Ada 的泛型程式設計 ── 以型別撰寫契約,零成本達成程式碼重用
系統性說明 Ada 的泛型程式設計,涵蓋泛型子程式、泛型套件、形式子程式參數、型別類別、正式套件參數,以及實務上的設計方針。搭配 Mermaid 圖與實例,整理型別安全的重用與零成本抽象化的思路。
SPARK形式驗證入門 ── 從Ada的契約到數學證明
本文是使用Ada子集語言SPARK進行形式驗證的實務入門文章,整理從契約(Pre/Post)進階到證明的方法、GNATprove的使用方式、迴圈不變條件、資料流契約、證明等級,以及在實際專案中導入的方法。
用 PerfView 與 dotnet-trace 找出「變慢」的原因 ── .NET 效能調查實務入門
當商用應用程式「變慢」「CPU 卡住不動」「偶爾當掉」時,該用哪個工具看什麼?本文整理 PerfView 與 dotnet-trace 的角色分工、CPU 取樣的讀法(inclusive/exclusive)、用 ThreadTime 調查阻塞時間,以及與 EventSou...
相關主題
與本文相近的主題頁面。以本文為起點,可進一步連到相關服務與其他文章。
Windows 技術主題
彙整 KomuraSoft LLC 關於 Windows 開發、故障調查與既有資產活用文章的主題中心。
常見問題
整理諮詢這個主題時常見的問題。
- Ada 的 Annex D 是什麼?
- 這是作為 Ada 語言規範一部分而標準化的一組即時系統功能。內容包括 FIFO_Within_Priorities 提供的優先權式搶佔排程、防止優先權反轉的 Ceiling_Locking 協定、以 delay until 進行絕對時刻週期執行、Ravenscar 設定檔、計時事件,以及透過 Ada.Execution_Time 依任務量測執行時間等。其特點在於這些機制不是後加的函式庫,而是直接內建於語言執行環境本身。
- 什麼是優先權反轉?Ada 如何防止它?
- 這是指低優先權任務持有鎖時被中優先權任務搶佔,導致等待該鎖的高優先權任務被無限期阻塞的現象。1997 年的 Mars Pathfinder 探測器就曾實際發生這個問題,導致探測器不斷重新啟動。Ada 將 Ceiling_Locking 協定作為語言功能提供:進入保護物件的任務會自動提升到天花板優先權,因此可以防止中優先權任務造成的搶佔。
- Ravenscar 設定檔是什麼?
- 這是為安全性極為重要的系統設計的設定檔,將 Ada 的任務功能限制在可靜態分析、具決定性的子集合中。動態任務建立、select 陳述式、abort 陳述式、相對 delay、requeue 陳述式等都被禁止。透過這些限制,可以更容易進行靜態時序分析,也更容易滿足 DO-178C(航空軟體)、ISO 26262(汽車功能安全)等安全標準所要求的特性。在 GNAT 中,於 gnat.adc 檔案寫入 pragma Profile (Ravenscar) 即可啟用。
- 在週期性任務中,為什麼用 delay until 而不是 delay?
- 因為使用相對時間的 delay 會讓每次迴圈的處理時間累加起來,導致週期逐漸偏移,產生累積漂移。delay until 是以絕對時刻為基準決定下一次執行時刻,因此即使某一次處理延遲,之後的執行時刻仍能保持正確。不過,如果處理時間超過下一次應該喚醒的時刻,delay until 幾乎會立即返回,因此需要另外設計偵測 deadline miss 並視為過載的機制。
作者檔案
本文作者的個人檔案頁面。
Go Komura
小村軟體有限公司 代表
以 Windows 軟體開發、技術諮詢與故障調查為中心,在難以重現的故障調查與既有資產仍在運作的專案上具有優勢。