取りこぼさない取り込み処理をAdaのタスクで作る ── 保護オブジェクトのリングバッファと優先度

産業用カメラの画像を受け取る処理では、カメラが一定の間隔でフレームを送り続け、受け取る側の都合では待ってくれません。検査の処理がたまに遅くなると、受け取る側が詰まり、どこかで画像が失われます。失われる場所がアプリの外（ドライバーや SDK の中）だと、アプリからは取りこぼしが見えません。

この記事では取り込みの骨組みを Ada で組み立てます。カメラの到着を模した周期タスク、処理タスク、保護オブジェクトで作る固定長のリングバッファを使い、画像そのものではなくスロットの番号を受け渡します。満杯のときは待たずにそのフレームを捨て、捨てた数を記録します。処理が遅いときに何が起きるかを実際に測り、Linux で優先度が効く条件と、SPARK で確かめられる範囲を書きます。

測ってみると、処理が平均して到着より遅いときは、スロットを増やしても捨てる割合は変わりませんでした。スロットで吸収できるのは一時的な遅れだけです。カメラのタスクを待たせる方式にするとアプリの中で捨てる数は0になりますが、その分はカメラの遅れとして積み上がり、実際のカメラではアプリの外で失われます。Linux では優先度にも優先度上限の検査にも権限が要り、権限がないとエラーを出さずに無視されました。

例はすべて実際にビルドして動かしました。Linux のコンテナ（Ubuntu 24.04、4 vCPU）で、Ubuntu のパッケージの GNAT 13.3.0 を使い、2026年11月10日に測っています。Windows では試していません。SPARK の証明は GitHub Actions の Ubuntu のランナーで、Alire で入れた GNATprove 16.1.0 を使いました。タスクと保護オブジェクトの基本は「Adaにおける安全な並行処理」に、優先度と delay until の一般的な説明は「Adaによるリアルタイムシステムプログラミング」にあります。この記事はカメラの取り込みに絞ります。

カメラのタスク・処理タスク・スロットの分担

フレームを受け取るカメラのタスク、画像を検査する処理タスク、画像を置くスロットの3つで取り込みを組み立てます。スロットは 640×480 の8ビットの画像を入れる領域で、起動時に8個だけ用意します。カメラのタスクは空いたスロットに画像を書き込み、処理タスクはそれを読んで検査します。

2つのタスクの間で受け渡すのは、画像ではなくスロットの番号です。1枚 300KB の画像を毎回写すと、100fps では1秒に 30MB を写すことになります。番号なら1個の整数で済み、メモリも起動時に確保した分から増えません。SDK もふつうは決まった数のバッファを順に使い回します。SDK のバッファの数と受け取り方は「GigE Vision・USB3 Visionカメラの取りこぼし」で扱いました。

番号を入れておく待ち行列は2つあります。空いているスロットの番号を入れる Free と、画像が入ったスロットの番号を入れる Ready です。番号は Free からカメラ、Ready、処理タスクを通って Free へ戻ります。

flowchart TB
        accTitle: スロットの番号が回る経路
        accDescr: 空いているスロットの番号がFreeからカメラのタスクへ渡り、画像を書き込んだあとReadyへ入り、処理タスクが取り出して検査し、Freeへ戻すことを示す。
        F["Free（空いているスロットの番号）"] --> C["カメラのタスク"]
        C -->|"画像を書き込む"| R["Ready（画像が入ったスロットの番号）"]
        R --> W["処理タスク"]
        W -->|"検査が済んだら返す"| F

図1: スロットの番号は Free、カメラのタスク、Ready、処理タスクの順に回り、Free へ戻る。

この構成では、どのスロットの番号もある時点では Free の中、カメラのタスクの手元、Ready の中、処理タスクの手元の4か所のどれか1つにだけあります。番号を持っているタスクだけがそのスロットに触るので、カメラが書き込んでいる最中のスロットを処理タスクが読むことはありません。番号の受け渡しは保護オブジェクトの待ち行列が行いますが、番号を持つタスクだけが触るという約束はコードの書き方で守るもので、画像の領域そのものは排他制御していません。

flowchart TB
        accTitle: 1つのスロットの番号がある4か所
        accDescr: 1つのスロットの番号は、Freeの中、カメラのタスク、Readyの中、処理タスクの4か所のどこか1つにだけあり、番号を持つタスクだけが画像の領域に触ることを示す。
        S["スロット3の番号"] --> A["Free の中"]
        S --> B["カメラのタスクが持つ"]
        S --> D["Ready の中"]
        S --> E["処理タスクが持つ"]
        B -.-> B2["書き込んでよい"]
        E -.-> E2["読んでよい"]

図2: 1つのスロットの番号はある時点では4か所のどれか1つにだけあり、持っているタスクだけが画像に触る。

図の実線は常に成り立つ関係、破線は条件付きの関係です(成立条件は詳細ページの各関係の説明に記載)。関係すべての一覧(全11件、根拠・確度つき)と主要概念の定義は知識マップ詳細ページにまとめています。データ: JSON-LD / Turtle

スロットの番号を入れるリングバッファ

Free と Ready は同じ保護型 Slot_Queue の2つの保護オブジェクトです。容量は8で、スロットの総数と同じにしています。番号は Ring_Index という法8のモジュラー型で数えるので、末尾から先頭へ戻る計算で桁あふれが起きません。

with Capture_Priorities;

