用 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 的即時功能,正是在語言層級提供一套容易進行這類分析的可預測執行模型

即時性要求超過期限 = 致命失敗偶爾超過可接受期限應完成的絕對時刻週期重複間隔WCET最壞執行時間抖動週期的變動硬即時飛行控制安全氣囊心律調節器軟即時影音傳輸遊戲UIAda Annex D支撐可預測性的機制FIFO_Within_PrioritiesCeiling_Lockingdelay until可排程性分析必要條件: WCET &lt;= 期限充分性須以回應時間分析確認

即時系統中最危險的現象之一是優先權反轉。這個問題曾在 1997 年的 Mars Pathfinder 上實際發生,造成探測器不斷重新啟動。

共享資源中優先權任務高優先權任務低優先權任務排程器共享資源中優先權任務高優先權任務低優先權任務排程器執行關鍵區段中H 喚醒,排程器中斷 L因等待鎖而阻塞!(L 持有中)H 在等待鎖,因此恢復執行 L繼續朝釋放鎖前進...M 喚醒,排程器中斷 LL 無法釋放鎖M 持續執行(H 與 L 都動不了)【優先權反轉】高優先權被無限期阻塞取得鎖嘗試取得鎖

低優先權任務持有鎖時被中優先權任務搶佔,導致高優先權任務被無限期阻塞。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 為基準來設計相對優先權。
低優先權任務(Priority=First)高優先權任務(Priority=Last)主任務排程器低優先權任務(Priority=First)高優先權任務(Priority=Last)主任務排程器T=0ms: 兩個任務都可執行輸出開始訊息輸出開始訊息T=100ms: HP 喚醒輸出完成訊息T=500ms: LP 喚醒輸出完成訊息(T=800ms) 主任務結束建立任務建立任務選擇優先權最高的 HP因 delay until T+100ms 而阻塞接著執行 LP因 delay until T+500ms 而阻塞執行 HP執行 LP
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 的運作方式:

  1. 在保護物件上以 pragma Priority (Ceiling) 設定天花板優先權
  2. 無論哪個任務進入保護物件,進入時都會自動提升到天花板優先權
  3. 因此,使用中的保護物件不會被中優先權任務搶佔。
  4. 離開保護物件後恢復原本的優先權。

下圖不是前一段範例程式碼的精確時間追蹤,而是概念圖,說明圖 2 的優先權反轉模式在 Ceiling_Locking 下如何被抑制。

保護物件(天花板優先權=30)高優先權任務(優先權=30)中優先權任務(優先權=20)低優先權任務(優先權=10)排程器保護物件(天花板優先權=30)高優先權任務(優先權=30)中優先權任務(優先權=20)低優先權任務(優先權=10)排程器呼叫端的活動優先權若高於天花板優先權則會 Program_Error圖中 H(30) 等於天花板(30),因此可以進入執行優先權提升為 30M 喚醒L 以天花板優先權 30 執行中M(20) 無法搶佔H 喚醒H(30) 通過天花板檢查但因 L 正在使用 PO 而等待優先權恢復為 10PO 釋放後執行 HH(30) = 天花板(30),競爭解除後可進入進入保護操作執行操作離開保護操作進入保護操作離開保護操作

設計原則: 保護物件的天花板優先權,應設定為使用該保護物件的所有任務中最高優先權以上。若違反此原則,讓活動優先權高於天花板優先權的任務呼叫保護操作,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 模式,之後所有週期性任務都會沿用。

週期超時 - deadline miss計算130ms下一次 = T+100msdelay until T+100ms 立即返回偵測延遲並視為過載delay until - 以絕對時刻為基準計算15ms下一次 = T+100msdelay until T+100ms → 於100ms喚醒計算10ms下一次 = T+200ms → 於200ms喚醒實際間隔: 100ms, 100ms...delay Period - 累積漂移delay 100ms → 於115ms喚醒T=0ms: 計算15ms計算10ms → 125msdelay 100ms → 於225ms喚醒實際間隔: 115ms, 110ms...誤差隨時間累積防止累積漂移

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_HierarchyDetect_Blocking 等與執行環境及可分析性相關的額外規則。

完整的 Ada 任務功能Ravenscar 設定檔限制必要政策禁止動態任務建立禁止 select 陳述式禁止 abort 陳述式禁止 Task_Attributes限制每個保護物件僅 1 個入口禁止 requeue 陳述式禁止相對delay使用delay until禁止動態變更優先權禁止任務終止所有任務不終止FIFO_Within_PrioritiesCeiling_Locking變得容易進行:靜態時序分析DO-178C航空軟體ISO 26262汽車功能安全IEC 62304醫療器材軟體