package Slot_Queues with SPARK_Mode is

   Capacity : constant := 8;

   type Slot_Index is range 1 .. Capacity;
   type Count_Type is range 0 .. Capacity;
   type Ring_Index is mod Capacity;
   type Slot_Ring is array (Ring_Index) of Slot_Index;

   protected type Slot_Queue (Initially_Full : Boolean)
     with Priority => Capture_Priorities.Camera_Priority
   is
      procedure Put (Slot : Slot_Index; Accepted : out Boolean);
      procedure Try_Take (Slot : out Slot_Index; Found : out Boolean);
      entry Take (Slot : out Slot_Index);
      function Length return Count_Type;
      function Max_Length return Count_Type;
   private
      Items : Slot_Ring := [for I in Ring_Index => Slot_Index (Integer (I) + 1)];
      Head  : Ring_Index := 0;
      Count : Count_Type := (if Initially_Full then Capacity else 0);
      Peak  : Count_Type := (if Initially_Full then Capacity else 0);
   end Slot_Queue;

end Slot_Queues;

判別子 Initially_Full が真なら、最初から1〜8の番号がすべて入った状態で始まります。Free はこの形で作り、Ready は空で作ります。型に付けた Priority は優先度上限で、あとで説明します。

配列 Items のうち、Head から数えて Count 個が入っている番号です。取り出すときは Head の位置から読んで Head を1つ進め、入れるときは Head + Count の位置に書きます。どちらも Ring_Index の計算なので、7の次は0に戻ります。

flowchart TB
        accTitle: リングバッファの位置の計算
        accDescr: 配列ItemsのうちHeadから数えてCount個が入っている番号で、取り出すときはHeadの位置から読んでHeadを進め、入れるときはHead足すCountの位置に書き、7の次は0に戻ることを示す。
        I["Items（添字0〜7）"] --> H["Head から Count 個が入っている番号"]
        H --> T["取り出す：Head の位置を読み、Head を1つ進める"]
        H --> P["入れる：Head + Count の位置に書く"]
        T -.-> M["7の次は0に戻る（法8のモジュラー型）"]
        P -.-> M

図3: 入っている番号は Head から Count 個で、位置の計算は法8で折り返す。

package body Slot_Queues with SPARK_Mode is

   protected body Slot_Queue is

      procedure Put (Slot : Slot_Index; Accepted : out Boolean) is
      begin
         if Count = Capacity then
            Accepted := False;
         else
            Items (Head + Ring_Index'Mod (Count)) := Slot;
            Count := Count + 1;
            if Count > Peak then
               Peak := Count;
            end if;
            Accepted := True;
         end if;
      end Put;

      procedure Try_Take (Slot : out Slot_Index; Found : out Boolean) is
      begin
         if Count = 0 then
            Slot := Slot_Index'First;
            Found := False;
         else
            Slot := Items (Head);
            Head := Head + 1;
            Count := Count - 1;
            Found := True;
         end if;
      end Try_Take;

      entry Take (Slot : out Slot_Index) when Count > 0 is
      begin
         Slot := Items (Head);
         Head := Head + 1;
         Count := Count - 1;
      end Take;

      function Length return Count_Type is (Count);

      function Max_Length return Count_Type is (Peak);

   end Slot_Queue;

end Slot_Queues;

操作は3種類あり、待つかどうかが違います。Put と Try_Take は手続きなので、バリアで呼んだタスクを待たせることはありません（保護オブジェクトのロックを取る短い間は待つことがあります）。満杯や空のときは、結果を Accepted や Found で返します。Take はエントリで、バリア when Count > 0 が真になるまで呼んだタスクを待たせます。カメラのタスクは待ってはいけないので手続きだけを使い、処理タスクは Ready.Take で次の画像を待ちます。

flowchart TB
        accTitle: Slot_Queueの3つの操作
        accDescr: Putは満杯ならAcceptedを偽にして返し、Try_Takeは空ならFoundを偽にして返し、Takeはバリアが真になるまで呼んだタスクを待たせることを示す。
        Q["Slot_Queue"] --> P["Put（手続き）"]
        Q --> T["Try_Take（手続き）"]
        Q --> K["Take（エントリ）"]
        P -.-> P2["満杯なら Accepted を偽にして返す"]
        T -.-> T2["空なら Found を偽にして返す"]
        K -.-> K2["Count が0の間は待たせる"]

図4: Put と Try_Take は待たせずに結果を返し、Take だけがバリアで呼んだタスクを待たせる。

バリアの Count > 0 はあとで使う Jorvik プロファイルでは書けますが、Ravenscar プロファイルでは書けません。Ravenscar のバリアは静的な式か、保護オブジェクトの成分（component）1つを指す名前に限られます（Simple_Barriers）¹。SPARK のユーザーガイドもそのために真偽値の成分を足す例を示しています²。Jorvik はこれを、保護オブジェクトの成分と静的な式を比較演算や論理演算で組み合わせた式（Pure_Barriers）まで緩めています³。

Peak にはそれまでに入っていた番号の数の最大を残します。Ready の Peak を見れば、処理が遅れた間にいくつ溜まったかが分かります。

カメラのタスクを待たせない

空いたスロットがないときの方針は2つ考えられます。カメラのタスクを空くまで待たせるか、そのフレームを捨てるかです。この記事の構成では捨てて、捨てた数を記録します。

待たせる方式なら取りこぼしがないように見えます。しかし実際のカメラは、アプリが受け取らなくても次のフレームを送ってきます。アプリが待っている間に届いたフレームは、ドライバーや SDK のバッファがあふれた時点で、アプリの知らないところで捨てられます。アプリの中で捨てて数えれば、処理が追いつかずに失ったフレームを、ネットワークや USB での取りこぼしと分けて記録できます。

flowchart TB
        accTitle: 空いたスロットがないときの2つの方針
        accDescr: 空いたスロットがないとき、捨てる方式はアプリの中で捨てた数を記録でき、待たせる方式はアプリの外のドライバーやSDKで失われてアプリから見えないことを示す。
        N{"空いたスロットがあるか"} -->|"ある"| OK["画像を書き込んで Ready へ"]
        N -->|"ない"| P{"方針"}
        P -->|"捨てる"| D["アプリの中で捨てた数を数える"]
        P -->|"待たせる"| W["カメラのタスクが止まる"]
        W -.-> L["その間のフレームはドライバーや SDK で失われる"]

図5: 待たせる方式では、失われたフレームがアプリから見えなくなる。

カメラのタスクは次のとおりです。Wait_For_Slot は比較のために待たせる方式へ切り替える設定で、既定は偽（捨てる方式）です。

   task body Camera is
      Next     : Time := Clock;
      Now      : Time;
      Slot     : Slot_Index;
      Found    : Boolean;
      Accepted : Boolean;
      Returned : Boolean;
      Left     : Count_Type;
   begin
      for Frame in 1 .. Frames loop
         Next := Ticks.Add (Next, Period);
         delay until Next;
         if Wait_For_Slot then
            Free.Take (Slot);   --  空きが出るまで待つ方式（比較用）
            Found := True;
         else
            Free.Try_Take (Slot, Found);
         end if;
         if Found then
            Frame_Pool.Fill (Slot, Frame, Next);
            Ready.Put (Slot, Accepted);
            if not Accepted then
               Free.Put (Slot, Returned);
            end if;
         else
            Accepted := False;
         end if;
         Now := Clock;
         Stats.Arrived (Frame, Ticks.Since (Now, Next), Dropped => not Accepted);
      end loop;

      --  すべてのスロットが空きに戻るまで待ってから終わりを知らせる
      loop
         Left := Free.Length;
         exit when Left = Capacity;
         Next := Ticks.Add (Next, Period);
         delay until Next;
      end loop;
      Stats.Finish;

      loop
         delay until Time_Last;
      end loop;
   end Camera;

delay until Next は前回の予定時刻に周期を足した絶対時刻まで待ちます。処理が少し遅れても次の予定時刻はずれないので、10ミリ秒ごとの到着を模せます。時刻の足し算と引き算は Ticks.Add と Ticks.Since で行います（理由は SPARK の節で説明します）。Frame_Pool.Fill は受信した画像を書き込む代わりに、フレーム番号で全画素を埋めます。Ready の容量はスロットの総数と同じなので Put が断ることはありませんが、断られた場合はスロットを Free へ戻します。Stats.Arrived には予定時刻からどれだけ遅れて処理を終えたかと、捨てたかどうかを記録します。

最後の loop は測定が終わったあとタスクを止めずに眠らせておくためのものです。理由は Jorvik プロファイルの制限にあり、次の節で説明します。

優先度と優先度上限の決め方

処理タスクは Ready から番号を取り出し、画像を読んで検査し、番号を Free へ返します。検査の代わりに、決めた時間だけ CPU を回し続けます（Simulate_Work）。Frame_Pool.Inspect は全画素の合計を計算するので、別のタスクが書き換えている最中の画像を読めば合計が合わなくなります。

   task body Worker is
      Slot     : Slot_Index;
      Frame    : Frame_Number;
      Arrival  : Time;
      Started  : Time;
      Sum      : Unsigned_32;
      Accepted : Boolean;
   begin
      loop
         Ready.Take (Slot);
         Started := Clock;
         Frame_Pool.Inspect (Slot, Frame, Arrival, Sum);
         Simulate_Work (Frame, Started);
         --  記録してからスロットを返す。先に返すと、カメラのタスクがすべての
         --  スロットが空いたのを見て、この記録より前に終わりを知らせうる。
         Stats.Processed
           (Frame, Ticks.Since (Started, Arrival),
            Intact => Sum = Frame_Pool.Expected_Checksum (Frame));
         Free.Put (Slot, Accepted);
      end loop;
   end Worker;

処理タスクは Stats.Processed で記録してから番号を Free へ返します。カメラのタスクは、すべてのスロットが Free に戻ったのを見て測定の終わりを知らせます。返すのを先にすると、最後のフレームの記録より前に終わりが知らされ、そのフレームが処理した数にも捨てた数にも入らないことがあります。

flowchart TB
        accTitle: 処理タスクが記録してから番号を返す順序
        accDescr: 処理タスクが検査を終えたらまずStatsに記録し、そのあとスロットの番号をFreeへ返し、カメラのタスクはすべての番号がFreeに戻ったのを見て終わりを知らせるので、最後のフレームも記録されてから終わりが知らされることを示す。
        W1["処理タスクが検査を終える"] --> W2["Stats に記録する"]
        W2 --> W3["番号を Free へ返す"]
        W3 --> C["カメラのタスクが全部戻ったのを見る"]
        C --> F["Stats.Finish で終わりを知らせる"]
        W3 -.-> X["逆の順だと最後のフレームの記録が漏れうる"]

図6: 処理タスクは記録してから番号を返すので、終わりが知らされる前に最後のフレームも記録される。

2つのタスクと2つの待ち行列はライブラリーレベルのパッケージに置きます。カメラのタスクの優先度は20、処理タスクは10です。処理が CPU を使い続けても、カメラのタスクが予定時刻に動けるようにするためです。

with Slot_Queues;    use Slot_Queues;
with Capture_Priorities; use Capture_Priorities;

package Capture_Tasks with SPARK_Mode is

   --  空いているスロットの番号と、画像が入ったスロットの番号
   Free  : Slot_Queue (Initially_Full => True);
   Ready : Slot_Queue (Initially_Full => False);

   task Camera with Priority => Camera_Priority;
   task Worker with Priority => Worker_Priority;

end Capture_Tasks;

保護オブジェクトの優先度は優先度上限（ceiling）として扱われます。Ceiling_Locking の方針では、タスクが保護オブジェクトの操作を呼ぶとき、そのタスクの優先度が上限より高くないかが検査され、高ければ Program_Error になります⁴。Free と Ready は優先度20のカメラのタスクから呼ばれるので、Slot_Queue の型の上限を20（Camera_Priority）にしています。最初は上限を判別子で渡していましたが、GNATprove 16.1.0 は呼び出すタスクの優先度が上限を超えないことを証明しきれなかったので、型に定数を書く形にしました。記録用の保護オブジェクト Stats も同じ20です。結果を表示するメインのプログラムは優先度1にして、どの保護オブジェクトの上限も超えないようにしています。メインのプログラムも Stats と Ready の関数を呼びますが、メインは SPARK の外なので、これらの呼び出しの上限の検査は後で書く GNATprove の件数に入っていません。

flowchart TB
        accTitle: タスクの優先度と保護オブジェクトの優先度上限
        accDescr: 優先度20のカメラのタスクと優先度10の処理タスクが、優先度上限20のFreeとReadyとStatsを呼び、優先度1のメインのプログラムもStatsを呼ぶことを示す。
        C["カメラのタスク（優先度20）"] --> Q["Free と Ready（上限20）"]
        W["処理タスク（優先度10）"] --> Q
        C --> S["Stats（上限20）"]
        W --> S
        M["メインのプログラム（優先度1）"] --> S
        Q -.-> R["上限は呼ぶタスクの優先度以上にする"]

図7: 保護オブジェクトの優先度上限はそれを呼ぶどのタスクの優先度よりも低くしない。

Jorvik プロファイルで決まること

プログラム全体には構成プラグマのファイル gnat.adc で Jorvik プロファイルを指定しています。SPARK でタスクを扱うには Ravenscar か Jorvik のプロファイルと、エラボレーションが終わるまでタスクを動かさない Partition_Elaboration_Policy (Sequential) が必要です²。

pragma Profile (Jorvik);
pragma Partition_Elaboration_Policy (Sequential);

Jorvik プロファイルはタスクのディスパッチ方針（Task_Dispatching_Policy）の FIFO_Within_Priorities、ロックの方針 Ceiling_Locking、Detect_Blocking と、いくつかの制限をまとめて指定したものと同じです³。制限にはタスクをライブラリーレベルにしか置けない（No_Task_Hierarchy）、タスクが終わらない（No_Task_Termination）、select 文を使わない（No_Select_Statements）などがあります。カメラのタスクが測定のあと眠り続けるのは、タスクが終わらない前提だからです。メインのプログラムは Stats.Wait_Done で測定の終わりを待ち、結果を表示したあと GNAT.OS_Lib.OS_Exit でプロセスを終えます。

flowchart TB
        accTitle: Jorvikプロファイルが指定するもの
        accDescr: Jorvikプロファイルが、タスクのディスパッチ方針FIFO_Within_Priorities、ロックの方針Ceiling_Locking、ライブラリーレベルのタスクや終わらないタスクなどの制限をまとめて指定することを示す。
        J["pragma Profile（Jorvik）"] --> D["FIFO_Within_Priorities"]
        J --> L["Ceiling_Locking"]
        J --> R["制限"]
        R --> R1["タスクはライブラリーレベルだけ"]
        R --> R2["タスクは終わらない"]
        R --> R3["select 文を使わない"]

図8: Jorvik プロファイルは優先度の扱いと並行処理の制限をまとめて指定する。

処理が遅いときに何が起きたか

10ミリ秒ごとにフレームが届く設定（100fps）で、1000フレームずつ流して測りました。処理タスクには CPU を回し続ける形で処理時間を与えています。root で動かしたので、優先度は効いています（次の節で確かめます）。「処理開始までの待ち」は到着の予定時刻から処理タスクが取り出すまでの時間、「カメラの遅れ」は予定時刻から、カメラのタスクがそのフレームの番号を Ready に入れ終える（捨てる場合は捨て終える）までの時間です。

どの測定でも、処理タスクが読んだ画像の合計は書き込んだ値と一致しました。カメラが書き込んでいる最中のスロットを読んだことを示す食い違いは、一度も出ていません。

処理が毎回12ミリ秒かかると、平均して到着より遅いので、スロットをいくつ用意しても足りません。捨てた数は1000フレーム中172で、到着の10ミリ秒と処理の12ミリ秒の比から見込まれる約17%（約167フレーム。最初に空いている8スロットの分を除くと約160フレーム）よりやや多くなりました（同じ条件のほかの3回は173〜177）。Ready には最大7個（残り1個は処理タスクが持つ）が溜まり、処理を始めるまでの待ちは約80ミリ秒に延びました。この状態でスロットを増やしても、捨てる割合は変わらず、待ちが長くなるだけです。Basler の資料も、バッファを増やすと一時的な遅れには耐えられる一方で待ち時間が増えることと、処理が続けて遅いとバッファがあふれるおそれがあることを挙げています⁵。平均して遅いときは、処理を速くするか、フレームを間引くかを決める必要があります。

flowchart TB
        accTitle: 処理の遅れの2つの型
        accDescr: 処理が平均して到着より遅い場合はスロットを増やしても捨てる割合は変わらず、一時的な遅れの場合は遅れの間に届くフレーム数よりスロットが多ければ吸収できることを示す。
        A{"処理の遅れの型"} -->|"平均して到着より遅い"| B["スロットを増やしても捨てる割合は同じ"]
        A -->|"一時的に遅い"| C{"遅れの間に届く数より空きが多いか"}
        C -->|"多い"| D["捨てずに吸収できる"]
        C -->|"少ない"| E["足りない分を捨てる"]
        B -.-> F["処理を速くするか間引くかを決める"]

図9: 平均して遅いならスロットでは解決せず、一時的な遅れならスロットの数で吸収できるかが決まる。

一時的な遅れを吸収できるかどうかは、スロットの数で決まります。100フレームごとに60ミリ秒かかる場合、その間に届く6フレームは Ready に溜まり（最大6）、捨てずに済みました。120ミリ秒の場合は、その間に12フレームが届きますが、溜められるのは7個までです。1回の遅れにつき5フレームを捨て、遅れは9回あったので計45フレームでした（1000番目のフレームでの遅れは、すべてのフレームが届いたあとに起きたので数に入りません）。必要なスロットの数は許したい遅れの長さを到着の間隔で割った数に、処理中の1個と余裕を足して見積もれます。

flowchart TB
        accTitle: 120ミリ秒の遅れで5フレームを捨てた内訳
        accDescr: 処理が120ミリ秒かかる間に10ミリ秒ごとに12フレームが届き、スロット8個のうち1個は処理中なのでReadyに溜められるのは7個までで、残りの5フレームを捨てたことを示す。
        A["処理に120ms かかる"] --> B["その間に12フレームが届く"]
        B --> C["Ready に溜められるのは7個"]
        C --> D["残りの5フレームを捨てる"]
        C -.-> E["スロット8個のうち1個は処理中"]

図10: 120ミリ秒の遅れの間に届く12フレームのうち、溜められない5フレームを捨てた。

待たせる方式と比べる

同じ条件で、カメラのタスクを空きが出るまで待たせる方式（カメラのタスクの Wait_For_Slot を真にしたもの）も試しました。

待たせる方式では捨てた数が0になりますが、処理が平均して遅い場合はカメラの遅れが積み上がり、最後には予定時刻から約2.2秒遅れていました。実際のカメラは10秒で1000フレームを送り終えるので、遅れた分のフレームはアプリに届く前に失われます。一時的な遅れの場合も、カメラのタスクは最大41ミリ秒止まりました。その間に届いたフレームを SDK のバッファが受け止めきれなければ、やはりアプリの外で失われます。

flowchart TB
        accTitle: 捨てる方式と待たせる方式の結果の違い
        accDescr: 処理が平均して遅いとき、捨てる方式は捨てた数を記録してカメラの遅れが小さいままで、待たせる方式は捨てた数が0でもカメラの遅れが約2.2秒まで積み上がることを示す。
        S["処理が平均して遅い"] --> D["捨てる方式"]
        S --> W["待たせる方式"]
        D --> D1["捨てた数172を記録"]
        D --> D2["カメラの遅れは最大6.0ms"]
        W --> W1["捨てた数は0"]
        W --> W2["カメラの遅れは最大約2.2秒"]
        W2 -.-> W3["実際にはアプリの外で失われる"]

図11: 待たせる方式で捨てた数が0になっても、その分はカメラの遅れになり、実際のカメラではアプリの外で失われる。

Linux で優先度が効く条件

ここまでの測定は root で動かしました。実行中に各スレッドのスケジューリングの方針を chrt -p で見ると、メインのプログラム、処理タスク、カメラのタスクのスレッドが SCHED_FIFO になり、優先度はそれぞれ2、11、21でした。Ada の優先度に1を足した値です。GNAT の Linux 版は FIFO_Within_Priorities のタスクのスレッドに SCHED_FIFO を設定します⁶。

同じプログラムを一般のユーザー（nobody）で動かすと、エラーは何も出ず、すべてのスレッドが SCHED_OTHER の優先度0になりました。GNAT の Linux 版はスレッドの優先度を pthread_setschedparam で設定し、権限がなくて失敗しても（EPERM）そのまま続けます⁶。一般のユーザーのスレッドが SCHED_FIFO を使えるかは CAP_SYS_NICE と RLIMIT_RTPRIO で決まります。CAP_SYS_NICE がなければ、RLIMIT_RTPRIO の値より高い優先度は設定できません⁷。このプログラムは Linux の優先度で2、11、21を求めるので、RLIMIT_RTPRIO は21以上が要ります。それより小さいと、値を超える優先度のスレッドは SCHED_OTHER のまま残るはずです。たとえば5なら SCHED_FIFO になるのはメインのプログラムだけで、カメラと処理のタスクは SCHED_OTHER のまま残り、優先度の順序が逆になります。この環境の RLIMIT_RTPRIO は0で、値を変えることも権限がなくできなかったので、RLIMIT_RTPRIO を上げた場合は試していません。

flowchart TB
        accTitle: タスクの優先度が効く条件
        accDescr: タスクの優先度は、CAP_SYS_NICEか十分なRLIMIT_RTPRIOがあればSCHED_FIFOとして効き、なければエラーを出さずにSCHED_OTHERになり、RLIMIT_RTPRIOは最高の優先度に1を足した値以上が要ることを示す。
        P["タスクの優先度"] --> P1{"CAP_SYS_NICE か十分な RLIMIT_RTPRIO"}
        P1 -->|"ある"| P2["SCHED_FIFO で効く"]
        P1 -->|"ない"| P3["SCHED_OTHER（エラーなし）"]
        P1 -.-> N["RLIMIT_RTPRIO は最高の優先度＋1 以上"]

図12: スレッドの優先度は CAP_SYS_NICE か十分な RLIMIT_RTPRIO があるときだけ効き、なければエラーを出さずに無視される。

優先度上限の検査はスレッドの優先度とは別の条件で決まります。GNAT の Linux 版のランタイムは、プロセスが root で動いているか、CAP_SYS_NICE を持つと判定できたときに、保護オブジェクトのロックに優先度上限つきのミューテックス（PTHREAD_PRIO_PROTECT）を使います。そうでないときは Ceiling_Locking の指定を無視します⁶。root については root かどうかしか見ないので、root でも CAP_SYS_NICE を持たないと、ロックは優先度上限つきになるのに、スレッドは SCHED_FIFO になりません。

CAP_SYS_NICE の判定は libcap の関数を弱い参照で呼ぶ作りです。Ubuntu の GNAT 13 の libgnat で、cap_get_proc が弱いシンボルであることを nm で確かめました。libcap がリンクされていないと、setcap cap_sys_nice=ep で権限を付けても判定は偽になります。Ubuntu の gcc は既定で --as-needed を使うので、リンクの指定に libcap を足すだけでは、使われていないライブラリとして外されます。

この環境で、上限を下げて違反が起きるようにしたプログラムを3通りにリンクし、setcap cap_sys_nice=ep を付けて nobody で動かしました。libcap をリンクしないものと、リンクの指定に libcap を足しただけのもの（ldd に libcap が出ない）は、違反を検出せずに最後まで動きました。-Wl,--no-as-needed を付けて libcap をリンクしたもの（ldd に libcap が出る）だけが、root のときと同じく違反で止まりました。上限を下げていない正しいプログラムを同じリンクで動かすと、nobody でもスレッドが SCHED_FIFO になり、最後まで動きました。この環境には libcap の開発用パッケージがないので、-l:libcap.so.2 で共有ライブラリを直接指定しています。ふつうは libcap-dev を入れて -lcap と書き、ldd で libcap がリンクされたことを確かめます。

flowchart TB
        accTitle: libcap のリンクの仕方と違反の検出
        accDescr: setcapでCAP_SYS_NICEを付けてnobodyで動かすと、libcapをリンクしない版とリンクの指定に足しただけの版は違反を検出せず、-Wl,--no-as-neededでリンクした版だけが違反で止まったことを示す。
        L["上限を下げたものを nobody で実行"] --> A["libcap をリンクしない"]
        L --> B["libcap を足しただけ"]
        L --> C["--no-as-needed でリンク"]
        A --> N["違反を検出しない"]
        B --> N
        C --> D["違反で止まる"]

図13: CAP_SYS_NICE を付けても、libcap が実際にリンクされたものだけが違反を検出した。

flowchart TB
        accTitle: 優先度上限の検査が効く条件
        accDescr: rootはCAP_SYS_NICEを持てば違反がProgram_Errorになり、持たなければ正しいコードでもProgram_Errorになること、一般のユーザーはCAP_SYS_NICEとlibcapのリンクが両方あれば違反を検出し、どちらかがなければ検査されないことを示す。
        C["優先度上限の検査"] --> C1{"root か"}
        C1 -->|"root"| C2{"CAP_SYS_NICE を持つか"}
        C2 -->|"持つ"| OK["違反は Program_Error"]
        C2 -->|"持たない"| BAD["正しいコードも Program_Error"]
        C1 -->|"一般のユーザー"| C3{"CAP_SYS_NICE と libcap のリンク"}
        C3 -->|"両方ある"| OK
        C3 -->|"どちらかない"| NO["検査されない（エラーなし）"]

図14: 優先度上限の検査は CAP_SYS_NICE を持つ root か、libcap をリンクして CAP_SYS_NICE を持つ一般のユーザーで効き、CAP_SYS_NICE のない root では正しいコードも止まる。

優先度が効かないと何が変わるかを見るため、プロセスを CPU 0 だけに固定し、同じ CPU で空回りする別のプロセスを動かしました。処理は毎回5ミリ秒です。

flowchart TB
        accTitle: CPU 0を分け合う測定の構成
        accDescr: カメラのタスク、処理タスク、空回りする別のプロセスがCPU 0を分け合い、SCHED_FIFOならカメラのタスクが先に動き、SCHED_OTHERなら3つで時間を分け合うことを示す。
        C["カメラのタスク"] --> CPU["CPU 0"]
        W["処理タスク（5ms 回り続ける）"] --> CPU
        H["空回りする別のプロセス"] --> CPU
        CPU -.-> F["SCHED_FIFO：カメラのタスクが先に動く"]
        CPU -.-> O["SCHED_OTHER：3つで時間を分け合う"]

図15: 3つが CPU 0 を分け合うとき、優先度が効くかどうかでカメラのタスクの動き方が変わる。

一般のユーザーでは、カメラのタスクが空回りするプロセスや処理タスクと CPU を分け合うので、予定時刻から平均3.2ミリ秒遅れました。同じ条件で別に3回測ったときは、平均が3.1〜3.6ミリ秒、最大が14.8〜17.6ミリ秒でした。root では4回とも平均0.1〜0.2ミリ秒で、最大は2.6〜6.2ミリ秒でした。この環境は仮想マシンの上のコンテナなので、root でも数ミリ秒の遅れは残ります。

優先度上限の違反の検出と権限

優先度上限を下げると何が起きるかも試しました。Slot_Queue の上限を処理タスクと同じ10にすると、優先度20のカメラのタスクが最初に Free.Try_Take を呼んだ時点で上限の違反になります。

root で動かすと、カメラのタスクで PROGRAM_ERROR : ... ceiling violation が発生しました。例外を処理していないので、カメラのタスクは何も表示せずに終わり、メインのプログラムは測定の終わりを待ち続けて止まりました（例外の内容は別に例外ハンドラーを足して確かめました）。一般のユーザーでは上で書いたとおり優先度上限つきのミューテックスを使わないので、違反は検出されず、プログラムは最後まで動いて結果を表示しました。GNAT の Linux 版は、このミューテックスのロックが EINVAL を返したときに上限の違反としています⁶。

CAP_SYS_NICE を外した root（setpriv --bounding-set -sys_nice で実行）では、上限を下げていない正しいプログラムが raised PROGRAM_ERROR : s-tarest.adb:664 explicit raise で止まり、終了コードは1でした。ロックは優先度上限つきなのにスレッドは SCHED_FIFO になれない組み合わせで、ロックが失敗したものと考えられますが、ロックが返した値までは確かめていません。Docker の既定の権限には CAP_SYS_NICE が入っていないので⁸、コンテナで動かすときは root であることだけでなく、/proc/self/status の CapEff で CAP_SYS_NICE を持つかを確かめます。

権限の違う環境で試すと、優先度もその検査も違う結果になります。装置に載せる前のテストは、本番と同じユーザーと権限で行う必要があります。

CPU の割り当て

Ada の参照マニュアルは、Ravenscar や Jorvik のもとでは、CPU を指定しないタスクはすべて1つの CPU で動くと注記しています³。この環境の GNAT 13.3.0 では、どのスレッドも4つの CPU すべてで動ける状態（Cpus_allowed_list が 0-3）でした。処理タスクとカメラのタスクが別の CPU で同時に動くこともあるので、1つの CPU を前提にした優先度の議論はそのまま当てはまりません。上の比較で CPU 0 に固定したのは、この差をなくして優先度の効果だけを見るためです。

Windows では試していません。Windows でのスレッドの優先度と CPU の割り当ては「Windowsアプリ開発者のためのCPU設定入門」と「普通のWindowsでソフトリアルタイムをできるだけ実現するための実践ガイド」で扱っています。

SPARK で確かめられることと確かめられないこと

待ち行列、2つのタスク、記録用の保護オブジェクトは SPARK_Mode の中に置き、GNATprove で解析しました。SPARK では、タスクが共有してよいのは保護オブジェクトやアトミックな変数のように同時アクセスから守られたものと、エラボレーションのあと変わらない変数だけです²。2つのタスクが守られていない変数を共有すると、GNATprove はデータ競合の可能性として報告します。

時刻の足し算と引き算（Next + Period や Now - Next）は、時刻が型の上限に近づいたときにあふれないことを GNATprove が示せませんでした。そこで、この計算だけを本体が SPARK の外にある小さなパッケージ Ticks にまとめています。測定の記録を集計して表示するメインのプログラムと、引数を読む本体も SPARK の外です。

flowchart TB
        accTitle: SPARK の中と外に置いたもの
        accDescr: 待ち行列、タスク、記録用のStats、TicksとFrame_Poolの仕様をSPARKの中に置いてGNATproveで解析し、TicksとFrame_Poolの本体、引数を読む本体、メインのプログラムを外に置いたことを示す。
        P["取り込みのプログラム"] --> S["SPARK の中（GNATprove）"]
        P --> O["SPARK の外"]
        S --> A["待ち行列・タスク・記録"]
        S --> D["Ticks と Frame_Pool の仕様"]
        O --> E["Ticks と Frame_Pool の本体"]
        O --> F["引数を読む本体とメイン"]

図16: 待ち行列、タスク、記録は SPARK の中に置き、時刻の計算と画像の領域の本体、メインは外に置いた。

with Ada.Real_Time; use Ada.Real_Time;

--  時刻の足し算と引き算。時刻が型の上限に近づいたときにあふれないことを
--  GNATprove は示せないので、本体を SPARK の検査の外に置く。
package Ticks with SPARK_Mode is

   function Add (T : Time; D : Time_Span) return Time
   with Global => null;

   function Since (Later, Earlier : Time) return Time_Span
   with Global => null;

end Ticks;

GitHub Actions の Ubuntu のランナーで GNATprove 16.1.0 を --level=2 で動かすと、「Success: all checks proved (43 checks).」と出て、43件の検査がすべて証明されました（フロー解析で25件、証明器で18件）。データ競合の報告はなく、証明された検査には範囲の検査、6つの呼び出しの組についての優先度上限の検査、2つのタスクが終わらないことが含まれます。

2つのタスクが同じスロットの画像に同時に触らないことは、番号を受け渡すという約束で守っているだけで、画像の領域そのものは守られていません。この約束は SPARK の型や保護オブジェクトでは表せないので、画像の領域を持つパッケージは、仕様だけを SPARK にして本体を解析の外に置きました。

with Interfaces;
with Ada.Real_Time;
with Slot_Queues;    use Slot_Queues;
with Capture_Config; use Capture_Config;

--  画像の領域。スロットの番号を持っているタスクだけが、そのスロットに触る。
--  この約束は SPARK では証明できないので、Synchronous と宣言して本体を検査の外に置く。
package Frame_Pool with
  SPARK_Mode,
  Abstract_State => (Pixels with Synchronous, External),
  Initializes    => Pixels
is

   procedure Fill
     (Slot    : Slot_Index;
      Frame   : Frame_Number;
      Arrival : Ada.Real_Time.Time)
   with Global => (In_Out => Pixels);

   procedure Inspect
     (Slot     : Slot_Index;
      Frame    : out Frame_Number;
      Arrival  : out Ada.Real_Time.Time;
      Checksum : out Interfaces.Unsigned_32)
   with Global => (In_Out => Pixels);

   function Expected_Checksum (Frame : Frame_Number) return Interfaces.Unsigned_32
   with Global => null;

end Frame_Pool;

抽象状態 Pixels に付けた Synchronous は、この状態が同時アクセスから守られたものだけでできていると宣言するものです²。Initializes => Pixels は、パッケージの初期化でこの領域ができることを宣言します。本体が SPARK の中にあれば GNATprove はどちらも確かめますが、ここでは本体を外に置いたので確かめられません。GNATprove は Pixels の初期化を proved と表示しましたが、これはこの宣言を前提にした結果で、本体の中身は解析していません。Synchronous は証明された事実ではなくスロットの番号の約束にもとづく前提です。

試しに Synchronous を外すと、GNATprove は possible data race when accessing variable "frame_pool.pixels" を2件報告しました。Slot_Queue の優先度上限を10に下げたコードでは、ceiling priority protocol might not be respected を2件報告しました。どちらも優先度20のカメラのタスクの呼び出し（Free.Take と Ready.Put）で、優先度10の処理タスクからの呼び出しは報告されませんでした。

flowchart TB
        accTitle: 証明したことと前提にしたこと
        accDescr: GNATproveが待ち行列とタスクのデータ競合と実行時エラーがないことを証明し、画像の領域が同時に触られないことと時刻の計算があふれないことは前提で、画像の領域は実行時の合計の照合で確かめたことを示す。
        G["GNATprove の証明"] --> A["データ競合がない"]
        G --> B["実行時エラーがない"]
        P["前提"] --> C["画像を同時に触らない"]
        P --> T["時刻の計算はあふれない"]
        C -.-> D["実行時に合計を照合した"]

図17: 待ち行列とタスクは証明でき、画像の領域を同時に触らないことは前提と実行時の照合で確かめた。

前提を補うため、処理タスクは読んだ画像の合計を、書き込んだ値から計算した合計と照合しています。上の測定ではどれも一致しました。ただし、これは試した実行で食い違いが起きなかったことを示すだけです。例えば同じ番号を Free へ2回返すような誤りを入れると、2つのタスクが同じスロットを持つことになりますが、SPARK の解析はそれを検出しません。番号の受け渡しを変えるときは、この約束が崩れていないかを、人がコードを読んで確かめる必要があります。

SPARK の証明の進め方は「SPARKによる形式検証入門」と「AIが書いたコードをSPARKで証明する」にあります。

試した環境と確かめていないこと

• ビルドと測定: Linux のコンテナ（Ubuntu 24.04、4 vCPU の仮想マシン、カーネル 6.18）、Ubuntu のパッケージの GNAT 13.3.0 と gprbuild。2026年11月10日。コンパイラーのスイッチは -gnat2022 -O2 -gnata で、構成プラグマのファイル gnat.adc で Jorvik プロファイルを指定しました（Slot_Queues の角括弧の集成体は Ada 2022 の書き方なので、-gnat2022 がないとコンパイルできません）。測定は root と nobody の両方で行い、権限の比較では setpriv と setcap で CAP_SYS_NICE を付け外ししました。
• 検証に使ったプロジェクト一式は公開しておらず、記事のコードはその抜粋です。
• 証明: GitHub Actions の Ubuntu 24.04 のランナーで、Alire で入れた GNAT 16.1.0 と GNATprove 16.1.0（FSF 版）を --level=2 で動かしました。記事の表の値はすべて上のコンテナの GNAT 13.3.0 で測ったものです。
• カメラはつないでいません。フレームの到着は delay until の周期で模し、受信はフレーム番号で画素を埋める処理で模しました。実際の SDK のコールバックや DMA の負荷は含みません。
• 検査の処理は CPU を回し続けることで模しました。実際の検査のようにメモリを大量に読む処理では、キャッシュやメモリの帯域の影響が加わります。
• Windows と、リアルタイム用のカーネル（PREEMPT_RT など）では試していません。RLIMIT_RTPRIO を0より大きくした場合も、値を変える権限がなく試していません。
• 測定の値はこの仮想マシンでの結果で、実行ごとにばらつきます。特に最大値は同じ条件でも数倍違うことがありました。

関連記事

この記事はシリーズ「Ada とマシンビジョン」の1本です。シリーズのほかの記事は次のとおりです。

• 画像処理の境界外アクセスをSPARKで証明する ── 2値化・平滑化・ラベリングとCとの速度比較
• GigE VisionのパケットをAdaの型で受ける ── 表現句の落とし穴とSPARKで証明するパーサー
• Adaの検査ロジックをC#の検査アプリから呼ぶ ── 画像バッファをコピーせずに渡し、境界で守る
• ピクセルとミリメートルを型で分ける ── Adaで寸法検査の単位の取り違えを防ぐ

関係する既存の記事は次のとおりです。

• GigE Vision・USB3 Visionカメラの取りこぼし ── 原因の切り分けとWindows側の設定
• Adaにおける安全な並行処理 ── タスクと保護オブジェクトの実践ガイド
• Adaによるリアルタイムシステムプログラミング ── 優先度・周期・実行時間制御の実践
• Ada 2022 の新しい機能 ── GNAT で動くものと、まだ書けない並列構文
• Alire で Ada のプロジェクトを始める ── プロジェクト作成、依存の追加、ビルドプロファイル、GNATproveまで

参考リンク

関連する記事

同じタグを共有する最新の記事です。さらに近い話題で知識を深められます。

関連トピック

このテーマと近いトピックページです。記事を起点に、関連するサービスや他の記事へ進めます。

このテーマがつながるサービス

この記事は次のサービスページにつながります。近い入口からご覧ください。

よくある質問

この記事のテーマについて、相談時によくある質問をまとめています。

カメラの画像を受け取るタスクは、バッファが満杯のとき待たせるべきですか？

この記事の構成では待たせず、そのフレームを捨てて数を記録しました。待たせる方式で試すと、捨てた数は0のままでしたが、到着の予定時刻からの遅れが最大で約2.2秒まで積み上がりました。実際のカメラは待ってくれないので、その間のフレームはドライバーやSDKの中で失われ、アプリからは見えなくなります。

リングバッファのスロットを増やせば、処理が遅くても取りこぼしは防げますか？

一時的な遅れは吸収できますが、処理が平均して到着より遅い状態は防げません。10ミリ秒ごとに届くフレームを12ミリ秒かけて処理した測定では、8スロットでも1000フレーム中172フレームを捨てました。100フレームごとに60ミリ秒かかる遅れは捨てずに吸収でき、120ミリ秒かかる遅れでは1回あたり5フレームを捨てました。

LinuxでAdaのタスクに優先度を付ければ、それだけで効きますか？

権限によります。スレッドの優先度が効くにはCAP_SYS_NICEか、Adaの最高の優先度に1を足した値以上のRLIMIT_RTPRIOが要ります。優先度上限の検査が効くのは、CAP_SYS_NICEを持つrootか、libcapを-Wl,--no-as-neededでリンクしたうえでCAP_SYS_NICEを持つ一般のユーザーです。この記事の環境では、一般のユーザーではすべてのスレッドがSCHED_OTHERになり、上限の違反も検出されませんでした。CAP_SYS_NICEを外したrootでは、正しいプログラムが上限の違反のProgram_Errorで止まりました。本番と同じユーザーと権限で試す必要があります。

著者プロフィール

記事の著者プロフィールページです。

小村 豪

合同会社小村ソフト 代表

Windows ソフト開発、技術相談、不具合調査を中心に、既存資産が残る案件や原因が見えにくい障害調査に強みがあります。