要啟用 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 或條件變數。所有需要的等待,都以保護物件的入口屏障來表示。

初始狀態Put (新增1個元素)Put / GetGet (取出最後一個元素)Put (填滿最後的空位)Get (產生空位)Get 會阻塞 (屏障 Count=0)Put 會阻塞 (屏障 Count=Buffer_Size)空 / Count=0部分 / Count=1..Buffer_Size-1滿 / 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_Workerdelay until 等待期間 CPU 時間不會增加,只有在實際計算處理中才會增加。主任務的 delay until Clock + Milliseconds(500) 期間,CPU 時間應該也幾乎為零。不過,CPU 時間的實測並不能保證真正的 WCET。包含快取、管線化、記憶體競爭等因素的 WCET,需要另外用靜態分析或在目標環境上進行驗證。

CPU時間內容: 僅實際計算CPU時間總計: 120ms牆上時間內容: 計算 + 等待 + 阻塞 + 被搶佔經過時間總計: 500ms差值 = 等待・阻塞・被搶佔時間CPU時間可觀察實際計算成本有助於WCET驗證與監控排除等待・阻塞・被搶佔時間注意實測不保證真正的WCET需要靜態分析或目標環境驗證

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。

慢速控制 週期3(release=950ms)慢速控制 週期2(release=550ms)慢速控制 週期1(release=150ms)1000-1010ms快速感測器 #10因P+3而中斷950-1000ms慢速控制 #3 前半1010-1040ms慢速控制 #3 後半600-610ms快速感測器 #6因P+3而中斷550-600ms慢速控制 #2 前半610-640ms慢速控制 #2 後半150-200ms慢速控制 #1 前半100-110ms快速感測器 #1200-210ms快速感測器 #2因P+3而中斷210-240ms慢速控制 #1 後半示範用假設快速感測器: 處理10ms慢速控制: 處理80ms

這種模式,是工業控制系統或機器人控制中常見的「快速感測器收集 + 慢速控制迴路」典型結構。

11. Ada 即時功能大放異彩的場景

Ada 的即時功能特別能在以下領域發揮價值。

Ada Annex D即時功能航太DO-178C鐵路EN 50128系列汽車ISO 26262醫療器材IEC 62304工業控制IEC 61508系列國防・高可靠性系統飛行控制應用實績豐富的領域衛星・太空載具控制號誌系統自動列車控制安全相關ECU的候選方案在以C / MISRA-C為主流的領域中有限且選擇性的應用心律調節器輸液泵機器人控制數控機床任務電腦長期運作系統

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 即時功能的本質在於「不是事後添加的」。防止優先權反轉的鎖定規則、用於週期執行的時刻指定、執行時間監控等,都作為語言規範的一部分提供。當然,能否真正達成期限,仍要靠設計與分析來確認,但語言執行環境已經替你準備好了所需的前提條件,這正是最大的優勢所在。

Ada Annex D即時系統排程FIFO_Within_Priorities任務優先權搶佔優先權反轉防止Ceiling_Locking 協定自動優先權提升天花板優先權規則週期執行delay until防止累積漂移以絕對時刻為基準Ravenscar 設定檔靜態任務集合決定性分析禁止相對delayDO-178C / ISO 26262計時事件不需輪詢的喚醒登錄保護處理器處理器以天花板優先權執行保護物件/佇列入口屏障空/部分/滿的狀態管理屏障式同步執行時間監控各任務CPU時間牆上時間 vs CPU時間有助於WCET驗證・監控整合設計多週期設計以屏障保護語言層級的安全性

接下來,如果想實際嘗試用 Ada 開發即時系統,可以到 Alire 安裝 GNAT 工具鏈,並用 gnatchop + gnatmake 建置本文的範例程式碼。

另外,關於 Ada 並行處理的基礎(任務、會合、保護物件),請參考上一篇文章「Ada 的安全並行處理」。

14. 參考資料

共用相同標籤的最新文章。能以相近的主題延伸理解。

Ada 的安全並行處理 ── 任務與保護物件實戰指南

本文是 Ada 語言內建並行處理機制──任務(task)與保護物件(protected object)的實戰入門文章。內容涵蓋會合(entry/accept)、選擇性接受(select)、生產者消費者模式、保護物件實現的排他控制、帶屏障的保護入口、附逾時的呼叫,以及任務優先...

與本文相近的主題頁面。以本文為起點,可進一步連到相關服務與其他文章。

常見問題

整理諮詢這個主題時常見的問題。

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 軟體開發、技術諮詢與故障調查為中心,在難以重現的故障調查與既有資產仍在運作的專案上具有優勢。

回到部落格一覽