ＲＴＵのリアルタイム拡張と並列処理
１９９６年１０月２３日 第４版 技術部
目次
§１ 初めに
§２ リアルタイムシステムに求められる機能
２．１ プリエンプション
２．２．１ 割込みの応答時間と予測性
２．２．２ 割込み処理のマルチプロセッサ化
２．２．３ システム資源制御
２．２．４ リアルタイム・スレーブ
§３ リアルタイム・スケジューリング
３．１ 固定優先度ＦＩＦＯスケジューリング
３．２ 固定優先度ラウンドロビン・スケジューリング
３．３ リアルタイムプロセス制御
３．４ イベント駆動型スケジューリング
３．４．１ ＡＳＴ（Ａｓｙｎｃｈｒｏｎｏｕｓ Ｓｙｓｔｅｍ Ｔｒａｐｓ）
３．４．２ 仮想割込み機構
３．５ タイマ割込みによる定周期スケジューリング
３．５．１ クロックＡＳＴ
３．５．２ ＦａｓｔクロックＡＳＴ
３．６ 高精度タイマとサービス
§４ ファイル管理
４．１ 連続領域ファイル
４．２ 非同期入出力
§５ メモリ管理とキャッシュ
５．２ 共有メモリ
５．２．１ ＳＶＲ４対応のシェアードメモリ
５．２．２ ＲＴＵにおける拡張メモリ
５．２．３ システム
５．２．４ スレッドの静的変数
§６ 並列処理
６．１ シンメトリシステムコール
６．２ 多重スレッド
６．２．１ スレッドモデル
６．３ スレッドの生成、起動、終了
６．４ スレッドの同期と排他制御
６．４．１ 排他制御変数（セマフォ）による同期
６．４．２ 条件変数による同期
６．５ その他の同期手段
６．５．１ 計数型排他制御変数（カウンタセマフォ）
６．５．２ 排他書込み多重読みだし
§７ 言語間インターフェース
§８ 例題
８．１ 一定時間ＷＡＩＴ
８．２ 周期タイマ割込み（クロックＡＳＴ）起動
８．３ ＦＯＲＴＲＡＮとＣの間の共有メモリ
８．４ プロセスの停止と再開
８．５ タイマ値取得と起動のオーバーヘッド
８．６ プロセスのプロセッサへの割付
８．７ スレッドによるダブルバッファ法の実現
§９ おわりに
§１０ 参考文献
![[back]](../../images/icons/back.gif)
Back
リアルタイム処理で処理時間に対する要求が厳しい場合には、従来よりスーパーミニコンピュータ上での「並列処理」を必要としてきましたが、それを実現するハードウェアとソフトウェアのアーキテクチャはシステム固有のものでした。
一般的に、ＵＮＩＸは割込みに対する応答時間やコンテキスト・スイッチ時間が長すぎるためリアルタイム処理には向いていません。ところが、ＵＮＩＸをＯＳとしているワークステーションの世界でも、バーチャル・リアリティやマルチメディアシステム、オンライン・トランザクション等におけるリアルタイム機能の要求が強くなってきました。
高速リアルタイムコンピュータの世界でも、高機能産業ロボットやＣＩＭ／ＦＡ等における人工知能応用やマンマシンインターフェースの充実、ネットワークによるエンジニアリング・ワークステーションとの統合等、ＵＮＩＸが得意とする機能が求められています。
それに加え、工業用標準バスとしてＶＭＥｂｕｓが定着したのに伴ってＶＭＥｂｕｓ上でのリアルタイムシステム構築に着手するヘビーユーザも多くなってきました。けれども、ＶＭＥボードコンピュータでは、シングルプロセッサでのパワーの不足、バス上での入出力ボトルネックなどに加え、専用リアルタイムＯＳでのソフトウェア開発のコスト増大などがシステム構築の問題として残ります。
このような、視点からＵＮＩＸシステムをとらえなおすと、標準ＵＮＩＸシステムのスーパセットとしての並列リアルタイムＵＮＩＸシステムが大きな意味をもってきます。それは並列リアルタイムＵＮＩＸを活用すれば、従来のリアルタイム処理分野のみならずビジネス分野においてもすぐれた開発性と高い実行性能によって大きな確実な効果をあげることが可能になるからです。
ＵＮＩＸは、ＵＳＬが開発しライセンスしています
ＰＯＳＩＸは、ＩＥＥＥの商標です。
ＲＴＵはコンカレントの商標です。
その他の本稿で記述された各製品は各社の商標です。
リアルタイムシステムには、満たさなければならないいくつかの基本仕様がありますが、リアルタイム処理をＵＮＩＸで行うには、種々の問題があります。
リアルタイム処理システムでは、優先度の高いプロセスがスケジュールされるまでの最長時間とハードウェア割込みに対する最大応答時間も保証しなければなりませんが、現在最も標準的に使用されているＳＶＲ４やＢＳＤ４には、いずれの時間を保証する機構も提供されていません。それに加え、プロセス管理、メモリ管理、ファイル管理にも多くの問題を含んでいます。本書では、リアルタイム拡張並列ＵＮＩＸ（ＲＴＵ）のリアルタイム拡張について解説します。
リアルタイム処理において、ＵＮＩＸカーネル最大の問題は、プリエンプション可能な構造になっていない点にあります。
これは低優先度プロセスのシステムコール中に、より優先度の高いプロセスが実行可能状態になっても、システムコール終了まで、コンテキスト・スイッチが遅延される現象を引き起こします。
このことは、ＵＮＩＸシステムが応答遅延を引き起こし、リアルタイム処理をまったく実行できないことにほかなりません。
プリエンプション可能であれば、いつでもプロセスを切り換えられますが、ＵＮＩＸカーネルのコード中には、大域変数を数多く使用しているため、同一のシステムコールを発行すると、プリエンプションされたプロセスが使用中の大域変数を破壊してしまいます。
システムコール発行 システムコール終了 システムコール発行 システムコール終了
↓ ↓ ↓ ↓
プロセスＡ ──┐ ┌─ プロセスＡ ──┐プリエンプション・ポイント┌─
（優先度低） │ │ （優先度低） │ ▽ │
プロセスＢ ││← 応答遅延 → ┌┼─┤ プロセスＢ ││←応答遅延→┌┼┐ │
（優先度高） ││ │ │ （優先度高） ││ │<>│ │ │
カーネル └┼─┼─┐┌──┼─┤ │ カーネル └┼─┼─┐├─┘ └───┘
│割込み││スケジュ│ │割込み││スケジュール遅延
割り込み処理 │ 遅延 └┘ール遅延│ 割り込み処理 │ 遅延 └┘
↑ ↑ ↑ ↑
Ｂの起動要求割込み Ｂの割込み処理 Ｂの起動要求割込み Ｂの割込み処理
プリエンプションと応答遅延
ところが、ＵＮＩＸカーネルは、まったくプリエンプション不可能なのではなく、カーネルコード上にプリエンプション可能な部分がいたるところに存在します。
このカーネル中のプリエンプション可能な点は、カーネルのデータ構造を更新し一貫性を保つ事の出来る場所か、排他制御により他のプロセスの活動を封鎖できる領域であり、ＲＴＵではカーネル中の約１００ヶ所に、プリエンション・ポイントを設けています。このプリエンプションポイントが多ければ、それだけスケジュール遅延が短くなります。しかしシステム応答性の決定的な要因はスケジュール遅延ではなく、むしろ割込み遅延時間のほうにあります。
ＲＴＵでは、先に述べた割込み遅延を解消するためにカーネルデータ構造の管理をセマフォ構造体による排他制御で行っています。
つまり複数のセマフォ構造体を導入し、カーネルのデータ構造を最小の単位で排他制御することにより、カーネルのデータ構造が許す最短時間で応答することを可能にしています。
通常このアプローチは、対象型のメモリ共有マルチプロセッサ・システムに用いられ、この構造では、すべてのプロセッサでシステムコール等を同様に処理でき、オペレーティング・システムの負荷分散にもなります。
多重割込み処理を、セマフォ構造体による排他制御に置き換え、すべてのプロセッサで割込みをハンドリングできるように拡張した結果、割込み応答時間は飛躍的に向上し、コンカレント７０００シリーズ（ＭＣ６８０４０ ２５ＭＨｚ）での、アプリケーションタイマの割込み応答遅延時間は、通常１００μ秒程度で応答します。
最高優先度 優先度低
システムコール発行 システムコール終了 ┌───┐ ┌┬┐ 優先度中
↓ ↓ │ │ │││ ┌┬┐
プロセスＡ ──┐ ┌─ └───┼───┐ │││ │││
（優先度低） │ │ │ │ │ └┴┘ │││
プロセスＢ ││応答┌┼┐ │ 処理 │ │ │ └┴┘
（優先度高） ││遅延│ │ │ ↓ │ │ │
カーネル └┤┌─┘ └───────┘ │ │ │ ┌┬┐
││ △ △ △ △ △ │ │ │ ┌┬┐ │││
割り込み処理 └┘プリエンプション・ポイント ┌───┼───┘ │││ │││
↑ │ │ │││ └┴┘
Ｂの起動要求割込み Ｂの割込み処理 └───┘ └┴┘
リアルタイム
ＲＴＵの応答遅延 プロセス 割込み処理
ここで考えなければならないもうひとつの問題は、イベントに対して一定時間内での応答を保証することです。
システムの応答性は、ＵＮＩＸカーネルに対して多くのプリエンプションポイントを設けることやセマフォ構造体を導入することによって向上しました。
しかしこのアプローチだけでは、応答時間の保証はできません。この理由は、システムトラップやデバイスドライバ等の割込みの処理を阻止できないからです。
割込み処理を阻止するのは、データ構造の一貫性を保証するために、非割込み部が割込み処理部の設定したデータをアクセスする場合に限られます。
この状態でも、ハード的に優先度の高い割込み処理は、多重に割り込んできます。
したがって総ての割込みを１つのプロセッサで処理すると下図のように応答時間が予測不可能になってしまいます。
割込み発生 プロセススタート
│ │ │ │ │ │ │
│高優先度│低優先度│システム│カーネル│ コンテキスト・スイッチ時間 │ │
│割込処理│割込処理│トラップ│スケジュ│ │ │ │プロセ│
│サービス│サービス│サービス│ ール│コンテキ│スケジュ│コンテキ│ス起動│
│ 時間 │ 時間 │ 時間 │遅延時間│スト待避│ーリング│スト復活│ 時間 │
│ │ ↑ │ ↑ │ ↑ │ │ │ │ │
│ │ │ └─ プリエンプションポイントの効果 │
│ │ └─ 8.3 ms か 16.7 ms に１回必ず起こる │
│ └─ 多重に割込みを受け付けるので時間の予測は不可能 │
│ │
│←──────────────割込み応答時間───────────────→│
シングルプロセッサでの割込み応答時間
前述のように、シングルプロセッサでは応答時間を保証することができないため、ＲＴＵではマルチプロセッサにすることで多重割込みの弊害をなくし、重要な外部イベントに対して一定の時間内の応答を保証することを考えました。このアプローチはＲＴＵ独自のアイデアで、これを実現するために、次の３点を可能にしています。
１）割込み処理のプロセッサ割付を自由に変更できる。
２）入出力処理をシンメトリックに処理する。
３）割込み処理をスレッドを使用して非割込み処理の環境で実行する
１）の機能は、７０００シリーズだけがソフトウェアで実行可能で、その他のシリーズではバックプレーンの変更を必要とします。７０００シリーズでは、アプリケーション・タイマ割込みのために用いられています。
つまり７０００シリーズでの、アプリケーション・タイマの割込み処理は完全に並列実行されるわけです。
２）の機能は、すべてのシリーズで使用可能です。ＲＴＵでは、システム・コンフィグレーション・ファイル中で下の例のように、デバイスドライバの（カーネル上位部）処理を実行するプロセッサを指定できるようになっています。（プロセッサマスクが−１の場合実装されているすべてのプロセッサで処理されます。）
lance lcl:00300000 プロセッサマスク
３）の機能はＵＮＩＸが割込み処理ルーチンからコンテキスト・スイッチを起こせないために導入された機能です。
通常、割込み処理ルーチンは、自分自身のコンテキストを持っていないために、割り込んだプロセスのコンテキストを使用して割込み処理を実行しています。このため割込み処理ルーチンでは、処理を終了するまで待ちに入る事を許されません。
ところが、割り込まれたプロセスが非常に優先度の高い処理であり、割込み処理ルーチンがもっとも低いプロセスのためであっても、割込み処理とプロセスの実行優先度には、なんの関係付けも行われていないため、割り込まれたプロセスの優先実行は、なんら保証されません。
つまり、”あらゆる割込み処理は、すべてのプロセスに対して優先的に実行される”という、暗黙の制約条件がついていると考える事ができるわけです。
したがって、リアルタイムの応答性を保証するには、カーネル下位部での割込み処理を長時間続けてはならないことになります。
しかし、現実のデバイス・ドライバでは、ステータス・レジスタ等のチェックなどに時間がかかり、短時間で終了できないケースがままあります。
この問題は、割込み処理においてコンテキストスイッチを起こすか、割込み処理を非割込み処理に移して処理させれば、割込み処理を行うプロセスと割り込まれたプロセスの間で、優先順位による調停が可能になるため、問題は解決できます。
ＲＴＵでは、割込み処理を非割込み処理に移すために "System Light Weight Prosess"（ＳＬＷＰ）と、"Dedicated Light Weight Prosess"（ＤＬＷＰ）を導入しています。
このような機構によって下図のように、割込みを受け付けたブートプロセッサが他のプロセッサに対して処理を動的に移動する事ができ、割込みの応答性を保証することができます。ＲＴＵではこのようにマルチプロセッサを利用してプリエンプションを高速に行う手法をハイブリッド・プリエンプション機構と呼びます。
割込み発生 プロセス・スタート
│ │ │ │
│高優先度│低優先度│ │
ブートプロセッサ│割込処理│割込処理│ │
│サービス│サービス│ │
│ 時間 │ 時間 │ │
├────┼────┼────┬──────────────┬───┤
│ └─→│システム│カーネル│ コンテキスト・スイッチ時間 │ │
│処理の │トラップ│スケジュ│ │ │ │プロセ│
非ブートプロセッサ│マイグレ│サービス│ ール│コンテキ│スケジュ│コンテキ│ス起動│
│ーション│ 時間 │遅延時間│スト待避│ーリング│スト復活│ 時間 │
│ │ │ │ │ │ │ │
│←────────────割込み応答時間────────────→│
マルチプロセッサでの割込み応答時間
ＲＴＵにおける、プロセッサ割付けやシステムクロックの制御などのシステム資源制御は、次のような構文の mpadvise システムコールで実現し、その機能は function フィールドで指定します。
┌─────────────────────────────────────────────┐
│ ｍｐａｄｖｉｓｅ：マルチプロセッサとカーネルのシステム資源の制御を行なう │
│ │
│ 形式 │
│ │
│ #include │
│ │
│ int mpadvise(function,cpu_mask[,arg]) │
│ int function,cpu_mask,arg │
│ │
│ 引数 │
│ │
│ function : MPA_CPU_EXIST システムに実装されているすべてのプロセッサを得る │
│ MPA_CPU_GET 非リアルタイムで使用できるプロセッサを得る │
│ MPA_CPU_SET 非リアルタイムで使用するプロセッサを cpu_mask で指定する │
│ MPA_RT_GET リアルタイムで使用できるプロセッサを得る │
│ MPA_RT_SET リアルタイムで使用するプロセッサを cpu_mask で指定する │
│ MPA_60HZ_ON cpu_mask のプロセッサのシステムトラップを許可する │
│ MPA_60HZ_OFF cpu_mask のプロセッサのシステムトラップを禁止する │
│ MPA_CLK_GET cpu_mask のプロセッサのシステムトラップクロックを得る │
│ MPA_CLK_SET cpu_mask のプロセッサにシステムトラップクロックをセットする │
│ MPA_TBINV_ON ＴＬＢの無効割込みを許可する（＊ │
│ MPA_TBINV_OFF ＴＬＢの無効割込みを禁止する（＊ │
│ │
│ cpu_mask : 図に示すようなビット指定で使用するプロセッサを指定する。 │
│ │
│ MSB boot LSB │
│ ┌─┬┬─┬─┬─┬─┬─┬─┬─┬─┬─┬─┐ │
│ │ ││ │ │７│６│５│４│３│２│１│８│ │
│ └─┴┴─┴─┴─┴─┴─┴─┴─┴─┴─┴─┘ │
│ プロセッサ・マスク │
│ │
│ arg : MPA_CLK_SET システムトラップの周期を指定する（６０〜１０００Ｈｚ） │
└─────────────────────────────────────────────┘
この mpadvise を使用して、システムトラップの周期を変更できます。システムトラップの周期が短ければ、ＵＮＩＸでの非同期イベントの認識は、図のように次のシステムトラップまで遅延されますので、イベントの認識を高速にできる効果があります。
if ( mpadvise(MPA_CPU_SET,cpu_mask) ) {
perror("ＴＳＳプロセッサの指定");
exit(1);
}
if ( (freq = mpadvise(MPA_CLK_SET,cpu_mask,1000))==-1 ){/*システムトラップを１０００Ｈｚにする*/
perror("システムトラップ");
exit(3);
}
イベントの認識 × × × × × × × × × ×
│ │ │ │ │ │ │ │ │ │
イベントの要求 │ │ │ │ │ │ │ │ │ │
─┬┴┬─┼─┬┴┬─┬─┼─┬─┬┴┬┴┬─┼─┬┴┬─┼─┬┴┬─┬─
システムトラップ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │
→時間
（＊ ＲＴＵでは、プロセステーブルのような可変データに対する仮想アドレスは、動的に割り当てられるため、プロセスが生成／消去されるたびに、ブートプロセッサから他のプロセッサに割り込みを発生し、各プロセッサが独自に管理しているマップテーブルが無効になったことを知らせている。このシステムコールによって、予めこの領域を割り付けておき、ＴＬＢ無効割り込みによる処理の中断をなくすことができる。（システムコンフィグレーション時にkmempoolを定義する必要あり）。
非同期イベントの認識には、システムトラップの周期を短くすることが大きな効果を生みますが、逆にシステムトラップを遅くすることで、システムオーバヘッドのを小さく抑えることができます。
ＲＴＵでのシステムトラップ周期は、デフォルトで１２０Ｈｚ（８．３ミリ秒）ですが、６０Ｈｚ〜１０００Ｈｚまで変化させることが可能で、非ブートプロセッサに限っては、システムトラップを停止することも出来ます。
この機能は、特に割込みでプロセスを直接起動する同期イベントや高速に演算を行ないたいプロセスにとっては、非常に効果があり、システムオーバーヘッドを最小にすることができます。
つまり、定周期システムトラップを停止し、他のプロセスをプロセッサから追い出すことによって、カーネルのプリエンプション時間とコンテキスト・スイッチング時間とを省略し、オーバーヘッド時間を最小にすることが可能になります。
割込み発生 プロセス・スタート
│ │ │ │
│ │ コンテキスト │ │
│割込処理│ スイッチ時間 │プロセ│
非ブートプロセッサ│サービス│ │ │ス起動│
│ 時間 │スケジュ│コンテキ│ 時間 │
│ │ーリング│スト復活│ │
│ │ │ │ │
│ │
│←─────割込み応答時間────→│
リアルタイム・スレーブ応答時間
ＲＴＵと７０００シリーズでは、すべての割込み信号をソフトウェアで指定したプロセッサへ直接入力することができ、他の影響を全く受けない環境（リアルタイムスレーブと呼びます）を構築できるよう設計され、ＦａｓｔクロックＡＳＴとしてユーザに提供されています。
このようなリアルタイムスレーブに設定するためには、以下のようにプログラムすることにより、非ブートプロセッサでのみ使用できます。この理由はおもに、システムトラップのためのクロック割込みを禁止する理由によります。
if ( mpadvise(MPA_RT_SET,cpu_mask) ) {
perror("リアルタイムプロセッサの設定");
exit(2);
}
if ( mpadvise(MPA_60HZ_OFF,cpu_mask) ) {
perror("システムトラップの停止");
exit(2);
}
if ( mpadvise(MPA_CLK_SET,cpu_mask,0) ) {
perror("アプリケーションクロックの禁止");
exit(3);
} この部分は、必ずしも必要ではない
if ( mpadvise(MPA_TBINV_OFF,cpu_mask) ) {
perror("ＴＬＢ無効割込みの禁止");
exit(4);
}
標準ＵＮＩＸとリアルタイムＯＳを比べた場合、スケジューリングの考え方に多くの相違点があります。
一般的なＵＮＩＸでは、５０Ｈｚ〜１２０Ｈｚのクロックでシステムトラップを発生させこのタイミングで実行状況を把握し、１秒ごとにそのデータベースを使って動的スケジューリングを行ないます。
このとき、過去から現在に渡ってプロセッサを占有しているプロセスは、優先順位を下げられ、結果として長時間連続して動作するようなプロセスは実行が後回しになり、ｌｓやｋｉｌｌなどの小さなプロセスの実行が優先されます。したがって実行権を与えられるまでの時間と実行終了の時間は予測できません。これは公平性に重きをおいた設計で、プログラム開発システムとしてのＵＮＩＸとしては妥当な設計でした。
しかしリアルタイムシステムでは、プロセスの応答時間に対する保証をしなければなりません。すなわち制限時間や優先度に従った固定優先度スケジューリングを行なわなくてはなりません。
ＵＮＩＸのスケジューリングでのプロセス優先度は、１秒ごとにＵＮＩＸカーネルによって、再スケジューリングされ、刻々と変化しますが、固定優先度はカーネルによってスケジューリングされません。
ＳＶＲ４と、ほとんどのリアルタイムＵＮＩＸシステムは、この固定優先度スケジューリングをカーネルに取り込んでいます。
ＲＴＵでは、標準ＵＮＩＸのｎｉｃｅ値（−２０〜＋１９）のうち一部（−２０〜−１１）をリアルタイムプロセス用の固定優先度として確保しています。
さらに、ＲＴＵでは、後述のリアルタイムプロセス専用のスケジューリング機構を持っています。
このスケジューリング・アルゴリズムでは、ＵＮＩＸカーネルのスケジューリングとは別に、リアルタイムＦＩＦＯ、リアルタイムラウンドロビンとタイムシェアリングを選択できるようになっています。
優先度 ポリシー 優先度低 優先度高 システムコール
優先度 ──┬─可変────┬─ラウンドロビン 19 -10 nice (SVR4)
│ └─SCHED_OTHER MIN_TS_PRI(0) MAX_TS_PRI(28) schedset (RTU)
│ (SCHED_TS)
└─固定────┬─ＦＩＦＯ -11 -20 nice (RTU)
│
├─SCHED_FIFO MIN_RT_PRI(0) MAX_RT_PRI(63) schedset (RTU)
└─SCHED_RR MIN_RT_PRI(0) MAX_RT_PRI(63) schedset (RTU)
ＲＴＵのプロセス・スケジューリング
このスケジューリングポリシーでは、優先度の高いプロセスが実行権を持っている場合、そのプロセスが（システムコール発行等で）待ちの状態になるか、より高い優先度を持ったプロセスが実行可能状態にならない限り実行権を放棄しません。同一の優先度の場合ＦＩＦＯキューで、先に実行されたプロセスの終了を待ちます。
ＰＯＳＩＸで提案されている固定優先度スケジューリングもこのメカニズムを採用しています。
ＲＴＵでは、すでに述べた nice 値の拡張に加えて、リアルタイムプロセス専用のシステムコール schedset で６４レベルの優先度スケジューリングを可能にしています。
優先度 クオンタム 計算量 ↓ ↓ ↓ ↓
プロセス Ａ 中位 なし ［５単位時間終了］ ┌────┐
プロセス Ｂ 低位 なし ［７単位時間終了］ │ └───────
プロセス Ｃ 高位 なし ［３単位時間終了］───┘
固定優先度スケジューリング
このスケジューリングポリシーでは同じ優先度のタスクに指定時間だけプロセッサの使用権を与えます。もし優先度の高いプロセスが実行可能になるとこちらが優先され、そのプロセスの実行が終了後、元のタスクの途中から実行を続けます。
ＳＶＲ４のカーネルでは、大域リアルタイム優先度（１５９〜１００）として実装されています。ＳＶＲ４のリアルタイム優先度をもったプロセスは、設定したクオンタム時間だけは、スケジューラに邪魔されず実行を続けることができますが、この時間を使いきるとシステムがトラップを発生し、スケジューラは他に優先度の高いプロセスが実行できるかどうかをチェックします。このクオンタム時間は、優先順位に関係づけてカーネル生成時にパラメータ設定するため動的に変化させることはできません。
ところで、リアルタイム処理のクオンタム時間は、少なくとも１μ秒程度の精度がないと実際の使用には役にたちませんが、典型的なＳＶＲ４のタイムシェアリングプロセスの単位時間（クオンタム）は１００Ｈｚつまり１０ｍ秒でしかありません。これでは、十分なリアルタイム処理は不可能です。
ＲＴＵでは、後述する６４０ｎｓの解像度を持ったシステムタイマを使用して、６４０ｎｓ〜６０分の任意の時間でラウンドロビンスケジューリングを制御できます。
優先度 クオンタム 計算量 ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
プロセスＡ 低位 １ ［４単位時間終了］ ┌────
プロセスＢ 中位 １ ［５単位時間終了］ ┌┐┌┐┌┐┌┐┌┐ │
プロセスＣ 中位 １ ［７単位時間終了］ │└┘└┘└┘└┘└──┘
プロセスＤ 高位 １ ［３単位時間終了］───┘
固定優先度ラウンドロビン・スケジューリング
ＲＴＵにおける、リアルタイムプロセス制御は、下表のシステムコールで制御します。
そのなかでもＲＴＵ独自のリアルタイム・スケジューリングは、次のような構文の schedset/schedget システムコールで指定します。
┌────────────────────────────────────────────────┐
│ｓｃｈｅｄｓｅｔ／ｓｃｈｅｄｇｅｔ スケジューリング・ポリシーの制御をする │
│ │
│形式 │
│ │
│ #include │
│ #include │
│ │
│ int schedset(pid,ppolicy,pprio,pquantum) │
│ int schedget(pid,ppolicy,pprio,pquantum) │
│ │
│引数 │
│ int pid; プロセスＩＤ（０の場合呼びだしたプロセス自身） │
│ int *ppolicy; SCHED_TS(SCHED_OTHER) タイムシェアリング・スケジューリング │
│ SCHED_FIFO リアルタイム・先入れ先だし・スケジューリング │
│ SCHED_RR リアルタイム・ラウンドロビン・スケジューリング │
│ int *pprio; MIN_TS_PRI 〜 MAX_TS_PRI タイムシェアリング・スケジューリングと併用する(0〜28)│
│ MIN_RT_PRI 〜 MAX_RT_PRI リアルタイムスケジューリングと併用する(0〜63) │
│ struct timestruc { │
│ unsigned long int ts_sec; 秒単位 │
│ long int ts_nsec; ナノ秒単位 │
│ } *pquantum; MIN_RT_QUAN 〜 MAX_RT_QUAN システム・クロックの解像度から６０分まで │
│ int nice(prio) −１１から−２０が固定優先度ＦＩＦＯスケジューリング │
└────────────────────────────────────────────────┘
次のプログラムは、セカンドプロセッサでラウンドロビンを行う例で、プロセスは、１０ミリ秒のラウンド・ロビン・スケジューリングを設定し子プロセスを生成します。
子プロセスは、ＵＮＩＸのセマンティクスに従い、親プロセスの環境を引き継いでいますので、親プロセスとまったく同じ値を持っている事をプログラムで確認します。
#include /* schedset(2) を使用 */
#include /* timestruc 構造体を定義*/
#include /* mpadvise(2) を使用 */
#include /* NULL を定義 */
#define RT_MASK 0x4 /* セカンドプロセッサ */
#define CLK_FREQ 100 /* システムトラップ周期 */
#define PROCESS_SLICE 10000000 /* １０ミリ秒をナノ秒単位で宣言 */
/* もし PROCESS_SLICE よりクオンタム時間がより小さい場合,タイミングは不正確になることがある
これは、リアルタイムプロセッサにおける CLK_FREQ のクロック割込み周期に依存する
*/
main ()
{
int pidnew, /* 子プロセスのプロセスＩＤ用の変数 */
pid, /* 親プロセスの wait(2) における関数の戻り値用の変数 */
mask, /* システムプロセッサマスク値用の変数 */
status; /* wait(2) 用のステイタス変数 */
int policy; /* schedget/schedsetで使用するスケジューリングポリシー用の変数*/
int prio; /* schedget/schedsetで使用するスケジューリング優先度用の変数*/
struct timestruc quantum;/* schedget/schedsetで使用するクオンタム時間用の変数*/
/* これから変更するために schedget を使用して現在の値を出力する。
プロセスＩＤが NULL なのでこのシステムコールはカレント・プロセス／スレッドを意味する*/
if (schedget (NULL, &policy, &prio, &quantum) < 0)
perror ("schedget"), exit (1);
/*リアルタイム、１０ミリ秒でのラウンドロビン・スケジューリングを設定するために変数を初期化する*/
policy = SCHED_RR; /* ラウンドロビン・スケジューリング・ポリシー */
prio = MIN_RT_PRI; /* 最小リアルタイム優先度 */
quantum.ts_sec = 0;
quantum.ts_nsec = PROCESS_SLICE; /* プロセスのクオンタム時間は１０ミリ秒 */
if (schedset (NULL, &policy, &prio, &quantum) < 0) /* スケジューリングをセットする */
perror ("schedset"), exit (2); /* もしエラーならば exit する */
mask = mpadvise (MPA_CPU_EXIST, 0); /* 現在のシステムのプロセッサ資源を得る */
/* もし望むプロセッサがシステムに存在すれば、リアルタイムプロセッサとして確保し、
そのプロセッサにプロセスを移動する */
if (mask & RT_MASK)
{
if (mpadvise (MPA_RT_SET, RT_MASK) < 0)
perror ("mpadvise MPA_RT_SET"), exit (3);
/* １０ミリ秒のクオンタム時間を実現するために、システム・トラップ時間を変更する */
if (mpadvise (MPA_CLK_SET, RT_MASK, CLK_FREQ) < 0)
perror ("mpadvise MPA_CLK_SET"), exit (4);
}
else
printf ("Running on a uniprocessor\n");
/* 変更したスケジューリングポリシーの各値を出力する */
if (schedget (NULL, &policy, &prio, &quantum) < 0)
perror ("schedget"), exit (5);
if (pidnew = fork ()) /* 子プロセスを生成する*/
{
pid = wait (&status);
if (status & 0xff) /* 子プロセスの終了ステイタスをチェックする */
{
fprintf (stderr, "Child exited: signal %d\n", status);
exit (10);
}
else
exit (status >> 8);
}
else /* 子プロセス */
{
int new_policy, new_prio;
struct timestruc new_quantum;
/* 子プロセスが親プロセスのスケジューリングを引き継いでいるか確認する */
if (schedget (NULL, &new_policy, &new_prio, &new_quantum) < 0)
perror ("schedget"), exit (5);
if (new_policy != policy)
fprintf (stderr, "Policy was not inherited\n"), exit (6);
if (new_prio != prio)
fprintf (stderr, "Priority was not inherited\n"), exit (7);
if (new_quantum.ts_sec != quantum.ts_sec ||
new_quantum.ts_nsec != quantum.ts_nsec)
perror ("Time quantum was not inherited\n"), exit (8);
if (mpadvise (MPA_RT_GET, 0) != RT_MASK)
perror ("Processor mask was not inherited\n"), exit (9);
printf ("Round-robin child with mask %#x\n",
mpadvise (MPA_RT_GET, 0));
exit (0);
}
}
またＲＴＵでは、リアルタイム・優先度を持ったプロセスのスケジューリングを制御する、hold,release,yield といったシステムコールも実装しています。
┌────────────────────────────────────────────────┐
│ｈｏｌｄ，ｒｅｌｅａｓｅ，ｙｉｅｌｄ プロセス・スケジューリングの制御 │
│ │
│形式 │
│ │
│int hold() リアルタイムプロセスをスケジューリングからはずす │
│int release(pid) ホールド状態のプロセスを解放し実行状態にもどす │
│int yield() リアルタイムプロセスをＦＩＦＯの最後に移動し再スケジューリングする │
└────────────────────────────────────────────────┘
以下のプログラムは、セカンドプロセッサに２つのリアルタイムプロセスを生成し、hold,release で同期をとるものです。
親プロセス mpadvise mpadvise fork schedset 処理 release wait exit
子プロセス schedset hold exit
#include
#include
#define cpu_mask 0x4
main()
{
int pid=0,status,ppolicy=SCHED_FIFO,pprio=MIN_RT_PRI,cprio=MIN_RT_PRI+1;
if ( mpadvise(MPA_RT_SET,cpu_mask) ) {
perror("リアルタイムスレーブ機能");
exit(1);
}
if ( mpadvise(MPA_60HZ_OFF,cpu_mask) ) {
perror("システムトラップの禁止");
exit(2);
}
if (pid = fork()) { /* 子プロセス */
if (schedset(pid,&ppolicy,&cprio,NULL)){
perror("スケジュールセット");
exit(2);
}
hold();/*親プロセスを待つ*/
/* 計算等 */
exit();
} else {/* 親プロセス */
if (schedset(pid,&ppolicy,&pprio,NULL)){
perror("スケジュールセット");
exit(2);
}
sleep(10);/*実際は前処理等*/
release(pid);
/* 計算等 */
wait(&status);
}
}
割込み駆動型とも呼ばれるスケジューリングで、一般的なＵＮＩＸではキーボード割込み等がこの部類に入ると考えられます。つまりなにか非同期に発生する事象をとらえてプロセスを起動する機構です。
標準ＵＮＩＸのシステムコールでいえば、ｓｉｇｎａｌ，ｋｉｌｌ等がこれに当たりますが、実際には現実のデバイスと結びつかないと意味がないため、標準デバイス以外は、デバイスドライバ・レベルのプログラムが必要になります。
しかし、標準デバイスでもＵＮＩＸにおける非同期事象に対する扱いは非常に手荒く、割込みをユーザプロセスに伝える機構であるｓｉｇｎａｌには、下記のような広く知られた欠点があります。
１）キューイングできないため、シグナルを失うことがある。
２）優先度付けができないため緊急度のあるものが後回しになる恐れがある。
３）総数(32)が決まっていてユーザ定義は２つ(SIGUSR1,SIGUSR2)しかない。
４）データ授受ができないため、単純な処理しかできない。
５）受信プロセスは送信プロセスからの割込みをブロックできない。
この非同期イベントの通知の問題に対してＰＯＳＩＸ提案では、キューイング機構と予測可能な時間内でのイベント通知、データの授受等を求めています。
ＲＴＵでは、ｓｉｇｎａｌにみられる欠点を解決するために以下の特徴を持ったＡＳＴ（Ａｓｙｎｃｈｒｏｎｏｕｓ Ｓｙｓｔｅｍ Ｔｒａｐｓ）と呼ばれる非同期イベントの通知機構を実装しています。（ＤＥＣ版ＢＳＤのＡＳＴとは意味が異なります）
１）送信した複数のＡＳＴはキューイングされるためシグナルを失うことがない。＊）
２）ハードウェア割込みのように、優先度を設定できるため緊急度のあるものが後回しになることがない。
３）送信できるＡＳＴの種類（数）は事実上制限がない（０〜３２７６７）
４）３２ビットデータ授受ができるため、１つのＡＳＴでも複数の処理が可能
５）受信プロセスは送信プロセスからの割込みをブロックできる。
ＡＳＴは、外部イベントとの同期はもとより、プロセス／スレッド間の通信、状態遷移に使用できるため、ユーザはリアルタイムアプリケーションを簡単かつ安全に保証された時間内で、起動させることが可能になっています。
ＡＳＴは、以下のシステムコールで、外部プロセス／スレッドから指定したプロセス／スレッドの関数を起動できます。
＊） 実際のキューイングサイズはシステムコンフィグレーション・パラメータ nast に制限されています（デフォルトは、１００）
┌────────────────────────────────────────────────┐
│＿ｓｅｎｄａｓｔ ＡＳＴ（非同期システムトラップ）の送信 │
│ │
│ #include │
│ │
│形式 │
│ │
│ int _sendast(pid,lo_handler,func,param,priority); │
│ │
│引数 │
│ │
│ int pid; 受信プロセスのプロセスＩＤ │
│ int (*lo_handler)(); 通常は _asthandler() を指定する │
│ int (*func)(); システムトラップ関数 │
│ int param; 送信パラメータ │
│ int priority; ０〜３２７６７の値を持つＡＳＴ優先度 │
└────────────────────────────────────────────────┘
pid に０を指定すると、送信プロセス／スレッドと同一のプロセスグループ識別子を持つ pid が０と１を除くすべてのプロセス／スレッドに送信されます。
起動されるプロセス・スレッドは、何か処理を実行していても良いし、処理が無い場合は、以下のシステムコールで待っていてもかまいません。
この場合、受信プロセス／スレッドは、ＡＳＴに対しての優先度を設定することにより、指定優先度以下のＡＳＴを禁止することが可能になっています。
また、ＡＳＴによって起動される割込み関数は、次の３つのパラメータを引き渡されます。
func(param,pc,pri)
int pram; /* _sendast() で引き渡されたパラメータ */
int pc; /* 割り込んだプロセスのプログラムカウンタ */
int pri; /* 受信したＡＳＴのＡＳＴ優先度 */
{
}
┌────────────────────────────────────────────────┐
│ａｓｔｐａｕｓｅ ＡＳＴ優先度設定とプロセスの停止 │
│ │
│形式 │
│ │
│ int astpause(priority,timeout); │
│ │
│引数 │
│ │
│ int priority; ０〜３２７６７の値を持つＡＳＴ優先度 │
│ int timeout; タイムアウト時間をミリ秒で指定する（実解像度はシステムトラップに依存）│
└────────────────────────────────────────────────┘
astpause の priority が正の値を持つ場合、待ちの間ＡＳＴ優先度を priority まで引き上げますので、送信プロセスのＡＳＴ優先度が低い場合ＡＳＴはキューイングされてしまいます。負の値を持つ場合、ＡＳＴ優先度は、変更されません。関数はタイムアウトを検出すると−１を返し、タイムアウト前にＡＳＴを検出するとタイムアウトまでの残り時間を返します。
プロセス実行中のＡＳＴ優先度を変更する場合には、setpri システムコールを使用します。これにより priority 以下のＡＳＴ優先度（同一優先度を含みます）を持つＡＳＴの送信は、キューイングされます。
┌────────────────────────────────────────────────┐
│ｓｅｔｐｒｉ ＡＳＴ優先度の設定 │
│ │
│形式 │
│ │
│ int setpri(priority); │
│ │
│引数 │
│ │
│ int priority; ０〜３２７６７の値を持つＡＳＴ優先度 │
└────────────────────────────────────────────────┘
次のプログラムは、２つのプロセッサにプロセスを割当て、親プロセスから子プロセスに対してＡＳＴを送信するプログラムです。
このとき、親プロセスは、ブート・プロセッサでＴＳＳスケジューリングですが、子プロセスは、セカンド・プロセッサ上で、リアルタイム・スレーブに設定されます。
このとき子プロセスの実行ＡＳＴ優先度は０で、送信ＡＳＴ優先度は１０であるため、優先度から子プロセスが実行状態でも割込み駆動で関数が起動されます。
リアルタイム・スレーブと非同期システム・トラップ例
#include /* mpadvise(2) に必要*/
#include /* #define NULL を使用*/
#include /* rtplock(2) に必要 */
#include /* schedset(2) に必要 */
#define NON_BOOT 0xFD /* ブートプロセッサ(0x2)以外のプロセッサマスク */
#define CHILD_CPU 0x4 /* 子プロセスは、セカンドプロセッサで実行 */
#define AST_PRI 10 /* ＡＳＴ優先度 */
#define END_TEST 4 /* ＡＳＴの送信数 */
int ast_rcvd; /* 子プロセスで受信したＡＳＴの数を格納する変数*/
main ()
{
int ast_sent; /* 親プロセスで送信したＡＳＴの数を格納する変数 */
int mask; /* システムに存在するプロセッサを格納する変数 */
int pidchi; /* 子プロセスＩＤ格納変数 */
int status; /* 子プロセス終了ステータス格納変数/
int _asthandler (), resume_proc (); /* ＡＳＴハンドラ */
mask = mpadvise (MPA_CPU_EXIST, 0); /* システムに存在するプロセッサ数を得る */
if (!(mask & NON_BOOT)) /* もしブートプロセッサのみなら終了する */
fprintf (stderr, "Program must run on a multiprocessor\n"), exit (1);
if (pidchi = fork ()) /* 親プロセス */
{
/* 子プロセスに対してＡＳＴを送信する */
for (ast_sent = 1; ast_sent <= END_TEST; ++ast_sent)
{
printf ("Parent sending AST; ast_sent = %d\n", ast_sent);
if (_sendast (pidchi, _asthandler, resume_proc,
ast_sent, AST_PRI) == -1)
perror ("_sendast"), exit (2);
astpause (-1, 100);
}
wait (&status);
if (status)
fprintf (stderr, "Child failed, status = %d\n", status), exit (3);
else
printf ("parent exiting\n"), exit (0);
}
else /* Child */
{
int i;
int policy = SCHED_FIFO;
int prio = MIN_RT_PRI;
/* 固定優先度ＦＩＦＯスケジューリングのスケジューリングを設定 */
if (schedset (NULL, &policy, &prio, NULL) < 0)
perror ("schedset"), exit (4);
if (mpadvise (MPA_RT_SET, CHILD_CPU) < 0)/* セカンドプロセッサで実行 */
perror ("Child MPA_RT_SET"), exit (5);
/* １２０Ｈｚのシステムトラップを停止 */
if (mpadvise (MPA_60HZ_OFF, CHILD_CPU) < 0)
perror ("MPA_60HZ_OFF"), exit (6);
/* プロセスセグメントを主記憶に常駐させる */
if ( rtplock(PROCLOCK) < 0)
perror("rtplock"), exit (7);
/* ＡＳＴ優先度を０に設定する */
setpri (0);
printf ("Child entering loop\n");
/* 受信ＡＳＴの数が END_TEST になるまでループする。常に実行状態にある事に注意 */
while (ast_rcvd < END_TEST)
;
printf ("child exiting\n"), exit (0);
}
}
resume_proc (current) /* 送信プロセスから送信ＡＳＴ番号が送信されてくる */
int current;
{
printf ("resume_proc: current = %d, ast_rcvd = %d\n",
current, ++ast_rcvd);
}
ブートプロセッサ セカンドプロセッサ
◎
┌────────────────┐
│ ブートプロセッサに移動 │
└────────────────┘
┌────────────────┐
│ ｆｏｒｋ │
└────────────────┘
┌──────────────┐
│ リアルタイムスレーブに設定 │
└──────────────┘
┌──────────────┐
│ ＡＳＴ優先度を０に設定 │
└──────────────┘
→
無限ループ
ＡＳＴは優先度を持ち ＡＳＴ優先度０で実行中
キューイング機能がある
この部分は割込み処理として実行される
┌────────────────┐
│ ＡＳＴ優先度１０のＡＳＴを送信 ├───────── ＡＳＴの受信と実行
└────────────────┘ＡＳＴ優先度１０
┌────────────────┐
│ 子プロセスの終了を待つ │
└────────────────┘
ＡＳＴは、受信ハンドラの関数アドレスを送信側で指定するため、_sendast() を使用するときには、十分注意して使用する必要があり、無関係なプロセス同士の通信には、ｓｉｇｎａｌかここで述べる仮想割込み機構を使用する必要があります。
仮想割込み機構は、プロセス間通信におけるバーチャル・サーキットの割込み版ともいえるもので、送信部と受信部が互いのインプリメントに影響を受ける事なく、抽象的な名前でコネクションを確立させる事が可能になります。
仮想割込み機構には、次の２つのインターフェースが存在します。
・ユーザプロセス
・ＲＴＵカーネル／デバイスドライバ
ここでは、ユーザプロセスのインターフェースのみについて説明をします。ＲＴＵカーネル／デバイスドライバの説明については、"Guide to Writing a UNIX Device Driver" を参照してください。
仮想割込み機構は、まず仮想割込みチャネルに対して仮想的なタグ名をつけて生成することによって使用できるようになります。
チャネルは通常、作成したユーザプロセス、デバイスドライバ、ＲＴＵカーネルによって所有され、作成者が存在するかぎり、チャネルは存在します。
ただし、チャネルは、ＲＴＵカーネルに引き渡す事ができますので、その場合は明示的に削除するかシステムがリブートするまで存在します。以降のチャネル接続のための属性や権利は、チャネル作成時に指定する事ができます。
チャネルの作成は、以下のような構文で生成されます。生成に失敗すると関数は−１を返し、成功するとチャネルＩＤを返します。
if ((chan = vi_create ("chnnel_name", 1, VIP_DORMANT)) == -1) {
perror ("could not create the virtual interrupt channel");
exit (1);
}
┌────────────────────────────────────────────────┐
│ｖｉ＿ｃｒｅａｔｅ 仮想割込みチャネルの生成 │
│ │
│形式 │
│ │
│#include │
│ │
│VICHANID vi_create(name,maxcon,perm,flags,arg) │
│ │
│引数 │
│ │
│unsigned char *name; チャネルを指定するシステムでユニークなタグ名を設定 │
│ （MAXPATHLEN以下でなくてはならない） │
│int maxcon; 一度に接続できるチャネルの最大接続数（デフォルトは 30 ） │
│unsigned int perm; チャネル作成の属性と権利を以下のように定義している │
│ VIP_SO_ANY チャネルに対して制約の無い発信接続を選択します │
│ VIP_SO_PRIV 特権プロセス（ルートＩＤ）用のソースチャネルに対して接続する │
│ VIP_SO_KERNEL カーネル用のソースチャネルに対して接続する │
│ VIP_SE_ANY ユーザプロセス用のチャネルをセンスする │
│ VIP_SE_PRIV 特権プロセス（ルートＩＤ）用のチャネルをセンスする │
│ VIP_SE_KERNEL カーネル用のチャネルをセンスする │
│ VIP_DORMANT チャネルを停止状態にする。この結果、vi_map()によるアクセスが禁止される │
│ チャネルを活性化してvi_map()を成功させるためにはvi_ctl()が必要である │
│unsigned int flags; 将来の拡張用に予約 │
│unsigned int arg; 将来の拡張用に予約 │
└────────────────────────────────────────────────┘
生成したチャネルは、vi_delete()で削除することができます。
関数呼びだしに成功すると関数は１を返し、失敗すると０を返します。
ここでチャネルは、システムコールが呼び出される以前に発行された、イベントや接続要求がすべて削除された後で削除されます。
if (vi_delete(chanid)==0) {
perror("could not delete the virtual interrupt channel");
}
┌────────────────────────────────────────────────┐
│ｖｉ＿ｄｅｌｅｔｅ 仮想割込みチャネルの削除 │
│ │
│形式 │
│ │
│#include │
│ │
│int vi_delete(chanid); │
│ │
│引数 │
│ │
│VICHANID chanid; チャネルを指定するチャネルＩＤ │
└────────────────────────────────────────────────┘
仮想割込みチャネルは、イベントの送信機構を、ｓｉｇｎａｌ，ＡＳＴ，ＰＯＳＩＸイベントの３つから選択することができます。（ただし現在のインプリメントにＰＯＳＩＸイベントは組み込まれていません。）
これらの管理は、 ACTION 構造体にイベント通知機構を登録する事によって可能になりますが、ユーザは次のライブラリを使用して ACTION 構造体を初期化します。
以下の例は、ACTION 構造体に対してＡＳＴを使用するように初期化するもので、関数 func をＡＳＴ優先度１０として、受信割込み関数として登録します。
ここでフラグは ACT_UPDATE になっていますので、パラメータは、送信側より送られてきた値に置き換えれますが、送信パラメータが無い場合 initial_val が関数 func に引き渡されます。
また、仮想割込みチャネルでは、イベントは（イベント・カウンタによって）キューイングされますが、データはキューイングされずに上書きされるので、使用にあったって注意して下さい。
if ( !(pact = actast (NULL, func, initial_val, 10, ACT_UPDATE)) {
perror ("ACTION structure could not be allocated");
exit (1);
}
送信プロセス ACTION 構造体 受信プロセス
┌───────┐ ┌────────┐ ┌──────┐
│ │ │ イベントタイプ │ │ actast │
│ │ │ │ │ │
│ │ │ ハンドラ │ │ │
│ │ │ │ │ │
│ │ │低レベルハンドラ│ │ │
│ │ │ │ │ │
│ │ │ ＡＳＴ優先度 │ │ │
│ │ │data area │SENSE │ │
│ │ MAPSOURCE │┌─┐ ┌───┴──────┘ │
│ ├──────┼┼─┤→│ ＡＳＴキュー func(arg) │
│ │ │└─┘ └───┬──────┐ │
│ │ │event countor │ │ │
└───────┘ └────────┘ └──────┘
┌────────────────────────────────────────────────┐
│ａｃｔａｓｔ イベント通知機構にＡＳＴを使用するように、ACTION構造体を初期化する │
│ │
│形式 │
│ │
│#include │
│ │
│ACTION actast(act,func,arg,pri,flags) │
│ │
│引数 │
│ │
│ACTION *act ACTION; ACTION 構造体へのポインタ。 │
│ ポインタが NULL である場合、ACTION 構造体は actast()によって割り当てられる│
│void (*func)(); ＡＳＴの発生時に呼び出される関数 │
│int arg; func に引き渡される静的なパラメータ │
│int pri; ＡＳＴ優先度 │
│unsigned int flags; ０の場合にはＡＳＴ受信側に常に arg が引き渡され、それ以外の場合以下の通り │
│ ACT_UPDATE ＡＳＴによって引き渡されたパラメータが存在する場合、arg にこれを代入する │
│ ACT_FIRST ＦａｓｔＡＳＴである場合のみ arg に代入する │
│ ACT_NOQUEUE 前のＡＳＴが送信されていない場合ＡＳＴを破棄する │
└────────────────────────────────────────────────┘
┌────────────────────────────────────────────────┐
│ａｃｔｓｉｇ イベント通知機構にｓｉｇｎａｌを使用するように、ACTION構造体を初期化する │
│ │
│形式 │
│ │
│#include │
│ │
│ACTION actsig(act,sig) │
│ │
│引数 │
│ │
│ACTION *act; ACTION 構造体へのポインタ。 │
│ ポインタが NULL である場合、ACTION 構造体は actast()によって割り当てられる│
│int sig; ｓｉｇｎａｌ番号 │
│ シグナル・ハンドラは、デフォルト動作に初期化されるため、これ以外の動作を │
│ 必要とする場合、signal(2),sigset(2),sigvec(2) のいずれかによって初期化を │
│ しなければならない │
└────────────────────────────────────────────────┘
ここで注意してほしいのは、actast() や actsig() によって ACTION 構造体が初期化されないうちに、イベントの発信が起こってしまうと、異常動作を起こしてしまう事です。
これを防止するために、仮想割込みチャネルを生成する際に、VIP_DOMANT を指定してチャネルを停止状態にしておきます。
ACTION 構造体が、希望のイベント通知機構に初期化できたら、以下のようにしてチャネルと接続します。
if ((vi_sense (chan, pact)) == -1) {
perror ("could not establish a sense connection");
exit (1);
}
vi_sense() は、仮想割込みチャネルへの接続を試み、処理に成功すると、vi_sense() は選択されたチャネルに固有の接続ＩＤを、失敗すると−１を返します。
┌────────────────────────────────────────────────┐
│ｖｉ＿ｓｅｎｓｅ 仮想割込みチャネルの接続 │
│ │
│形式 │
│ │
│#include │
│ │
│VICONID vi_sense(chanid,pact,flags,arg) │
│ │
│引数 │
│ │
│VICHANID chanid; 仮想割込みチャネルを指定する │
│ │
│action pact; ACTION 構造体へのポインタ │
│ │
│unsigned int flags; 将来の拡張用に予約 │
│ │
│unsigned int arg; 将来の拡張用に予約 │
└────────────────────────────────────────────────┘
接続を解除するためには、以下の関数を使用します。
これは、仮想割込みチャネル chan から接続ＩＤ conid の削除を試みます。成功すると関数は０を返し、失敗すると−１を返します。
if ((vi_nonsense (chanid, conid)) == -1) {
perror ("could not establish a nonsense deconnection");
exit (1);
}
┌────────────────────────────────────────────────┐
│ｖｉ＿ｎｏｎｓｅｎｓｅ 仮想割込みチャネルの接続を解除する │
│ │
│形式 │
│ │
│#include │
│ │
│int vi_nonsense(chanid,conid) │
│ │
│引数 │
│ │
│VICHANID chanid; 仮想割込みチャネルを指定する │
│ │
│VICONID conid; 仮想割込みチャネルの接続ＩＤを指定する │
└────────────────────────────────────────────────┘
接続に成功したら、停止状態にある仮想割込みチャネルを活性化しなくてはいけません。このためには、vi_ctl() を使用します。
vi_ctrl() は、仮想割込みチャネルの制御を行う関数で、以下のように使用します。
if ((vi_ctl (VICTL_ACTIVATE, chanid, 1)) == -1) {
perror ("channel could not be activated");
exit (1);
}
┌────────────────────────────────────────────────┐
│ｖｉ＿ｃｔｌ 仮想割込みチャネルの制御 │
│ │
│形式 │
│ │
│#include │
│ │
│int vi_ctl(cmd,arg1,arg2) │
│ │
│int cmd,arg1,arg2; ただし arg1 は、(VICHANID)arg1; あるいは (unsigned char*)arg1; の場合もある │
│ │
│引数 │
│ │
│cmd 以下に示す機能を定義する │
│ │
│ VICTL_MAXCHAN arg1にゼロ／非ゼロを指定して、チャネルＩＤの制限を設定／入手する │
│ 設定は特権操作であり、arg2 に制限を設定し、関数は戻り値として以前の値を返す│
│ │
│ VICTL_MAXSECON arg1にゼロ／非ゼロを指定して、接続ＩＤの制限を設定／入手する │
│ 設定は特権操作であり、arg2 に制限を設定し、関数は戻り値として以前の値を返す│
│ │
│ VICTL_NCHAN 現在設定されているチャネルの数を得る │
│ │
│ VICTL_IMAPSZ arg1にゼロ／非ゼロを指定して、マッピングテーブルのサイズを設定／入手する │
│ 設定は特権操作であり、arg2 にサイズを設定する。テーブルは既に存在しない場合│
│ にのみ作成することができる │
│ │
│ VICTL_DELALLTASK これは特権操作でありユーザが作成したすべての仮想割込みチャネルを削除する│
│ │
│ VICTL_DELMYTASK 呼びだしプロセスによって所有されているすべての仮想割込みチャネルを削除する│
│ │
│ VICTL_DELALL これは特権操作であり、すべての非永続のユーザおよびシステム所有の仮想割込み │
│ チャネルを削除する │
│ │
│ VICTL_MAXSENSE (VICHANID)arg1で示されるチャネルＩＤの最大接続値を関数の戻り値として得る │
│ │
│ VICTL_NSENSE (VICHANID)arg1 で示されるチャネルＩＤの既接続数を関数の戻り値として得る │
│ │
│ VICTL_NMAP (VICHANID)arg1 で示されるチャネルＩＤに接続されているイベント発信元の │
│ 数を関数の戻り値として得る。 │
│ │
│ VICTL_PERM (VICHANID)arg1 で示されるチャネルＩＤの権利を関数の戻り値として得る │
│ │
│ VICTL_NAMETOID (unsigned char*)arg1 で示されるシステムでユニークなタグ名からチャネルＩＤを│
│ 関数の戻り値として得る。ここで arg1 は、vi_create() によって命名されたタグ │
│ 名へのポインタである。arg2 にゼロ／非ゼロを指定することにより、チャネルを活│
│ 性化／非活性化する。 │
│ │
│ VICTL_ACTIVATE (VICHANID)arg1 で示されるチャネルＩＤを、arg2 にゼロ／非ゼロを指定すること │
│ により、チャネルを活性化／非活性化する。 │
└────────────────────────────────────────────────┘
仮想割込みを発生させるためには、vi_map()によって希望のチャネルに対して接続をする必要があります。
このとき、接続に成功すれば、チャネルＩＤをマップインデックスに変換しますが、失敗した場合−１を返します。
if ((map_index = vi_map (chanid, 0)) == -1) {
perror ("could not perform input mapping");
exit (1);
}
┌────────────────────────────────────────────────┐
│ｖｉ＿ｍａｐ 仮想割込みチャネルに接続しチャネルＩＤをマップインデックスに変換する │
│ │
│形式 │
│ │
│#include │
│ │
│VIMAPID vi_map(chanid,flags) │
│ │
│引数 │
│ │
│VICHANID chanid; チャネルを指定するチャネルＩＤ │
│ │
│unsigned int flags; (flag & VIMAP_NOWAIT)を指定すると、チャネルが停止状態にある場合 │
│ ブロックせずエラーを返し、０ならブロックする。 │
└────────────────────────────────────────────────┘
また、マップされたチャネルは、次のようにして削除されます。
if (vi_unmap (mapid) == -1) {
perror ("could not unmapping input mapping");
exit (1);
}
┌────────────────────────────────────────────────┐
│ｖｉ＿ｕｎｍａｐ 仮想割込みチャネルのマップを削除し、マッピング・テーブルのエントリを解放する │
│ │
│形式 │
│ │
│#include │
│ │
│int vi_unmap(mapid) │
│ │
│引数 │
│ │
│VIMAPID mapid 仮想割込みチャネルを指定するマップＩＤ │
└────────────────────────────────────────────────┘
マップに成功したイベントの発信元プロセスは、以下の vi_mapsource を使用してイベントを発信します。このとき、イベント通知機構に関しては、発信元はいっさい関知しません。
if ((vi_mapsource (vir_map_indx, arg)) == -1) {
perror ("could not source the interrupt");
exit (1);
}
┌────────────────────────────────────────────────┐
│ｖｉ＿ｍａｐｓｏｕｒｃｅ 仮想割込みチャネルに対してパラメータと共に発信する │
│ │
│形式 │
│ │
│#include │
│ │
│int vi_mapsource(mapid,arg) │
│ │
│引数 │
│ │
│VIMAPID mapid; 仮想割込みチャネルを指定するマップＩＤ │
│ │
│int arg; イベントと共に送信される３２ビットパラメータ │
└────────────────────────────────────────────────┘
例題として、仮想割込みチャネルの簡単な用法を示します。この例では、２つのプロセス間で、抽象的な fred と言う名前をタグに使用し、仮想割込みチャンネルを生成し、コネクションを確立し、受信側で割込みで起動される関数として printf を定義しています。
受信プロセス 送信プロセス
│ │
"fred"という名前で ┌───┴───┐ ┌───┴───┐仮想割込みチャネルを"fred"と
停止状態の仮想割込み │ vi_create │ │ vi_ctl │言う名前を使って探す。３番目の
チャネルを生成する └───┬───┘ └───┬───┘引数が非ゼロ(1)であるため、
│ │ プロセスは仮想割込みチャネルが
イベント通知機構にＡＳＴ┌───┴───┐ │ 要求を受け入れるまでブロック
を使用し、printf()を割り│ actast │ │ される
込み関数として初期化する└───┬───┘ │
│ │
仮想割込みチャネルへ ┌───┴───┐ ┌───┐ │
接続する │ vi_sense │<<│ACTION│ │
└───┬───┘ │構造体│ │
仮想割込みチャネルを ┌───┴───┐ │ │ │ブロックが解除される
アクティブにする │ vi_ctl │→│ │→→→→→│
└───┬───┘ │ │ ┌───┴───┐仮想割込みチャネルに
┌───┴───┐ │ │<<│ vi_map │送信プロセスをマップする
スリープする │ sleep │ │ │ └───┬───┘
└───────┘ └───┘ │
割込みで起動 │
↓ │
┌───────┐ ┌───┴───┐
│ printf │←──────┤vi_map_source │イベント発信
└───────┘ └───────┘
プロセス間の仮想割込み機構
ユーザプロセスによる仮想割込み機構の使用例その１
受信プロセス側が仮想割込みチャネルを生成する場合
＜＜注意：この例では、データセグメントのコピーを持っているため、文字列を送信出来ますが、別プロセスの場合には、データセグメントをシェアードメモリに指定する必要があります。＞＞
#include
#include
extern int printf(); /* ｃｌｉｂの関数 printf() を割込み関数として使用する*/
extern ACTION * actast ();
void main () {
ACTION * pact;
VICHANID chan;
VIMAPID vir_map_indx;
int vir_chan_id, retv;
if (retv = fork ()) { /* Parent process follows */
if (retv == -1) {
perror ("error forking a child process");
exit (1); }
if ((vir_chan_id = vi_ctl (VICTL_NAMETOID, "fred", 1)) == -1) {
perror ("parent could not find the channel");
exit (1); }
if ((vir_map_indx = vi_map (vir_chan_id, 0)) == -1) {
perror ("parent could not perform input mapping");
exit (1); }
if ((vi_mapsource (vir_map_indx, "DYNAMIC\n")) == -1) {
perror ("parent could not source the interrupt");
exit (1); }
printf ("Parent process exiting normally\n");
}
else { /* Child process follows */
if ((chan = vi_create ("fred", 10, VIP_DORMANT)) == -1) {
perror ("child could not create the virtual interrupt channel");
exit (1); }
pact = actast (NULL, printf, "STATIC\n", 10, ACT_UPDATE);
if (!pact) {
perror ("ACTION structure could not be allocated by child");
exit (1); }
if ((vi_sense (chan, pact)) == -1) {
perror ("child could not establish a sense connection");
exit (1); }
if ((vi_ctl (VICTL_ACTIVATE, chan, 1)) == -1) {
perror ("channel could not be activated by child");
exit (1); }
if (sleep (5))
printf ("Child process exiting normally\n");
else
printf ("Child process exiting abnormally\n");
}
}
ユーザプロセスによる仮想割込み機構の使用例その２
送信プロセス側が仮想割込みチャネルを生成する場合
#include
#include "sys/param.h"
#include "sys/vi.h"
extern int printf ();
extern ACTION * actast ();
void main () {
ACTION * pact;
VICHANID chan;
VIMAPID vir_map_indx;
int vir_chan_id, retv,i;
if ((chan = vi_create ("fred", 10, VIP_DORMANT )) == -1) {
perror ("child could not create the virtual interrupt channel");
exit (1); }
if ((vir_map_indx = vi_map (chan, 0)) == -1) {
perror ("parent could not perform input mapping");
exit (1); }
for(i=0;i<100;i++){
if ((vi_mapsource (vir_map_indx, i)) == -1) {
perror ("parent could not source the interrupt");
exit (1); }
}
printf("end\n");
pause();
}
#include
#include "sys/param.h"
#include "sys/vi.h"
extern int printf ();
extern ACTION * actast ();
vis_server(arg)
int arg;
{
static int j=0;
printf("%d:%d\n",j++,arg);
}
void main () {
ACTION * pact;
VICHANID chan;
VIMAPID vir_map_indx;
int vir_chan_id, retv;
if ((chan = vi_ctl (VICTL_NAMETOID, "fred", 1)) == -1) {
perror ("parent could not find the channel");
exit (1); }
pact = actast (NULL, vis_server,-1 , 10, ACT_UPDATE);
if (!pact) {
perror ("ACTION structure could not be allocated by child");
exit (1); }
if ((vi_sense (chan, pact)) == -1) {
perror ("child could not establish a sense connection");
exit (1); }
if ((vi_ctl (VICTL_ACTIVATE, chan, 1)) == -1) {
perror ("channel could not be activated by child");
exit (1); }
pause();
}
これはタイマに同期させて周期的にプロセスを起動するもので、シミュレータやプロセス制御などのアプリケーションには、このスケジューリング・アルゴリズムをよく使用します。
シミュレータ等の場合１ｍ秒から５０ｍ秒ぐらいのフレームタイムで処理しますが、一定周期で起動できる最小時間がリアルタイムシステムの性能を決定するため、ユーザの要求はよりきびしいものになります。（１ｍ秒でどのぐらいの計算が可能でしょうか？）
← １０ｍｓ →← １０ｍｓ →← １０ｍｓ →
メジャサイクル ├─────────┼─────────┼─────────┤
マイナサイクル ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
プロセッサ１ Ａ├─┤ ├─┤ ├─┤ ├─┤ ├─┤ ├─┤
Ｂ ├─┤ ├─┤ ├─┤
Ｃ ├┤ ├┤ ├┤
プロセッサ２ Ｄ├──────┤ ├──────┤ ├──────┤
Ｅ ├─┤ ├─┤ ├─┤
定周期スケジューリング
プロセス制御やシミュレータ等のアプリケーションプログラムは、このメジャーサイクルの中で［入力］−［計算］−［出力］を終えなければなりません。
したがって、マイナーサイクルの周期が短くなれば、サイクルに占めるオーバーヘッドの割合が多くなり、オーバーヘッド時間の変動が問題になります。
これは、前述のすべてのパフォーマンスが影響してくるのに加えて、カーネルのチューニングが巧くいっていないと、（たとえば１００回に１回、起動が遅れるといったように）周期的に誤動作するといったことも起こるためどうしても長いマイナーサイクルになりがちです。
多くの場合この原因はシステム設計にあり、タイマのイベント処理部がブートプロセッサでしか実行できない、他のＵＮＩＸスケジューリング機構と競合を起こしている等が考えられます。（このため、リアルタイムＵＮＩＸでも周期スケジュール機能を提供しているベンダは少数です。）
ＲＴＵでの、周期スケジューリングは、クロックＡＳＴとして１ｍ秒でのサイクルを可能にしています。（ＲＴＵでは、このスケジューリングを使用するアプリケーション用にスケジュール機能をパッケージ化したＳＰＳ(Synthetic Period Scheduler)も別に用意しています。）
クロックＡＳＴは、mpadvise() によるシステムトラップ時間の変更で、マイナサイクルを決定し clkspec_t 構造体で次図のようにメジャサイクルとポーズ時間を決定します。このときメジャサイクル時間が０の時にはワンショットタイマの機能になります。
clkspec_t 構造体と時間の関係
mpadvise(MPA_CLK_SET,cpu_mask,CLK_HZ); マイナサイクル
↓↓↓↓↓↓↓↓↓↓↓
struct clkspec_t { │←メジャサイクル→│
struct timestruc { ├──┬──┬───┤
unsigned long ts_sec; │←→│←→│ │
long ts_nsec; │ │
} cs_period;
struct timestruc {
unsigned long ts_sec; プロセス実行時間
long ts_nsec;
} cs_pause; ポーズ時間
};
クロックＡＳＴは以下のように記述し、関数の呼びだしに成功すると関数はＡＳＴＩＤを返し、失敗すると−１を返します。このとき設定する processor_no は、先に説明した processor_mask とは違い、ブートプロセッサを１とする整数値です。
if ((ast_id=clkast(_asthandler, func,CLK_PRI,&pcs, processor_no))== -1){
perror("Failed to start clock");
exit(1);
}
ＲＴＵは、物理的なクロック割込みを検出するとそのイベント通知機構にＡＳＴを使用します。
したがって、クロックＡＳＴの使用方法はＡＳＴと同様です。
┌────────────────────────────────────────────────┐
│ｃｌｋａｓｔ タイマ割込みによる周期ＡＳＴを設定／解除する │
│ │
│形式 │
│ │
│#include │
│ │
│int astid = clkast(lo_handler,func,priority,pcs,processor_no) │
│int canclkast(astid) │
│ │
│引数 │
│ │
│int (*lo_handler)(); 通常は _asthandler()を指定する │
│int (*func)(); システムトラップ関数 │
│int priority; ０〜３２７６７の値を持つＡＳＴ優先度 │
│struct clkspec_t pcs; 前図参照 │
│int pid; 送信プロセスのプロセスＩＤ │
│int prosessor_no; １から８の整数で使用するプロセッサを指定する。 │
│ │
│int astid; キャンセルするＡＳＴＩＤ │
└────────────────────────────────────────────────┘
次の例は、１秒ごとのクロックＡＳＴを５回くりかえすとキャンセルして終了するものです。
クロックＡＳＴのプログラム例
#include
#include
#include
#include
#define NAST 5
#define ASTPRI 10 /* ＡＳＴ優先度を１０に定義 */
int tick (); /*ＡＳＴで起動される割込み関数の宣言 */
int _asthandler (); /* 低レベル割込みハンドラの宣言 */
struct clkspec_t interv;
int astcnt = 0;
main ()
{
int id;
/* 最初のＡＳＴは起動要求から２秒遅れて始まり、以降１秒毎に起動される */
interv.cs_period.ts_sec = 1;
interv.cs_period.ts_nsec = 0;
interv.cs_pause.ts_sec = 2;
interv.cs_pause.ts_nsec = 0;
/* クロックＡＳＴを１秒毎の繰返し起動させる。関数の戻り値としてＡＳＴＩＤを得る*/
if ((id = clkast (_asthandler, tick, ASTPRI, &interv, MASTER_ID)) == -1)
perror ("clkast"), exit (1);
while (astcnt < NAST)
{
astpause(-1, 5000);
/* astpause() は、タイムアウト前にＡＳＴの受信を検出すると０を返し停止状態から脱出する。*/
printf("main done astpause\n");
}
/* クロックＡＳＴをＡＳＴＩＤを使用してキャンセルする */
if (canclkast (id) < 0)
perror ("canclkast"), exit (2);
if (astcnt == NAST)
exit (0);
else
exit (3);
}
/* 割込み関数部 */
tick (overrun)
int overrun;
{
/* ＡＳＴの受信回数の出力 */
printf("In tick, AST count = %d\n", ++astcnt);
if (overrun) /* 関数の実行に遅れが生じた場合 overrun 変数にキューイングされている
イベントの数が代入される */
fprintf (stderr, "%d clock AST backlog\n", overrun);
}
ＦａｓｔＡＳＴは、ＦａｓｔクロックＡＳＴか、またはユーザが作成したカスタムデバイスのみで使用することができます。
クロックＡＳＴは特に信頼性を重視した設計になっていますが、高速にプロセス／スレッドを起動する事を重視して設計された、ＦａｓｔクロックＡＳＴも実装しています。
これは、ＡＳＴにおけるキューイング機能を削除し、さらに割込みによって保存されるレジスタセットに浮動小数点レジスタを含めるかどうかを選択できるようになっているのが特徴です。
またこのＦａｓｔＡＳＴは、リアルタイムスレーブ環境でのみ動作させることができ、カーネルのすべての割込みを mpadvise() を使用して停止しなければなりません。
したがってＲＴＵカーネルのＴＬＢ割込みも禁止しますが、これは他のプロセッサからのキャッシュ無効ブロードキャスト割込みを禁止しているため、安全なシステムコールの発行が保証できません。
このためプロセスまたはスレッドは、常に実行状態か、fastpause()のような、ビジーウェイト状態でなくてはならず、システムコールを発行してはいけません。
┌────────────────────────────────────────────────┐
│ｆｃｌｋａｓｔ タイマ割込みによる周期ＦａｓｔＡＳＴを設定／解除する │
│ │
│形式 │
│ │
│#include │
│ │
│int fclkast(lo_handler,func,priority,pcs,processor_no,opt) │
│int canclkast(astid) │
│ │
│引数 │
│ │
│int (*lo_handler)(); 通常は _fasthandler() を指定する │
│int (*func)(); システムトラップ関数 │
│int priority; ０〜３２７６７の値を持つＡＳＴ優先度 │
│struct clkspec_t pcs; 前図参照 │
│int pid; 送信プロセスのプロセスＩＤ │
│int prosessor_no; １から８の整数で使用するプロセッサを指定する。 │
│int opt; ｆｃｌｋａｓｔで浮動小数点レジスタを保存するオプション（FAST_FPUSED／0）│
│ │
│int astid; キャンセルするＡＳＴＩＤ │
└────────────────────────────────────────────────┘
┌────────────────────────────────────────────────┐
│ｆａｓｔｐａｕｓｅ ＦａｓｔＡＳＴ優先度設定とプロセスの停止 │
│ │
│形式 │
│ │
│ int fastpause(priority); │
│ │
│引数 │
│ │
│ int priority; ０〜３２７６７の値を持つＦａｓｔＡＳＴ優先度 │
└────────────────────────────────────────────────┘
ＦａｓｔクロックＡＳＴのプログラム
#include
#include
#include
#include
#include
#include
#define RT_CPU_MASK 4 /* セカンドプロセッサのプロセッサマスク値 */
#define RT_CPU_ID 2 /* セカンドプロセッサのプロセッサ番号 */
#define AST_PRI 10 /* ＦａｓｔＡＳＴ優先度 */
int handler(), _fasthandler(); /* ＦａｓｔＡＳＴの割込み関数と低レベルハンドラ */
volatile int done = 0; /* メイン処理の終了フラグ */
int counter = 0;
int overrun = 0;
main()
{
int pprio, ppolicy, fastid;
int freq = 100; /* 100 hertz (10 ms) */
struct clkspec_t pcs;
/* 固定優先度リアルタイムＦＩＦＯプライオリティ */
ppolicy = SCHED_FIFO;
pprio = MAX_RT_PRI;
if (schedset (0, &ppolicy, &pprio, NULL) < 0)
perror ("schedset"), exit (1);
if (mpadvise (MPA_RT_SET, RT_CPU_MASK) < 0) /* セカンドプロセッサはリアルタイム・スレーブ */
perror ("MPA_RT_SET"), exit (2);
/* 処理は１００Ｈｚで動作する（これは１２０Ｈｚのシステムトラップを停止することを意味する）
プログラムのデータセグメントとテキストセグメントをメモリに常駐させる */
if (mpadvise(MPA_CLK_SET, RT_CPU_MASK, freq) < 0)
perror ("MPA_CLK_SET"), exit (4);
if (rtplock(PROCLOCK) < 0)
perror ("rtplock"), exit (5);
/* プロセッサ間のＴＬＢ無効割込みを停止する。この指令を有効にするためには、
システムコンフィグレーション・パラメータの kmempool を十分確保しておかなくてはならない */
if (mpadvise(MPA_TBINV_OFF, RT_CPU_MASK) < 0)
perror ("MPA_TBINV_OFF"), exit (7);
/* １秒の遅れを伴って、１０ミリ秒ごとにシステムトラップを発生させる */
pcs.cs_period.ts_sec = 0;
pcs.cs_period.ts_nsec = 10000000;
pcs.cs_pause.ts_sec = 1;
pcs.cs_pause.ts_nsec = 0;
if ((fastid = fclkast(_fasthandler, handler, AST_PRI, &pcs,
RT_CPU_ID, 0)) < 0)
perror ("fclkast"), exit (6);
while (!done) /* Ｆａｓｔの発生をループで待つ */
{
if (fastpause(0) < 0)
perror("fastpause"), exit (8);
/* Do main loop processing (if any) */
}
printf("%d fast ASTs received with %d overruns\n", counter, overrun);
/* 処理が終了するとＦａｓｔクロックＡＳＴをキャンセルする */
canclkast(fastid);
}
handler(param, pc, pri)
int param, pc, pri;
{
/* 前の処理が終了する前に割込みがかかったかをチェックする */
if (param)
overrun += param;
else
counter++;
if (counter == 10)
done = 1;
}
すでに説明したように、一般的なＵＮＩＸのスケジューリングには高精度のタイマを必要としないため、プログラムで読み出すことのできる時間は、１／５０秒〜１／１２０秒の精度です。また特定のプロセスに対する複数のタイマ割込みの機能は提供していません。
リアルタイム処理では、システムで発生するイベントのタイムスタンプや事象の同期用に高精度で汎用性にとんだタイマサービスを提供しなくてはなりません。
コンカレント７０００シリーズでは、３つのプロセッサに対して専用の３２ビットタイマ（４０ｎｓ精度）を用意し、ユーザのアプリケーションプログラムから、定周期処理やワンショットタイマの機能を簡単に利用することができます。しかも、同一のプロセス上で複数のタイマ割込みのサービスを提供しています。
もっとも単純な利用方法としては、４８ビットリアルタイムカウンタを読み出すことが考えられますが、このクロックソースには、システムクロックを１６分周したものが供給されていますので、７０００シリーズ（２５ＭＨｚ）の場合６４０ｎ秒の解像度で下図のように任意のタイミングで読み出せます。
┌────────────────────────────────────┐
│ ４８ビットリアルタイムカウンタのアクセス例 │
│ 上位アドレス ←──────────────────→下位アドレス │
│ 63 56 48 40 32 24 16 8 0 │
│ ┌───┬───┬───┬───┬───┬───┬───┬───┐ │
│ └───┴───┴───┴───┴───┴───┴───┴───┘ │
│ ├───────────────┼───────────────┤ │
│ high low │
│ ├───────────────┤ │
│ middle │
│ │
│ low = *(unsigned long int *)0xfe001004;/* 単位：640 ns */ │
│ middle = *(unsigned long int *)0xfe001002;/* 単位：640 * 2１６ ns */ │
│ high = *(unsigned long int *)0xfe001000;/* 単位：640 * 2３２ ns */ │
└────────────────────────────────────┘
外部ゲート入力 ────────────────┐
┌────────┐ │
│システムクロック│ │
└┬───────┘ ↓
│ ┌──────────┐ ┌───────────────┐
└─┤４ビットプリスケーラ├┬→│４８ビットリアルタイムカウンタ│→ユーザプログラムで時刻を読出し
└──────────┘│ └───────────────┘
│ ┌───────────────┐
├→│ ３２ビット方形波ジェネレータ │→マルチシステム同期用外部クロック
│ └───────────────┘
│ ┌───────────────┐
└→│ ２０ビットシステムタイマ │────┬┬┐１２０Ｈｚの
└───────────────┘ │││システムトラップを出力
┌────────────────┐│││ ┌──────┐
→┌┴───────────────┐├┼┼┴┬┴─────┐│
→┌┴───────────────┐├┴┼┴┬┴─────┐├┘
→ │３２ビットアプリケーションタイマ├┴─┴─┤ プロセッサ ├┘
外部クロック入力１−３ └────────────────┘ └──────┘
↑ 各プロセッサの複数プロセス／スレッドを
外部ゲート入力１−３ ────────────┘ 任意のフレームタイムでスケジューリング
マルチシステム対応 高解像度内蔵タイマ・カウンタ
８０００シリーズでは、システム構成によって使用できる高解像度カウンタ／タイマが違います。
標準構成ではＩＯＰのものを、オプションとしてＭＶＩＯ−１のものがユーザで使用可能です。詳細は、ilt(4) をご覧ください。
以下のプログラムは、７０００／８０００シリーズ上でＣとｆ７７で共通に使用できる２つの関数 readtime_()と adjust_() の例です。
/* readtime.c */
#include
#include
#include
adjust_(start,finish,realtime)
unsigned long *start,*finish;
float *realtime;
{
#ifdef mc7000
*realtime = (float)( *finish - *start ) * 640.0 / 1000.0;
#endif
#ifdef mc8000
#ifdef MVIO
if (*start < *finish) { /* account for underflow */
*start += 0x1000000;
}
*realtime = (float)( *start - *finish );
#else
*realtime = (float)( *finish - *start ) * 62.50 / 1000.0;
#endif
#endif
}
readtime_(nowtime)
unsigned long *nowtime;
{
#ifdef mc7000
*nowtime = *(unsigned long int *)0xfe001004;
#endif
#ifdef mc8000
#ifdef MVIO
*nowtime = *(unsigned long int *)MAPPED_ILT_PAGE;
#else
*nowtime = *(unsigned long int *)MAPPED_COUNTER_PAGE;
#endif
#endif
}
コンパイル方法
cc -c readtime.c
Ｃ言語 ｆ７７
リンク方法
cc xxx.c readtime.o f77 xxx.f readtime.o
呼びだし方法
unsigned long int t0,t1; INTEGER*4 T0,T1;
float realtime,overhead; REAL*4 REALTIME,OVERHEAD;
/* オーバーヘッド時間の計測 */
readtime_(&t0); CALL READTIME(T0)
readtime_(&t1); CALL READTIME(T1)
adjust_(&t0,&t1,&overhead); CALL ADJUST(OVERHEAD)
readtime_(&t0); CALL READTIME(T0)
│ │
│ │
readtime_(&t1); CALL READTIME(T1)
adjust_(&t0,&t1,&realtime); CALL ADJUST(REALTIME)
printf("elapsed time was %g microseconds\n", PRINT *,"ELAPSED TIME WAS ",REALTIME-OVERHED,
realtime-overhead); " MICROSECONDS"
ＵＮＩＸのファイルは、ｉノードと呼ばれる固定長のブロックで管理されているため、容量の大きいファイルはディスクの物理的に連続していない領域に配置されてしまうことがあります。
このようなファイルに対してアクセスした場合、ディスクのシーク時間がファイルブロックの物理的な位置に影響されパフォーマンスが一定でなくなり、高速ファイルシステム設計の障害になります。そこでリアルタイムシステムのファイル構造は、物理的に連続した位置にファイルを割り当てる機能があります。
ＰＯＳＩＸでは、このような物理的に連続した領域にファイルを配置する機能をリアルタイム・ファイルと呼んでいます。
ＲＴＵでは、このファイルをコンティギャス・ファイルと呼んでいますが、これをシステムのスワップファイルにも利用しています。さらにマルチ・スワップファイルをサポートしているためタイムシェアリングのプロセスがスワップされる時に高速に動作します。また、フリーリストの管理を内部的にビットマップ管理をしているため、ｉノードのポインタをたどる必要がなく連続したブロックを高速に得る事ができ、すべてのプロセスがこの効果を得る事ができます。
┌────────────────┐ ┌────────────────┐
│ ブートブロック │ │ ブートブロック │
├────────────────┤ ├────────────────┤
│ ジオメトリブロック │ │ スーパブロック │
├────────────────┤ ├────────────────┤
│ ディスク・パーティション０−６ │ │ 再配置可能ｉノードリスト │
├────────────────┤ │ ────────────── │
│ 重複ジオメトリ・ブロック │ │ 不良ｉノードブロック │
├────────────────┤ │ ────────────── │
│ ブートストラップ領域 │ │ ルートｉノード │
├────────────────┤ │ ────────────── │
│ ダンプ領域 │ │ファイルとディレクトリ用ｉノード│
├────────────────┤ ├────────────────┤
│ 保守領域 │ │ ビットマップ │
├────────────────┤ ├────────────────┤
│ ハードウェア不良ブロックリスト │ │ データブロック │
└────────────────┘ ├────────────────┤
│ 重複スーパブロック │
└────────────────┘
ＲＴＵは、このようなファイルを図のようなジオメトリ・ブロックとＵＮＩＸファイルシステムを利用して物理ディスクの情報を管理しています。
ジオメトリ・ブロックは、ブートプログラムの場所、不良ブロックのリスト、およびファイル・システム・パーティションを含むディスク構造を維持し、物理ディスクのブロック１に存在し、ブロック０は、ブート・ブロック用に確保されています。
このジオメトリブロックにより、単一物理ディスクを最大７つのパーティションに分割管理して、論理ディスク装置として使用できます。
各ディスク・パーティションは、次に説明する６つの領域をもっています。
ブートブロック：
ブートストラップローダ用の領域、必ずしも内容は存在しない。
スーパブロック：
ここには以下のようなディスク・ファイル構造に関する全体情報が含まれている。
・作成できるファイル数
・ディスク上のブロック数
・現在利用できる空きブロックに数
・再配置可能なｉノードリストの位置
再配置可能なｉノードリスト：
ディフォルトでは、スーパブロックの次に、再配置可能なｉノードリストが存在する。
各ｉノードは、６４バイトの情報を持ち、不良ディスクブロックの存在しない位置に置かれている。
各ｉノードには、以下のようなファイルに関する情報が格納されていると共に、ファイルのデータ・ブロックの位置が格納されている。
・ファイルの種類（ファイル、ディレクトリ、スペシャルファイル）
・所有者
・ファイル・プロテクション
・ファイルのリンク数
・大きさ
・最終更新時刻
・最終アクセス時間
ビットマップ：
ｉノードリストに続き、ディスク・ブロック利用状況を管理しているビットマップが存在する。
ビットマップにより、ディスクブロックの割当てが高速になり、連続領域ファイルの割当てが可能になる。
データブロック：
上記ブロック以外のディスクブロックは、このデータブロックに割り当てられる。
重複スーパブロック：
このディスクブロックは、スーパブロックのコピーである。これは、元のスーパブロックが失われた場合に、ディスクユーティリィティによって使用される。
連続領域ファイルがとくに意味を持つのは、アクセスするファイルが大きい場合ですが、このときに２つの問題がクローズアップされます。
１つ目の問題は、ｒｅａｄ／ｗｒｉｔｅシステムコールは、カーネル内部にディスク転送バッファを持っているため結果としてデータ転送が２回起こってしまいます。（ユーザデータ領域→カーネルバッファ→ディスクブロック）
このような先読み後書き機構（read-ahead/write-behind）は、アクセスするデータが小さい場合はキャッシングの効果がありますが、データが大きい場合あきらかな無駄時間になり、入出力時間の予測を難しくします。
この問題を解決するためには、システムのバッファキャッシュの利用を禁止し、ユーザデータ領域とディスクブロックとの直接的なＤＭＡ転送を行う、直接ディスク転送を可能にしなくてはなりません。通常このような機能は、ディスク装置をｒａｗデバイスとして扱うことによって可能になりますが、オペレーティング・システムを離れて、ファイルを管理しなければならないため、非常に困難であるばかりでなく危険でもあります。
ＲＴＵでは、この連続領域ファイルに対する直接ディスク転送の機能は、ファイル作成時のモードの追加のみで利用できます。これにより、安全でかつ最小の変更で連続領域ファイルを使用することが可能になります。
┌────────────────────────────────────────────────┐
│ｏｐｅｎ ファイルをオープンする（連続領域ファイル関係のみ説明） │
│ │
│形式 │
│ │
│ #include │
│ │
│ int open(path,oflag[,mode[,size]]); │
│ │
│引数 │
│ │
│ char *path; 対称ファイルのパス名へのポインタ │
│ int oflag; 制御フラグ。以下のいずれかを単独、あるいは組み合わせて指定する │
│ O_CTG 連続ファイル領域のオープン │
│ O_TRCTG 連続ファイル領域のサイズの変更 │
│ O_DIRIO カーネルバッファをバイパスして直接ファイル入出力を可能にする │
│ O_SYNC 連続ファイル領域への同期ＷＲＩＴＥをセットする │
│ int mode; ファイル・モード │
│ size; O_CTG を指定した場合に確保するファイスサイズを指定する │
└────────────────────────────────────────────────┘
連続領域ファイルに対する直接ディスク転送プログラム
#include /* オープンフラグを定義 */
#include /* NBPG を使用−メモリ管理の１ページサイズ */
#define FSIZE 1048576 /* 連続領域ファイルの大きさ */
main ()
{
int fd, oflags, ret;
long offset = 0;
union {/* union を使用して、データ領域をページバウンダリに設定する */
char buf[NBPG];
short dummy[NBPG/2];
} mybuf;
/* １Ｍバイトの連続領域ファイルを生成する */
oflags = O_CREAT | O_CTG;
if ((fd = open ("newfile", oflags, 0600, FSIZE)) < 0)
perror ("Open O_CTG with FSIZE bytes"), exit (1);
close (fd); /* この後の直接入出力のためにクローズする */
/* 直接入出力のために連続領域ファイルを再オープンする */
oflags = O_DIRIO | O_WRONLY;
if ((fd = open ("newfile", oflags, 0600)) < 0)
perror ("open O_DIRIO"), exit (2);
while (offset < FSIZE){
if ((ret = write (fd, mybuf.buf, NBPG)) == -1)
perror("write"), exit(3);
if (ret != NBPG)
printf ("Write failed to transfer complete data\n");
offset += ret;
}
}
２つ目の問題は入出力操作のシステムコールが、ＵＮＩＸでは完了復帰型の設計になっているため呼びだしたプロセスが長時間封鎖されることです。複数のプロセスからのアクセスが競合した場合、最後にシステムコールを発行したプロセスは、すべての入出力動作が終了するまで封鎖され、貴重な実時間を無為に費やす結果になります。
入出力処理が非同期に実行できる場合、図のように実際の入出力操作時間は同じでも処理全体の時間は短くなります。ＲＴＵの場合、テープデバイスに対してこの機能が提供され、ディスクファイルの入出力に対しては、後で説明するスレッドを使用して実現しています。
�´↓� �きキΝЛ┃�
プロセス Ａ ───┐ ┌──────
│ │
ディスク出力 └──┬───┴───┐
│ │ ��
│ │ 書き込みデータの準備
プロセス Ｂ ───┘ └────── ��
�´↓� �きキΝЛ┃� �� ディスク書き込み
��
完了復帰型システムコール（同期入出力処理） �� ディスク操作とは関係無く実行可能
��
�´↓� �きキΝ鵝 ´Л┃� ��
プロセス Ａ ───┐┌───┐ ┌─── �� ディスク操作後しか実行できない
││ │ │ ��
ディスク出力 └┴─┬┬┴─┴┬──┐ �� 書き込み完了待ちシステムコール
││ │ │
プロセス Ｂ ───┘└───┘ └───
�´↓� �きキΝ鵝 ´Л┃�
非完了復帰型システムコール（非同期入出力処理）
┌────────────────────────────────────────────────┐
│ｉｏｃｔｌ 装置の制御（非同期入出力処理のみ説明） │
│ │
│形式 │
│ │
│ #include │
│ │
│ int ioctl(fildes,request,arg) 非同期入出力（ブロックデバイスはテープのみ） │
│ │
│引数 │
│ │
│ int fildes; ファイル記述子 │
│ request; FIOASYNC 非同期入出力の完了を待つ（ブロックデバイス） │
│ FIOAWRITE 非同期ＷＲＩＴＥを行なう（ブロックデバイス） │
│ FIOAREAD 非同期ＲＥＡＤを行なう（ブロックデバイス） │
│ FIONBIO 非ブロックデバイスの非同期ファイル入出力を制御する │
│ arg; struct _io { │
│ char *io_address; データバッファアドレス │
│ int io_count; 転送データ数 │
│ struct io_result { │
│ int io_acount; 実際に転送されたバイト数 │
│ int io_error; エラーコード │
│ } io_result; │
│ int (*lo_handler)(); 低レベルトラップハンドラへのポインタ │
│ int (*func)(); 転送終了時起動関数へのポインタ │
│ int io_param; パラメータ │
│ int io_priority; ＡＳＴプライオリティ │
│ }; │
└────────────────────────────────────────────────┘
以下に、ＡＳＴを使用した非同期入出力プログラムの例を説明します。
このプログラムは、ＲＴＵのテープドライバの非同期入出力機能を使用し、ストリーミング・テープ・デバイスへの入出力を実行します。このプログラムのコマンド行の引数は以下のようになります。
a.out [-f device] record_size
"device" のデフォルトは、/dev/rmt0 であり "record_size" は、各書込みのサイズをバイト単位で指定します。
このサイズは、１〜６３キロバイトの範囲でなければならず、たとえば以下のように指定します。
a.out -f /dev/rctp 1024
#include
#include
#include
#include
#include
#define MIN_SIZE 1024
#define MAX_SIZE (63 * 1024)
/* This is our I/O buffer structure. It ties the disk
* buffer in with the async I/O packet.
*/
typedef struct ioq {
struct _io m; /* The asyncio structure */
struct ioq *next; /* Forward pointer for chaining on lists */
char *buffer; /* Pointer to the actual I/O buffer */
} IOQ;
int _asthandler(); /* C-language low level handler */
char *malloc();
int completion(); /* AST completion routine */
int aborted(); /* ^C trap routine. */
IOQ *get_free_io_buffer(); /* Routine to return a free I/O buffer */
int tape; /* Channel to perform tape I/O on */
char asyncio; /* Holds total number of I/O buffers */
char iofree; /* Holds # of free buffers at any time */
IOQ *ioq_list = 0; /* The head of the free buffer list */
IOQ *pending_io = 0; /* Special list for better performance */
IOQ *last_pending = 0; /* Look in the write-track subr for info */
int record_size; /* Size in bytes of each I/O buffer */
int bytes_transferred; /* Total # bytes xferred for statistics */
char exit_flag = 0; /* Global flag gets set by ^C */
char *tape_name = "/dev/rmt0"; /* Default tape device to use */
/* ASYNC IO TAPE PROGRAM */
main(argc,argv)
int argc;
char **argv;
{
register short i;
register IOQ *io;
register int buffer_count;
int start_time,finish_time;
double xfer_rate;
/* First, decode the command line looking for record size and tape */
/* drive name. Record size must be specified, tape name is optional */
for(i = 1; i < argc; i++){ /* Read command line arguments */
if(strcmp(argv[i],"-f") == 0){ /* Tape drive flag */
tape_name = argv[++i]; /* Point to tape drive name */
continue; /* Go to next argument */
}
record_size = decode_size(argv[i]); /* Get size of records */
}
/* Insure that record size is in 1\-63 Kbyte range */
if(record_size < MIN_SIZE || record_size > MAX_SIZE){
printf("Illegal record size %d\n",record_size);
exit(2);
}
if((tape = open(tape_name, O_WRONLY)) < 0){ /* Open raw tape device */
printf("Open failed for tape drive %s\n", tape_name);
exit(1);
}
init_tapeio(tape); /* Call routine to init async I/O */
buffer_count = (40 * 1024 * 1024) / record_size; /* Calc buffers required */
bytes_transferred = 0; /* Init number of bytes transferred so far */
sigset( SIGINT, aborted ); /* Set up the interrupt trap routine */
time(&start_time); /* Record starting time of day */
while(1){ /* Loop until all records written */
io = get_free_io_buffer(); /* Get an I/O buffer to use */
write_track_to_tape(io); /* Dump buffer to tape */
if(--buffer_count <= 0) {
break; /* If done, quit loop */
}
if(exit_flag) {
break; /* If interrupt, quit loop */
}
}
flush_pending_io(); /* Flush any I/O still waiting */
close(tape); /* Close the tape drive */
time(&finish_time);
finish_time -= start_time; /* Get delta seconds it took to do */
xfer_rate = (bytes_transferred / finish_time) / 1024.0;
printf("%d seconds = %7.3f Kbytes/Second\n",finish_time, xfer_rate);
}
/* INIT_TAPEIO - Routine to set up the async I/O data structures */
init_tapeio(channel)
int channel;
{
register int i;
register IOQ *io;
/* Next several lines figure out how many "record_size" I/O buffers to
* use. The driver has a limited number of buffers, and the FIOASYNC
* ioctl returns this to us. We then figure out, with that as a maximum,
* how many buffers we want to have so that we end up with about 200K of
* buffer space.
*/
/* Get maximum number of outstanding async i/o requests the device
* can support. This becomes our maximum number of buffers.
*/
asyncio = ioctl(channel,FIOASYNC,0);
if(asyncio == 0) {
printf("Driver does not support async I/O\n");
exit(1);
}
/* Try for 200K worth of buffer space, if possible */
if((asyncio * record_size) > (200 * 1024)){
asyncio = (200 * 1024) / record_size;
}
printf("Using %d buffers\n",asyncio);
/* Initialize # of free I/O buffers and allocate packet identifiers */
iofree = asyncio;
ioq_list = (IOQ *)malloc(asyncio * sizeof(IOQ));
io = ioq_list; /* Get address of first packet */
for(i = 0; i < asyncio; i++){ /* Initialize packets */
io->m.io_count = record_size; /* Byte count for transfer */
io->m.io_handler = _asthandler; /* Address of std AST handler */
io->m.io_param1 = completion; /* Address of our AST handler */
io->m.io_param2 = (int)io; /* Address of this packet */
io->m.io_priority = 10; /* AST priority */
io->next = &io[1]; /* Point to the next packet */
io->buffer = malloc(record_size); /* Allocate a disk buffer */
/* Lock buffers in core so we don't lose time page-faulting them in */
if(plockin(io->buffer,record_size) < 0){
printf("Cannot lock down I/O buffer\n");
exit(2);
}
io++; /* Point this to the next packet */
}
io--; /* Back this up to the last packet */
io->next = 0; /* No next pointer on the last packet */
}
/* GET_FREE_IO_BUFFER - This routine returns an I/O buffer to use */
IOQ *
get_free_io_buffer()
{
register IOQ *io;
setpri(10); /* Block asts while we manipulate list */
while((io = ioq_list) == 0){ /* Loop while there are no free buffers */
astpause(0,1000); /* Sleep at low priority until an AST */
setpri(10); /* Pop priority back up */
}
ioq_list = io->next; /* Unchain first block from the list */
iofree--; /* Decrement number of free buffers */
setpri(0); /* Lower priority */
return(io);
}
/* FLUSH_PENDING_IO - Routine to insure all remaining buffers get written */
flush_pending_io()
{
register IOQ *io;
setpri(10); /* Avoid conflict with ASTs */
io = pending_io; /* Get beginning of pending list */
while(io){ /* Dump entire list to the driver */
io->m.io_address = io->buffer; /* Load address of the buffer */
ioctl(tape,FIOAWRITE, &io->m); /* Send it to the driver */
io = io->next; /* Chain to next packet on list */
}
pending_io = last_pending = 0; /* Zero the pending list */
setpri(0); /* Drop priority back down */
return;
}
/* WRITE_TRACK_TO_TAPE - Routine to send an I/O buffer to the tape driver */
write_track_to_tape(io)
register IOQ *io; /* The I/O buffer to be sent */
{
static int stream_count = 0; /* Remembers xfers and re-positions */
setpri(10); /* Block ASTs */
/* If no buffers are outstanding, then a re-position has occurred.
Use the pending list instead of immediately queuing packet
*/
if(iofree == (asyncio-1)){ /* We have 1 buffer, io */
io->next = 0;
if(pending_io) last_pending->next = io;
else pending_io = io; /* Insert onto the end of the pend list */
last_pending = io; /* since the order that the buffers get */
setpri(0); /* written to tape is important. */
printf("Begin New pending list (%d records written)\n",stream_count);
stream_count = 1; /* Reset count of buffers per reposition*/
return; /* Ok, just leave it on the pend list */
}
/* We get here if all of the buffers are not free. However, we may still
* have just had a re-position. If there is anything on the pending list, this
* is the case. Therefore, if anything is on the pending list, we put this
* buffer on it also.
*/
stream_count += 1; /* Count number of records / reposition */
if(pending_io){ /* If pending IO already */
io->next = 0; /* just chain onto the end of the list */
last_pending->next = io; /* until we have a lot of disk built up */
last_pending = io; /* Indicate we are now final packet */
}
/* Else, if there is nothing on the pending list, the tape drive is still
* moving. In this case, we want to get this packet to the driver quickly
* so that it helps to keep the drive moving.
*/
else {
io->m.io_address = io->buffer; /* io_address to track buff */
ioctl(tape, FIOAWRITE, &io->m); /* Give the packet to the driver */
setpri(0); /* Unblock ASTs now */
return; /* Go back to get more disk blocks */
}
/* Here we know there is I/O on the pending list. Check whether
* we need to send the list because all I/O buffers are on the list.
*/
if(iofree == 0) { /* If all I/O Packets are exhausted... */
io = pending_io; /* Get the beginning of the pending list */
while(io) { /* Dump entire list to the driver */
io->m.io_address = io->buffer; /* Load address of buffer */
ioctl(tape,FIOAWRITE, &io->m); /* Send it to the driver */
io = io->next; /* Chain to next packet on list */
}
pending_io = last_pending = 0; /* Zero the pending list */
}
setpri(0); /* Drop priority back down */
return;
}
/* COMPLETION - AST routine entered when a buffer has finished */
completion(io, pc, pri)
register IOQ *io; /* Address of finished I/O buffer */
int pc;
int pri;
{
register struct _io *m; /* Pointer to _io structure */
m = &io->m; /* Set to address of the _io structure */
/* if no errors, increment count of number of bytes transferred */
if(m->io_result.io_acount == m->io_count) {
bytes_transferred += m->io_count;
} else { /* Else, there WAS an error */
if(m->io_result.io_acount == 0) { /* If end of tape error */
exit_flag = 1; /* Then quit now */
} else {
printf("Error %d\n",m->io_result.io_error);
}
}
io->next = ioq_list; /* Chain beginning of list to our packet */
ioq_list = io; /* Put our packet back at the beginning */
iofree++; /* Increment count of free buffers */
}
/* ABORTED - Here on a CTRL-C (interrupt) from the user */
aborted()
{
exit_flag = 1; /* Tell the main loop to stop */
}
/* DECODE_SIZE - Here to perform an atoi function
*
* Function:
*
* This routine acts like atoi, except that it recognize
* a 'K' to mean times 1024. Called by code in "main" that
* reads arguments.
*/
decode_size(string)
register char *string;
{
register int value;
register int c;
value = 0; /* Initialize value */
while(c = *string++){ /* Run until we hit a null */
if(c >= '0' && c <= '9'){ /* If a numeric digit */
value *= 10; /* multiply value up by the base */
value += c - '0'; /* Take out ASCII offset, add digit */
continue; /* Go process next character */
}
if(c == 'K' || c == 'k') { /* If he is specifying in K-bytes */
value *= 1024; /* Then multiply by a K */
continue;
}
printf("?Illegal character [%c] in numeric string\n",c);
}
return(value);
}
ＵＮＩＸのプロセスは主記憶を大量に消費します。ＵＮＩＸが出回り始めた当初は少ない主記憶と遅い二次記憶装置であったために、少し大きなプログラムを実行するとシステムがスワップイン／スワップアウトを繰り返し、その実行速度の遅さに耐えられないユーザが多くいました。
メモリとディスクが速く安く大容量になった現在、こんなことが問題になるケース少ないでしょうが、リアルタイムＵＮＩＸでは、ちょうど同じような現象がおこります。
この場合耐えられないのは人間ではなくリアルタイム処理です。リアルタイムプロセスがページング／スワッピングされるとディスク装置からスワップインされる分だけ応答時間が遅くなります。したがってリアルタイムプロセスは、主記憶に常駐させる機構が必要です。
ＳＶＲ４では、このメモリロック機構がｐｌｏｃｋシステムコールとして実装されています。
このシステムコールでは、プロセスのテキストセグメント、データセグメントのどちらかあるいはその両方をロックします。けれどもこの方法だけでは、非常に大きいデータ構造をあつかう場合たいへん不利です。もしデータセグメントが大きすぎ、主記憶に常駐できない場合、ロック機構自体が使用できなくなるからです。したがって、現実的には、ＲＴＵのようにＳＶＲ４のロック機構に加えて、データをページ単位でロックする機構を加える必要があります。
┌──────────────────┬───────────────────────────────┐
│pinfo(address,size,ptevec,cmapvec) │カレントプロセスの仮想空間のページテーブルとコアマップを得る │
│plock(op) │テキスト、データセグメントのどちらかあるいは両方をロックする │
│rtplock(op) │リアルタイムプロセス専用 plock システムコール │
│plockin(address,size) │メモリをページ単位でロックする │
│punlock(address,size) │メモリをページ単位でアンロックする │
│brk(end_datasegment) │プロセスのデータセグメントを拡張する │
│sbrk(increases_size) │プロセスのデータセグメントのアドレスを得る │
│mmap(addr,len,prot,share,fd,off) │プロセスの仮想空間にメモリかシステムオブジェクトをマップする │
│munmap(addr,len) │プロセスの仮想空間のオブジェクトをアンマップする │
└──────────────────┴───────────────────────────────┘
メモリ管理を考えるときに、もう一つ問題になるのは、キャッシュの一致性とバスアクセス競合の問題です。
現在ＵＮＩＸで使用される、ほとんどの代表的なマイクロプロセッサにはキャッシュが内蔵されていますし、マルチプロセッサ構成のワークステーションも一般的になってきました。
キャッシュの不一致問題は、このようなマルチプロセッサ構成はもとよりシングルプロセッサでもＤＭＡコントローラ等のマルチマスタ構成だとこのキャッシュの不一致が起こります。
（キャッシュのプロトコルは各種ありますので、ここではモトローラ６８０Ｘ０で採用されているプロトコルについて解説します。）
キャシュの不一致問題とは図の初期状態のように正しい状態で、同一データを共有しているときに、いずれかのプロセッサが、データを更新すると図の中間状態のように、キャッシュの内容が不一致状態になることをいいます。
図Ａで示したライトスループロトコルの場合ただちに同様のデータをメモリに書き出し、他のキャッシュのタグをダーティ状態にします。その後ダーティ状態のキャッシュを読みだそうとするとメモリから正しいデータを読み出しキャッシュを更新します。
このプロトコルの欠点は、キャッシュ更新時に必ずデータをメモリに書き出すためマルチプロセッサ（マスタ）構成時にバストラフィックを増加させる原因になることです。
またＤＭＡなどでメモリの内容を直接書き換えた場合、キャッシュは更新されませんので、特別な操作でキャッシュをフラッシュし正常な動作を保証しなくてはなりません。（デバイスドライバでキャッシュをフラッシュします）
これらの欠点を解決したのがコピーバック・プロトコルです。（スヌーピング・プロトコルとも呼ぶます）
このプロトコルを実現するためには、バスを監視する専用モジュールが存在することを前提にしています。このようなコントローラをスヌーピング・コントローラと呼び、ＭＣ６０８０４０プロセッサはこれを内蔵しています。
基本的な考え方は、メモリアクセスを最小限におさえるため、キャッシュの更新時にキャッシュのみ書換えをおこない、メモリの内容はダーティ状態のままにしておきます。メモリの内容が更新されるのは、キャッシュの内容がキャッシュから排出される際にメモリに書き戻されます。
他のプロセッサのキャッシュの更新は、スヌーピングコントローラによりキャシュ間で高速に行われ、他のバスマスタ（ＤＭＡ等）のアクセスに対しても同様の動作になります。
┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐
主記憶│○ A=0 │ │× A=0 │ │○ A=1 │ │○ A=1 │
└──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘
─┬──┴──┬─ ─┬──┴──┬─ ─┬──┴──┬─ ─┬──┴──┬─
┌─┴─┐ ┌─┴─┐┌─┴─┐ ┌─┴─┐ ┌─┴─┐↑┌─┴─┐┌─┴─┐↓┌─┴─┐
キャッシュ│○ A=0│ │○ A=0││○ A=1│ │× A=0│ │○ A=1│↑│× A=0││○ A=1│↓│○ A=1│
├───┤ ├───┤├───┤↑├───┤ ├───┤ ├───┤├───┤ ├───┤
プロセッサ│ │ │ ││ A++ │↑│ │ │ │ │ ││ │ │ B=A │
DMA └───┘ └───┘└───┘ └───┘ └───┘ └───┘└───┘ └───┘
［データ更新］ ［メモリに書き出す］ ［キャッシュを更新］
初期状態 中間状態 最終状態
図Ａ ライトスルー方式のキャッシュ制御
┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐
主記憶│○ A=0 │ │× A=0 │ │× A=0 │ │○ A=1 │
└──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘
─┬──┴──┬─ ─┬──┴──┬─ ─┬──┴──┬─ ─┬──┴──┬─
┌─┴─┐ ┌─┴─┐┌─┴─┐ ┌─┴─┐ ┌─┴─┐ ┌─┴─┐┌─┴─┐↑┌─┴─┐
キャッシュ│○ A=0│ │○ A=0││○ A=1│ │× A=0│ │○ A=1│→│○ A=1││ │↑│ │
├───┤ ├───┤├───┤↑├───┤ ├───┤ ├───┤├───┤ ├───┤
プロセッサ│ │ │ ││ A++ │↑│ │ │ │ │ B=A ││ │ │ │
DMA └───┘ └───┘└───┘ └───┘ └───┘ └───┘└───┘ └───┘
［データ更新］ ［キャッシュ間コピー］ ［メモリを更新］
初期状態 中間状態１ 中間状態２ 最終状態
図Ｂ コピーバック方式のキャッシュ制御
ＲＴＵでは、５０００／６０００シリーズには、ライトスルー方式のキャッシュを、７０００／８０００シリーズには、コピーバック方式のキャッシュを使用しています。
これらのキャッシュ制御は、chpage(2) システムコールを使用して、指定されたアドレスで開始するページの属性を変更でき、キャッシングを使用可能にしたり、禁止したりできます。
┌────────────────────────────────────────────────┐
│ｃｈｐａｇｅ ページ保護属性の変更 │
│ │
│形式 │
│ │
│ #include │
│ │
│ int chpage(address,size,attr,sel) │
│ │
│引数 │
│ │
│ unsigned int address; ページ保護属性の変更を開始する仮想空間のアドレス │
│ int size; バイト単位のデータサイズ │
│ unsigned int attr; 設定する保護属性ビット │
│ unsigned int sel; 選択する保護属性ビット │
└────────────────────────────────────────────────┘
address および size は、呼びだしの影響を受けるアドレス空間を表します。attr と sel は、変更されるページ属性を設定および選択します。どの部分も指定されたアドレス範囲内で生成されるすべてのページが呼出しの影響を受けます。
chpage は、保護ビットまたは、キャッシュビットが設定されたページの整数を関数の戻り値として返します。
attr と sel の値は、以下の組合せの論理和によって割り当てられます。
CP_READ
読込みアクセス権を設定または選択する。
CP_WRITE
書込みアクセス権を設定または選択する。
CP_EXEC
実行アクセス権を設定または選択する。すべてのＲＴＵシステムで、CP_EXEC は、CP_READ と同時に指定されなくてはならない
CP_CHACHE
すべてのキャシュにアクセスされるアクセス権を設定または選択する。
CP_FPP
浮動小数点プロセッサ空間のアクセスを設定または選択する。この値は５０００／６０００システムのみ存在する。
CP_DI
”キャシュでページは無効化されない”を設定または選択する。これにより代入されたハードウェア設定値を重ね書きできる。８０００シリーズには、適用されない。
以上の値に加えて７０００シリーズでは以下の設定を使用できます。これらの値と上記の値の論理和によって７０００シリーズのキャッシュモードを選択できます。これらの２つ以上の値の論理和演算を行うと、ＲＴＵがシステム性能が最大となる値を選択します。以下の順番は、性能が低い順番に並べてあります。
CP_SERIAL
直列キャッシュ・モードを設定または選択する
CP_NSERIAL
非直列キャッシュ・モードを設定または選択する
CP_WRTHRU
ライトスルー・キャッシュ・モードを設定または選択する
CP_CPYBCK
コピーバック・キャッシュ・モードを選択または設定する
注意：属性を変更するプロセスは、少なくともページの読込み許可（CP_READ）を持っていなくてはなりません。プロセスがひとたび権利を CP_READ 以下に設定すると、二度と権利を回復できません。
プロセスは、以下の呼びだしを行ってプロセスの最初のテキスト・ページを読込み専用にすることができます。
numpages = chpage(0,NBPG,CP_READ,CP_READ|CP_WRITE|CP_EXEC[|CP_CPYBCK]);
ページを読込み専用にして、キャッシングをオフにするには、以下のように記述します。
numpages = chpage(0,NBPG,CP_READ,CP_READ|CP_WRITE|CP_EXEC|CP_CHACHE[|CP_CPYBCK]);
逆に、読込み、書込み、実行許可を復元し、キャッシングをオンに戻すには以下のように記述します。
numpages = chpage(0,NBPG,CP_READ|CP_WRITE|CP_EXEC|CP_CHACHE[|CP_CPYBCK]
,CP_READ|CP_WRITE|CP_EXEC|CP_CHACHE[|CP_CPYBCK]);
ＲＴＵにおいて共有メモリは、以下の３つの方法で実現できます。いずれも同様の機能を提供しますが、それぞれ特徴があります。
１）ＳＶＲ４対応のシェアードメモリ
shmem(2)コールの形での標準共有メモリ。この方法によりプロセスは、データを共有データとして宣言できその利点は移植性にあります。
２）バークレイ規格のメモリマッピング
ＳＶＲ４シェアードメモリと同様に、プロセスはデータを共有できますが、移植性が低いためＶＭＥバス上の外部デバイス用に使用する場合に大きな効果を生みます。
３）ＲＴＵの拡張であるスレッド
多重スレッドは、あまりシステム資源を必要とせず、プロセスより制御が容易なもっとも便利で効果的なデータの共有方法です。しかし移植性は良くありません。
複数のプロセスが、共有メモリに対してアクセスする場合、通常なんらかの機構によって排他制御を行わなくてはなりません。ＲＴＵでは、この排他制御機構に２つの方法を用意しています。
・ＳＶＲ４標準のセマフォシステムコールによる方法
・プロセッサに固有のテスト＆セット（ｔａｓ）命令を使用する方法
どちらの方法も、上記の共有メモリに対する排他制御機構に対して有効ですが、スレッドに対しては専用のライブラリが別に用意されています。
また、ｔａｓ命令を使用するには、ロックを設定および解除するために必要なアセンブリ言語を記述し、そのコードを高水準言語プログラムにリンクする必要があります。以下に５０００／６０００／７０００シリーズにおけるテスト＆セット命令のプログラムを示します。なお８０００シリーズでは、標準Ｃライブラリに tset(3),tclr(3) ルーチンがすでに提供されています。（７０００シリーズ以外のシステムでは、外部メモリに対してのテスト＆セット命令は使用できません。）
ライトスルーキャッシュを使用している５０００／６０００シリーズ上でテスト＆セット命令を適切に実行するためには、テストするバイトのあるメモリ・ページのキャッシュを使用不可にする必要があります。
このため、ＲＴＵではプログラムの互換性を重視して、共有メモリに対してキャッシュは使用不可になっていますが、chpage(2) を使用してユーザプログラムから使用可能にすることができるようになっています。
tset() & tclr() ルーチン ┌────────────────────┐
│tset() & tclr() 使用例 │
.text │ │
.globl _tset │ int lock; │
_tset: │ │
movl sp@(4),a0 │ lock = 0;/* 初期化 */ │
1$: │ │ │
tas a0$@ │ │ │
jpl 2$ │ tset(&lock); │
moveq #1,d0 │ ＜クリティカル・セクション＞ │
divs d0,d0 │ tclr(&lock); │
divs d0,d0 │ │
jra 1$ └────────────────────┘
2$ rts
.globl _tclr
_tclr:
movl sp@(4),a0
movw a0@,d0
movq #0,d1
1$ cas.w d0,d1,a0@
jne 1$
rts
ＳＶＲ４共有メモリ（shared memory:shm）は、複数のプロセス間で共通に参照されるメモリ空間で、各プロセスに同一メモリ空間がマップされます）。
ＲＴＵでの実装上の特徴はシンメトリックになっているためどのプロセッサからでも同時に実行できるためパフォーマンスが向上しています。
プロセスＡ プロセスＢ
┌─────────┐ ┌─────────┐
│ │ │ │
│テキストセグメント│ │テキストセグメント│
│ │ │ │
├─────────┤ ├─────────┤
│ │ │ │
│ データセグメント │ │ データセグメント │
│ │ │ │
│ │ │ │
├─────────┤ ├─────────┤
│ 共有メモリ │ マッピング │ 共有メモリ │
│ セグメント │ │ セグメント │
└─────────┘ └─────────┘
実記憶
┌─────────┐
│ │
│ │
└─────────┘
共有メモリを操作するシステムコールは、次の４つです。
shmget() 共有メモリを得る
shmat() 共有メモリを割り付ける
shmdt() 共有メモリの割付けを解除する
shmctl() 共有メモリを操作する
また、使用における制限は、カーネルの以下のシステムコンフィグレーションによって決定されます。
shmmax バイト単位での共有メモリセグメントの最大サイズ
shmmin バイト単位での共有メモリセグメントの最小サイズ
shmmni システムが提供する共有メモリセグメントの最大数
shmall 同時に使用できる共有メモリセグメントの最大数
shmseg 任意のプロセスに１度に割り付けることができる共有メモリの最大数
shmbrk ＲＴＵ用に予約
ＳＶＲ４共有メモリとセマフォの例題
この例では、１つの制御プロセスから、論理的に同一のサーバとして機能する３つの子プロセスが生成されています。制御プロセスは、ひとつのキューによってサーバにコマンドを送り、サーバは２つ目のキューによって制御プロセスに応答を返します。すべてのプロセスが初期化されると、制御プロセスから各サーバに STAT メッセージが送信されます。このメッセージを受信すると、サーバは処理をシミュレートするために停止し（実際の処理は存在しません）、制御プロセスに応答を返します。最後に制御プロセスからサーバに、 END メッセージが送信されサーバプロセスは終了します。このときセマフォおよび共有メモリはプログラム終了前のクリーンアップによって解放されます。
４つのプロセスは、すべて１つの共有メモリセグメントを共有し、このセグメントには要求キューと応答キューの２つが存在します。要求キューは、制御プロセスからサーバへのメッセージを保持し、応答キューはサーバから制御プロセスへの応答を保持します。各キューには２つのセマフォからなる、セマフォセットがあります。セマフォは、常に１つのサーバのみが要求キューを読みとるか、または応答キューに書き込むように排他制御します。待機セマフォは、キューにメッセージが存在する場合に読取りプロセスを制御します。
制御プロセスとサーバが応答または要求の検索中の場合には、待機セマフォで待機状態になっています。制御プロセスは、サーバへの要求を生成すると、要求キューの待機セマフォにこれを通知し、サーバは制御システムの応答がある場合に応答キューのセマフォに通知します。
#include
#include
#include
#include
#include
#include
struct queue {
short load; /* load pointer as index into qentry */
short unload; /* unload pointer, load == unload means full */
short size; /* number of qentrys in queue */
int semid; /* semaphore identifier */
struct qentry *qstart; /* start of queue */
};
struct qentry {
short command; /* what to do */
short argument; /* what to do it to */
short response; /* how it did */
short server; /* which one did it */
};
struct queue *qrequest, *qresponse; /* in shared memory */
/* queueing macros */
#define qincr(q,ptr) q->ptr = (q->ptr + 1) % q->size
#define qptr(q,ptr) (q->qstart + q->ptr)
/* commands for queue */
#define START 1
#define COMPL 2
#define END 3
/* number of servers */
#define NCHILD 3
#define RSPQSIZE (NCHILD + 1)
#define REQSIZE (NCHILD + 1)
/* protections on ipc stuff */
#define RUSER 0400
#define WUSER 0200
/* which semaphore in set is which */
#define WAITING 0
#define MUTEX 1
/* semaphore macros: P is wait, V is signal */
struct sembuf psemop[] = {
{ WAITING, -1, 0 },
{ MUTEX, -1, 0 }
};
struct sembuf vsemop[] = {
{ WAITING, 1, 0 },
{ MUTEX, 1, 0 }
};
#define P(q,sem) semop(q->semid, &psemop[sem], 1)
#define V(q,sem) semop(q->semid, &vsemop[sem], 1)
char *shmaddr, *sbrk(); /* address shared memory attached at */
int shmid, shmsize;
main ()
{
char *shmat();
int i;
int pid;
int status;
int nchild = 0;
struct qentry *qp;
int nbytes;
/* round up end of data segment to nearest page boundary
* and allocate pages
*/
shmsize = 2 * sizeof (struct queue) +
((RSPQSIZE + REQSIZE) * sizeof (struct qentry));
shmaddr = (char *) (((unsigned) sbrk(0) + 0xfff) & ~0xfff);
/* get the amount of memory we need (rounded up to page boundary) */
nbytes = btoc(shmsize) * NBPG;
if (brk (shmaddr + nbytes)) {
perror ("could not break");
exit (1);
}
/* get a shared memory segment rounded up to nearest page size */
if ((shmid = shmget (IPC_PRIVATE, nbytes, RUSER | WUSER)) == -1) {
perror("Could not grab shared memory segment");
exit (1);
}
/* attach to nearest page boundary (round down) */
shmaddr = sbrk(0) - nbytes;
if ((shmaddr = shmat (shmid, shmaddr, SHM_RND)) == (char *) -1) {
perror ("Could not attach shared memory segment");
goto freeseg;
}
/* Initialize queue data structures, allocate semaphore sets */
qrequest = (struct queue *) shmaddr;
qresponse = qrequest + 1;
qrequest->size = REQSIZE;
qrequest->qstart = (struct qentry *) (qresponse + 1);
qrequest->load = qrequest->unload = 0;
qrequest->semid = -1;
qresponse->size = RSPQSIZE;
qresponse->qstart = qrequest->qstart + REQSIZE;
qresponse->load = qrequest->unload = 0;
qresponse->semid = -1;
/* Allocate the semaphores for each queue */
if ((qrequest->semid = semget (IPC_PRIVATE, 2, RUSER | WUSER)) == -1) {
perror ("Could not grab request semaphore");
goto freesem;
}
if ((qresponse->semid = semget(IPC_PRIVATE, 2, RUSER | WUSER)) == -1) {
perror ("Could not allocate response semaphore");
goto freesem;
}
/* Initialize waiting semaphores to 0 (nothing in queue),
* and mutex semaphores to 1 (nobody playing with queue).
*/
semctl(qrequest->semid, WAITING, SETVAL, 0);
semctl(qrequest->semid, MUTEX, SETVAL, 1);
semctl(qresponse->semid, WAITING, SETVAL, 0);
semctl(qresponse->semid, MUTEX, SETVAL, 1);
/* fork off children */
for (i = 0; i < NCHILD; i++)
switch (pid = fork()) {
case 0:
child (i, qrequest, qresponse);
break;
case -1:
perror ("error forking child");
goto zapchild;
default:
nchild++;
break;
}
/* send "START" messages to all children */
for (i = 0; i < NCHILD; i++) {
qp = qptr(qrequest, load);
qp->command = START;
qp->argument = i;
qincr(qrequest, load);
/* signal a child that now there's a request */
V(qrequest, WAITING);
}
/* now wait for "COMPLETION" messages */
for (i = 0; i < NCHILD; i++) {
P(qresponse, WAITING);
qp = qptr(qresponse, unload);
printf ("response received, child %d, response %d\n",
qp->server, qp->response);
qincr(qresponse, unload);
}
zapchild:
/* send END messages to each child */
for (i = 0; i < nchild; i++) {
qp = qptr(qrequest, load);
qp->command = END;
qp->argument = i;
qincr(qrequest, load);
V(qrequest, WAITING);
}
/* wait for children to exit */
for (i = 0; i < nchild; i++) {
pid = wait (&status);
printf ("child process %d exited, status %d\n", pid, status);
}
freesem:
/* free up the semaphores for both queues */
if (qrequest->semid != -1)
semctl (qrequest->semid, IPC_RMID, 0);
if (qresponse->semid != -1)
semctl (qresponse->semid, IPC_RMID, 0);
freeseg:
shmdt (shmaddr);
shmctl (shmid, IPC_RMID, 0);
exit (0);
}
child (number, qrequest, qresponse)
int number;
struct queue *qrequest;
struct queue *qresponse;
{
struct qentry *qp;
int req, arg;
int retarg;
shmat(shmid, shmaddr, SHM_RND);
/* wait for there to be a request */
P(qrequest, WAITING);
/* wait to be the only one manipulating the unload pointer */
P(qrequest, MUTEX);
/* pull the request off the queue */
qp = qptr(qrequest, unload);
req = qp->command;
arg = qp->argument;
/* bump the unload pointer */
qincr (qrequest, unload);
/* signal that others can get into the queue */
V(qrequest, MUTEX);
if (req == START) {
/* fake some processing time */
sleep (arg * 2 + 1);
/* prepare a response and put it on the response queue */
P(qresponse, MUTEX);
qp = qptr(qresponse, load);
qp->command = COMPL;
qp->argument = arg;
qp->response = arg * 2;
qp->server = number;
/* bump the load pointer */
qincr(qresponse, load);
/* no longer manipulating the queue */
V(qresponse, MUTEX);
/* signal the response queue waiting semaphore */
V(qresponse, WAITING);
/* wait for an END message */
P(qrequest, WAITING);
P(qrequest, MUTEX);
qp = qptr(qrequest, unload);
if (qp->command != END)
printf ("server %d got message %d instead of end\n",
qp->command);
V(qrequest, MUTEX);
retarg = 0;
}
else retarg = 1;
shmdt(shmaddr);
exit(retarg);
}
拡張メモリはＢＳＤ版ＵＮＩＸに基礎を置き、標準の共有メモリ（システムコール）と同様の機能を提供します。
このインターフェースは、ハードウェアから独立しているので、メモリマッピングを使用するプログラムは（ハードウェアの設定変更があっても）システム・コンフィグレーション・ファイルやプログラムを変更せずにＲＴＵシステムで実行できる特長を持ち、主な用途として以下の２つがあります。
・メインメモリの一部をＲＴＵの管理下から外し、予約されたメモリ（リザーブドメモリ）としてプロセス間通信に使用する。
・ＶＭＥｂｕｓデバイスを外部メモリとして使用し、プロセス間通信あるいはプログラムＩ／Ｏデバイスとして使用する。
リザーブドメモリは、メインメモリの一部を予約メモリとして、ＲＴＵのメモリ管理から除外します。
したがって、主記憶を物理的に減少させるため、システムのパフォーマンスを劣化させる場合もありますが、ＲＴＵの管理を離れますので、プロセス終了後にもメモリイメージが存在し、使用方法が完全にユーザプロセスに任されます。
この機能は、ＶＭＥｂｕｓ上のメモリ、あるいはプログラムＩ／Ｏデバイスとしても使用できるため、仮想記憶を使用せず大容量の共有メモリを使用したい場合や、リフレクティブメモリを使用してマルチシステムを構成する場合にも使用されます。
この機能を動作させるためには、以下に示す手順で、カーネルを再コンフィグレーションして、再立ち上げをおこなう必要があります。
・以下のエントリでｎページ分のリザーブドメモリを確保し、マイナー番号 255 で、ｎは１６進で 0x1000 の整数倍であること）
extmem 255 sys:0 n sys
・mknod(8) を使用して メジャー番号３ マイナー番号 255 のキャラクタ型デバイスとして "/dev/resmem"を生成する
% mknod /dev/resmem c 3 255 （名前は何でも良い）
% chmod 666 /dev/resmem
% config MASSCOMP
% make -f uts.mk
% chmod 544 unix
% mv /unix /unix.old
% mv unix /unix
% reboot
ユーザに解放されているＶＭＥｂｕｓアドレス領域
７０００ ８０００
A16 C000 - FFFF C000 - FFFF
A24 200000 - 7FFFFF C00000 - FFFFFF
A32 C0000000 - FFFFFFFF 2E000000 - 2FFFFFFF
ユーザプロセスは、mmap(2) システムコールを使って、予約されたメモリセグメントのどの部分でもユーザ自身のデータ空間にマップすることができます。以下にmmap(2)の使用方法を示します。
┌────────────────────────────────────────────────┐
│ｍｍａｐ 物理メモリ空間の仮想メモリ空間へのマップ／アンマップ │
│ │
│形式 │
│ │
│ #include │
│ #include │
│ │
│ int mmap(addr,len,prot,share,fd,off) │
│ int munmap(addr,len) │
│ │
│引数 │
│ │
│ chaddr_t addr; マップする仮想メモリ空間の先頭アドレスでページバウウンダリでなくてはならない │
│ int len; システムページサイズの倍数であるバイト数（NBPG=4096） │
│ int prot; 保護属性（現在のインプリメントでは、無視され PROT_READ|PORT_WRITE になる） │
│ int share; 共有の有無（現在のインプリメントでは、無視され MAP_SHARED になる） │
│ int fd; ファイル記述子 │
│ int off; オフセット値 │
└────────────────────────────────────────────────┘
以下のプログラム例では、brk() によってデータセグメントの最終アドレスを得た後、roundup マクロでページ境界にそろえて、sbrk() によって SIZE バイト分（これはページサイズの倍数）データセグメントを拡張します。
その後、mmap() によって、マップしたいスペシャルファイルをマップします。
データセグメントの ページ境界を データセグメント 実デバイスを
アドレスを得る 計算する を伸張する 仮想空間にマップする
┌──┐ ┌──┐ ┌──┐ ┌──┐
│ │ │ │ │ │ │ │
│ │ │ │ │ │ │ │
│ │ │ │ │ │ │ │
│ │ │ │ │ │ │ │
sbrk→└──┘ roundup↓├──┤ ├──┤ │ │ /dev/resmem
addr→└──┘ brk├──┤ ├──┼────┬──┐
↓│SIZE│ ページ │ │ mmap →│ │ 実デバイス
addr+SIZE→└──┘ サイズ └──┴────┴──┘
� ´� �� ��
#include
#include
#include
#include
#define SIZE 0x80000 /* size of external memory (bytes) */
#define roundup(x) (((int)(x) + NBPG - 1) & ~(NBPG - 1))
char *addr;
main()
{
int memfd, fildes, status, size;
char *sbrk();
�� addr = sbrk(0);
�� addr = (char *)roundup(addr);
status = brk(addr+SIZE);
�� if (status != 0) {
perror("brk failed");
exit(1);
}
if ((memfd=open("/dev/resmem",O_RDWR))==-1) {
�� perror("mem open failed");
exit(1);
}
status=mmap(addr, SIZE, PROT_READ|PROT_WRITE, MAP_SHARED, memfd, 0);
if (status < 0 ) {
perror("mmap");
exit(2);
}
＜＜この区間 addr から SIZE 分の空間が拡張メモリにマップされている＞＞
status = munmap(addr, SIZE);
if (status < 0) {
perror("munmap");
exit(1);
}
}
ＲＴＵにおけるメモリマップでは、以下の図のようにメモリのボトムからＲＴＵが常駐し、メモリのトップから予約メモリがとられます。
FFFFFFFF┌────────┐ ┌───┐ ┌───────┐FFFFFFFF
┌─────┐ │ │ │ VBA │ │ │
│予約メモリ│ │ ＶＭＥ＿Ａ３２ │ │ │ │ユーザ使用領域│
├─────┤ │ │ │ │ │ │
│ │ 80000000├────────┤ │ │ ├───────┤C0000000
│ プロセス │ │ 未使用 │ │ │ │ │
│ 利用可能 │ n7FFFFFF├────────┤ │ │ │ │
│ 領域 │ │ システムメモリ │ │ │ │ │
│ │ n4000000├────────┤ │← DMA│ ├───────┤1FFFFFFF
├─────┤ │ 未使用 │ │ │ │ 直接アクセス │
│ │ n2FFFFFF├────────┤ │ │ │ 領域 │
│ ＲＴＵ │ │ ＶＭＥ＿Ａ２４ │ │ │ ├───────┤14000000
└─────┘ n2000000├────────┤ │ │ │ │
│ 未使用 │ │ │ │ │
n0FEFFFF├────────┤ │ │ │ │
│ ＶＭＥ＿Ａ１６ │ │ │ │ │
（ｎ：プロセッサ番号） n0FE0000├────────┤ │ │ │ │
│ローカルデバイス│ │ │ │ │
n0000000└────────┘ └───┘ └───────┘
システム・アドレス空間 ＶＭＥ＿Ａ３２アドレス空間
このメインメモリの利用状況は、以下のように /usr/adm/messages ファイルの表示をみることで確認できます。
UNIX/6.1: mc7000 unix generic
CPU1 CPU2 CPU3 are 68040's
MEMORY SUMMARY POOL CPU PAGES (BYTES)
33554432 installed 0 1 6896 (28246016)
4096 reserved
33550336 real
28246016 available
│
│
したがって、システム・アドレス空間の 0x40000000 + 33550336 から 4096 バイトがリザーブドメモリとして確保されているのが理解できます。
ＶＭＥ＿Ａ３２からこの空間を見ると 0x14000000 + 33550336 から 4096 バイトリザーブドメモリを確保したように見えます。
この空間は、ＶＭＥｂｕｓから、ＶＢＡのマップレジスタを経由せずにアクセス出来る、ダイレクトアクセス領域ですので、ＶＭＥ＿Ａ３２上のこの領域にＤＭＡ転送を行えば、システムメモリ上にＤＭＡ転送できる訳です。
後は、ＤＭＡデバイスをＶＭＥｂｕｓにプログラムＩ／Ｏデバイスとして登録し、制御すれば良いわけですが、注意しなければ行けない事が１つだけあります。
それは、割込みを禁止してＤＭＡ転送を実行しなければならないことです。プログラムＩ／Ｏデバイスで使用される extmem デバイスドライバには、割込みをハンドリングする機能が含まれていないため、ＤＭＡ終了割込みなどの割込みは認識されないスプリアス割込みとして処理されるため、システムに致命的な障害を与えてしまいます。よって、必ず割込みは禁止して使用して下さい。
また、通常のデバイス・ドライバで必要になる、オープン／クローズ／異常終了時の処理やタイムアウトの機能はすべてユーザプログラムで使用する必要があります。
以下に、ＩＣＯＮ社のＤＲ１１Ｗインターフェースを使用したプログラムを掲げます。
プログラムＩ／ＯとリザーブドメモリによるＤＭＡ転送のプログラム例
#include
#include
#include
#include
#include
#include
#include
#define RESMEM_SIZE 0x1000 /* size of reserved memory (bytes) */
#define DR11W_SIZE 0x1000
#define roundup(x) (((int)(x) + NBPG - 1) & ~(NBPG - 1))
#define RDMA 0x8000
#define RATN 0x4000
#define RPER 0x2000
#define MCLR 0x1000
#define CYCL 0x0100
#define TERM 0x0080
#define IENB 0x0040
#define SDIR 0x0010
#define FNC3 0x0008
#define FNC2 0x0004
#define FNC1 0x0002
#define GO 0x0001
#define ADD_MOD 0x09
struct DR11W_REGISTER
{
ushort dr_icr;
ushort dr_data;
unsigned char dr_ad_mod;
unsigned char dr_vec;
ushort dr_pulse_cmds;
ushort dr_pad0[5];
ushort dr_setaddrl;
ushort dr_countl;
ushort dr_curaddrl;
ushort dr_pad1;
ushort dr_setaddrh;
ushort dr_counth;
ushort dr_curaddrh;
};
struct RESMEM
{
ushort data[RESMEM_SIZE/2];
};
#define DATACNT (4096/2)
struct DR11W_REGISTER *dp;
struct RESMEM *bp;
struct RESMEMf *bpf;
trap()
{/* ＾Ｃの処理 */
dp->dr_icr = TERM; /*ＤＭＡ転送を強制終了*/
dp->dr_pulse_cmds = MCLR; /*ボードリセット*/
exit(1);
}
main()
{
unsigned long addr;
int i,drfd,drstatus;
static caddr_t draddr;
int bffd,bfstatus;
static caddr_t bfaddr;
/*ＤＲ１１ＷをプログラムＩ／Ｏデバイスとしてマップする*/
draddr = (caddr_t)sbrk(0);
draddr = (caddr_t)roundup(draddr);
drstatus = brk(draddr + DR11W_SIZE);
if(drstatus < 0){
fprintf(stderr,"Memory Alocation error status is %d\n",drstatus);
return(1);
}
if((drfd = open("/dev/dr11w",O_RDWR)) == -1){
fprintf(stderr,"open error:DR11W\n");
return(1);
}
drstatus = mmap(draddr,DR11W_SIZE,PROT_READ|PROT_WRITE,MAP_SHARED,drfd,0);
if(drstatus < 0) {
fprintf(stderr,"mmap io error status is %d\n",drstatus);
return(1);
}
dp = (struct DR11W_REGISTER *)draddr;
/*デバイスを初期化する*/
dp->dr_pulse_cmds = MCLR;
dp->dr_ad_mod = (unsigned char)ADD_MOD;
dp->dr_vec = 0x0;
dp->dr_icr = 0; /* desable interrupt */
/*ＤＭＡ転送アドレスを初期化する*/
addr = 0x14000000 + 16773120 ;
addr = (addr - 1) >> 1;
dp->dr_setaddrl = (addr & 0x0000ffff);
dp->dr_setaddrh = ((addr & 0xffff0000) >> 16);
dp->dr_countl = (unsigned short int)0;
dp->dr_counth = (unsigned short int)0;
/*＾Ｃ割込みによるトラップ関数を定義*/
signal(SIGINT ,trap);
/*リザーブドメモリをマップする*/
bfaddr = (caddr_t)sbrk(0);
bfaddr = (caddr_t)roundup(bfaddr);
bfstatus = brk(bfaddr + RESMEM_SIZE);
if(bfstatus<0){
fprintf(stderr,"Memory Alocation error status is %d\n",bfstatus);
return(1);
}
if((bffd = open("/dev/resmem",O_RDWR)) == -1) {
fprintf(stderr,"open errojr: reserved memtory\n");
return(1);
}
bfstatus = mmap(bfaddr,RESMEM_SIZE,PROT_READ|PROT_WRITE,MAP_SHARED,bffd,0);
if(bfstatus < 0){
fprintf(stderr,"mmap io error status is %d\n",bfstatus);
}
bp =(struct RESMEM *)bfaddr;
bpf =(struct RESMEMf *)bfaddr;
/*リザーブドメモリ領域にデータをセットする*/
for(i=0;i< DATACNT;i++){
bp->data[i] = i;
}
/*ＤＭＡ転送アドレスをセットする*/
dp->dr_setaddrl = addr & 0x0000ffff;
dp->dr_setaddrh = ((addr & 0xffff0000) >> 16);
dp->dr_countl = (unsigned short int )DATACNT;
dp->dr_counth = 0;
/*ＤＭＡ転送の方向を制御する*/
#if READ
dp->dr_icr = SDIR;
#else
dp->dr_icr = 0;
#endif
/*ＤＭＡ転送を行う*/
dp->dr_pulse_cmds = GO|CYCL|RDMA;
/*ＤＭＡ転送の終了をフラグスキャンして待つ*/
while(1){
if(dp->dr_icr & RDMA) break;
}
#if READ
for(i=0;idata[i]);
}
#endif
/* ＤＲ１１Ｗをリセットしデバイスをクローズする */
dp->dr_pulse_cmds = MCLR;
close(drfd);
close(bffd);
}
６章で、説明するスレッドを使用すると、複雑なシステムコールを使用せずに共有メモリを自然に実現できるため、セマフォ等の排他制御機構と組み合わせることで、大量のデータを効率よく共有できます。
ＲＴＵのスレッドでは、ＵＮＩＸ標準のｓｉｇｎａｌに加えてＡＳＴをもスレッド単位でサポートし、システムコールも多数スレッド対応になっています。
スレッド対応と言う意味は、システムコールの実行をスレッド単位で実行出来るため、他のスレッドに影響を与える事無く、システムコールを呼び出せます。したがって呼びだした結果、封鎖されるのは、プロセス全体ではなくスレッドだけです。
スレッド対応システムコール
_getuptime adjtime alarm astpause canclkast
chpage clkast dup exit fchmod
fchown fclkast *fcntl flock fstat
fstatfs *fsync ftime *ftruncate getdirentries
getdtablesize getid getgroups gethostid gettimer
getpagesize getpid getpriority getrusage gettimeofday
gettimer getuid hold ioctl kill
lock lockf madvise mincore *mpadvise
mprotect mremap nice oldstat oldselect
olockf osigvec pause pprotect prof
ptrace *read *readv release rlockf
rread rwrite schedget schedset seek
select _sendast setgid setgroups sethostid
setitimer setpgrp setpri setpriority setregid
setreuid settimeofday settimer setuid sig
sigblock sigpause sigsetmask sigstack sigvec
sps_creat sps_ctl sps_cycend sps_map t_create
t_csleep t_cwake1 t_cwakeup t_cwant t_exit
t_getdata t_getdptr t_geterror t_gettid t_kill
t_ncancel t_nhandler t_notify t_nowait t_sendast
t_setdata t_setname t_wakeup t_wait time
times ucbfcntl umask utssys vi_create
vi_ctl vi_delete vi_map vi_mapsource vi_nonsense
vi_sense vi_unmap wait *write *writev
yield
＊ はブートプロセッサにスイッチして実行することがある
スレッドによる共有メモリの制限は、プログラムの仮想空間と同一ですので、ＲＴＵのソフトウェアリミットおよびハードウェアリミットによって制限をうけます。ハードウェアリミットは、コンフィグレーションファイルの以下の情報に依存しています。
maxtsiz 仮想空間で使用できるテキスト・セグメントの最大サイズ
maxdsiz 仮想空間で使用できるデータ・セグメントの最大サイズ
maxssiz 仮想空間で使用できるスタック・セグメントの最大サイズ
すでに述べたように、ＲＴＵはすべてのプロセッサで協調して動作します。つまり、すべてのプロセッサで同時並列に実行され、ユーザは、それを意識する必要はありません。
したがって、基本的にはすべてのプロセッサで同等の処理ができますが、内部的にブートプロセッサで実行されるシステムコールもありますが、これも内部で自動的にスイッチングされるためユーザはそれを意識することはありません。（詳細を下表に示します。）
ブートプロセッサのみで実行できるシステムコール
accept acct bind break cinfo
close connect create exec exece
fork gethostname getpeername getsockname getsockopt
link listen mkdir mknod mmap
mount munmap nfs_svc open pinfo
plockin pmapm punlock punmap reboot
recv recvfrom recvmsg rename rmdir
send sendmsg sendto sethostname setsockopt
shutdown socket socketpair symlink sync
umount unlink unmount vadvise vfork
vswapon
すべてのプロセッサで実行できるシステムコール
access chdir chmod chown chroot
exportfs getcwd getmsg getpmsg getrlimit
getuniverse gtty lstat msgctl msgget
msggrcv msgsnd nfs_getfh oldstat openobject
pipe poll putmsg putpmsg quota
readlink sem semctl semget semop
setquota setrlimit setuniverse shm sinfo
stat statfs stime stty truncate
ulimit utime utimes vhangup vlimit
vsetpgrp vtime async_daemon setdomainname getdomainname
例外条件としては、以下の２つがあります。
１）システムの立ち上げは、ブートプロセッサのみで行える。
２）リアルタイムスレーブの機能は、非ブートプロセッサのみで実行できる。
またＲＴＵにおける入出力命令に関しては、以下のように２つの方法が存在します。
１）ＲＴＵのデバイスドライバを介する方法。
２）メモリマップデバイスとしてプログラムＩ／Ｏを使用する方法。
いずれの方法でも、どのプロセッサからでも実行可能です。ただし、例外として、デバイスドライバが非対称型で記述されている場合は除きます。標準的なＲＴＵのデバイス・ドライバはすべて対称型で記述されているため、たとえばディスクに対する read,write システムコールは、すべてのプロセッサで実行できます。
多重スレッドは、ライトウェイトプロセスとも呼ばれ、ＭＡＣＨ、ＳｕｎＯＳ５．０、ＲＴＵ等に実装され、ＰＯＳＩＸにおいても、スレッドの機構がリアルタイム小委員会で検討されています。シングルプロセッサにおいてもスレッドは興味深い特徴を持っていますが、本当にその効果を発揮するのは、共有メモリ型マルチプロセッサで、スレッドを異なるプロセッサで並列に実行した場合です。（これを多重スレッドと呼びます）
こういったマルチプロセッサ上に実現された多重スレッドは、リアルタイムＵＮＩＸにおいて、たいへん大きな意味をもっています。それは、リアルタイム処理において満たさなくてはならない基本仕様を多重スレッドを使って、効果的に実現できるからにほかならないからです。
たとえばスレッドは、プロセスのもつすべてのデータ構造を受け継ぎながら、他のスレッドに影響を与える事無く封鎖できるため、信頼性の高い非同期イベント通信機構と組み合わせる事により先に説明した非同期入出力機構を簡単に実現できます。
こういった特徴を利用すればマルチプロセッサ上で、計算のベクタ化や、データ移動のダブルバッファ法、パイプライニングやキューイング等の手法を効率よく実現できます。
ここでは、ＲＴＵに実装された多重スレッドを中心にＭＡＣＨとの比較をまじえ簡単に解説します。
下の表にＲＴＵで使用されるスレッド関係のプリミティブシステムコールをまとめておきます。
多重スレッドの同期と排他制御に使用されるプリミティブシステムコール
┌──────────────────┬───────────────────────────────┐
│t_getdptr(); │スレッド用のプライベート構造体のポインタを得る │
│t_setdata(data); │子スレッドに引渡す整数データをプライベート構造体にセットする │
│t_getdata(); │親スレッドから引渡された整数データをプライベート構造体から得る│
│t_geterror(); │カレントスレッドに通知されたシステムエラーコードを得る │
│t_gettid(); │カレントスレッドのスレッドＩＤを得る │
│t_setname(id,thread_name); │スレッドＩＤで指定したスレッドに名前をつける │
│struct threaddata { │プライベート構造体 │
│ int td_data; │t_create,t_getdata,t_setdata で使用する整数データ │
│ int td_errno; │システムがセットするエラー番号 t_geterror で得る │
│ int td_tid; │システムにただ１つのスレッドＩＤ番号 t_gettid で得る │
│ char *td_staacktop; │スレッドのプライベートスタックアドレスが t_create でセット │
│ char td_comm[MAXCOMLEN]; │t_create,t_setname で使用するスレッド名 │
│} threaddata_t; │ │
│t_create(func,stkptr,arg); │スレッドを生成する stkptr は、スレッドのスタックトップを指定 │
│t_exit(status); │カレントスレッドを終了する │
│t_wait(thread_id); │指定スレッドの資源を t_exit 時に解放させ、その終了を待つ │
│t_nowait(thread_id); │指定スレッドの資源を t_exit 時に解放させ、その終了を待たない │
│t_csleep(condition); │カレントスレッドを condition を獲得までスリープ状態にする │
│t_cwake1(condition); │condition でスリープしているスレッドを１つ実行可状態にする │
│t_cwakeup(condition); │condition でスリープしているスレッドをすべて実行可状態にする │
│t_cwant(condition); │t_csleep,t_cwakeup で使用する condition を獲得／解放する │
│t_twakeup(thread_id); │スレッドＩＤで指定したスレッドを実行可状態にする │
│t_notify(channel); │スレッド用の非同期イベント通知チャネルを生成する │
│t_ncancel(channel); │スレッド用の非同期イベント通知チャネルを消去する │
│t_kill(thread_id,sig); │スレッドＩＤで指定したスレッドにシグナルを送信する │
│t_sendast(thread_id,lo_handler,func,│スレッドに非同期イベントを通知（ＡＳＴを送信）する │
│ param,priority); │ │
│t_nhandler(lo_handler,func,param, │スレッド用の非同期イベント通知チャネルのワークエリアを用意する│
│ priority); │ │
└──────────────────┴───────────────────────────────┘
スレッドは、プロセスと比較すると幾つかの点でその考え方が異なります。
似ている点：スレッドはｆｏｒｋのように動的に実行を開始しｗａｉｔのように同期することができます。
：スレッドは子プロセスのように、オープンされたファイルを引継ぎ、終了時にエラー番号を返します。
異なる点 ：プロセスと違って親子関係がなく、必ずしもスレッド生成時に実行を始めなくてもかまいません。
：スレッドは、システム割当ＩＤとカレント実行ＩＤの２つのスレッドＩＤを持ちます。
：スレッドは子プロセスと違って、静的宣言されたデータと外部宣言されたデータをすべて共有します。
プロセスはルートスレッドを含みます。すなわちプロセスを１つ生成することはルートスレッドを１つ作成する事と同じです。
下図にプロセスとスレッドの関係を示します。注意して欲しいのは、スタックのあつかいで、スタックの管理がユーザに任されている点です。この例で示されているプログラムは、内部動作が理解し易いようにシステムプリミティブだけで作成されていますが、通常はこの上位層のスレッドライブラリを使用します。
ＲＴＵのスレッドライブラリを使用するとスタック領域の確保等の面倒くさいプログラムは不要になり、色々な同期機構もこのレベルで実現されています。
またスレッドの自動変数以外のプライベートデータの割当は、Ｃコンパイラの管理に任されることになります。もしこの管理をコンパイラがサポートしない場合、自動変数以外はすべてグローバル変数でユーザの管理の元でプログラムする事になります。
プロセスとスレッドの関係
┌─────────────┐──┬─
│ プロセス環境 │ │
┌───────┬────────────────────────┼─────────────┘ │
│データ │テキスト・セグメント │プロセスＩＤ、グループＩＤ │
│ セグメント│ │親プロセスＩＤ、 ル│
│ int array[2];│ ルートスレッド（初期プロセス） │プロセス優先度 ｜│
│ │┌──────────────────────┐│ファイル・ディスクリプタ ト│
│ ││main() { int id,status; ││セグメントサイズリミット ス│
│ ││ id=t_create(child_thread,stack[SIZE-1],0)││ レ│
│ ││ root_thread(1); ││ ッ│
│ ││ status=t_wait(id); ││ ド│
│ ││} ││ 空│
│ │├──────────────────────┤│ 間│
│ ││root_thread(arg) int arg;{ ││ │
│ ││ array[arg]*=array[arg]; ││ │プ
│ ││} ││ │ロ
│ │└──────────────────────┘│ ─┤セ
│ 子スレッド │ │ │ス
│┌──────┼┐ ┌┼─────────────┐ │空
││stack[SIZE];││ ││スレッドプライベート環境 │ 子│間
│└──────┼┼──────────────────────┼┼─────────────┘ ス│
│ ││child_thread(arg) int arg;{ ││ システム割当スレッドＩＤ レ│
│ ││ arg = t_getdata(); ││ スレッド名 ッ│
│ ││ array[arg]++; ││ エラー番号 ド│
│ ││ t_exit(arg); ││ 引き渡しデータ領域 空│
│ ││} ││ スレッドスタックポインタ 間│
│ │└──────────────────────┘│ │
│ │ │ │
└───────┴────────────────────────┘ ──┴─
│ ユーザプロセス管理領域 │ システム管理領域 │
├────────────────────────────────┴─────────────┤
│ プロセス空間 │
ＲＴＵ、ＭＡＣＨともにスレッド生成時に与えることのできるデータは３２ビット整数ですが、スレッド終了時のデータ受け渡しについては異なっています。
ＲＴＵでは t_wait で、スレッドの終了ステータスを返しますが、ＭＡＣＨの場合は、cthread_join で任意の値を返すことが理解できます。
この点を除いて、スレッドの生成、起動、終了に使用されるライブラリは、ＲＴＵ，ＭＡＣＨとも同じようなものです。この fork - wait モデルの特別な場合、つまり相手の終了を待たない場合は、以下のようにＲＴＵ、ＭＡＣＨともスレッド生成時に指定します。
┌──────────────────────────────────────────────────┐
│ ＲＴＵ ＭＡＣＨ │
│ 前処理 │
│ #include #include │
│ void cthread_init(); │
│ 生成 │
│ int mp_create(func,arg); cthread_t cthread_fork(func,arg); │
│ void (*func)(); any_t (*func)(); │
│ int arg; any_t arg; │
│ 終了 │
│ void mp_exit(exitcode); cthread_exit(result); │
│ int exitcode; any_t result; │
│ 終了待ち │
│ int t_wait(thread_id); any_t cthread_join(t); │
│ int thread_id; cthread_t t; │
│ 終了を待たない │
│ int t_nowait(thread_id); void cthread_detach(t); │
│ int thread_id; cthread_t t; │
│ │
│ ［ただし any_t は任意の型で、 cthread_t は３２ビットサイズのスレッドＩＤ］ │
│ │
│ スレッドの終了を待つ場合 │
│ │
│ id = mp_create(func,arg); id = cthread_fork(func,arg); │
│ | | │
│ status = t_wait(id); any_t = cthread_join(id); │
│ │
│ スレッドの終了を待たない場合 │
│ │
│ t_nowait(mp_create(func,arg)); cthread_detach(cthread_fork(func,arg); │
└──────────────────────────────────────────────────┘
ＲＴＵの場合、アイドルスレッドを予め生成しておき後から実際のコードを割当て実行する方法があります。これは、リアルタイム処理において出来るだけ高速にスレッドを制御するためです。
アイドルスレッドの割当 実スレッドの割当と起動
mp_create(NULL,NULL); mp_call(func,arg); mp_calln(func,arg,n);
スレッド生成起動 スレッド終了待ち スレッド生成 スレッド起動
↓ ↓ ↓ ↓
ルートスレッド ──┐ ┌────┐ ┌─ ─┐ ┌┐ ┌─────
│ 5250μ │ │170μ │ │5200μ││84μ│
ＲＴＵ ├───┬┘ ┌─┴───┘ └───┘├──┴┐
│3376μ│ │ │151μ │
子スレッド │ └───┘ │ └────
↑スレッド終了
７０００シリーズ（ＭＣ６０８０４０ ２５ＭＨｚ）のスレッドの生成、起動、終了
並列計算モデルにおいては、スレッドに対する処理の依頼と同期にかかるコストの和は、そのままスレッドの粒度になるため、スレッドを予め生成しておき後から実際のコードを割当て実行する方法が有利です。
図で理解できるようにスレッドの起動のみのコストは、１５１μ秒であり、終了待ちのコストは、１７０μ秒です。
したがって、処理を２つに分解する場合のコストは、３２１μ秒かかる事になり、並列処理を行えるのは約６５０μ秒以上の処理になります。ＭＣ６８０４０でのＣＰＩは、１．３ですので、この時間に行える処理は、約１２３５０命令でありこれ以上の処理であれば並列実行の効果が得られます。
以下のプログラム例では、ベクトル列の和を、２つのスレッドに分割実行しているあいだルートスレッドでは、２つのベクトル乗算を並列に実行しているものです。
#include
#define MAX 50
float sum1, sum2, dot;
void vinit (), vsum1 (), vsum2 (), vdot ();
main ()
{
int t[2], ex, result;
float vec1[MAX], vec2[MAX];
vinit (vec1, vec2); /* 配列の初期化 */
/* スレッドを生成し総和を計算する。結果はグローバル変数(sum1,sum2)に格納される */
if ((t[0] = mp_create (vsum1, vec1)) <= 0)
perror("Couldn't create first thread"), exit(1);
if ((t[1] = mp_create (vsum2, vec2)) <= 0)
perror("Couldn't create second thread"), exit(2);
/* ルートスレッドは、２つのベクトルの乗算を計算した後、２つの子スレッドの
終了を待ち、結果を出力する */
vdot (vec1, vec2);
ex = t_wait (t[1]);
ex = t_wait (t[2]);
printf ("sum1 = %f, sum2 = %f, dot = %f\n", sum1, sum2, dot);
/* ルートスレッド上で結果の確認をシーケンシャルに行い終了する */
vfy (&result, vec1, vec2);
exit (result);
}
/* ベクトルの総和を求める関数、結果は外部変数に格納される */
void vsum1 (v)
float v[];
{
int i;
sum1 = 0;
for (i = 0; i < MAX; i++)
sum1 += v[i];
mp_exit (0); /* スレッドの終了ステータスでルートスレッドに０を返す */
}
void vsum2 (v)
float v[];
{
int i;
sum2 = 0;
for (i = 0; i < MAX; i++)
sum2 += v[i];
mp_exit (0); /* スレッドの終了ステータスでルートスレッドに０を返す */
}
void vdot (v1, v2) /* ２つのベクトルの乗算を求める関数 */
float v1[], v2[];
{
int i;
dot = 0;
for (i = 0; i < MAX; i++)
dot += v1[i] * v2[i];
}
void vinit (v1, v2) /* テスト配列の初期化 */
float v1[], v2[];
{
int i;
for (i = 0; i < MAX; i++)
{
v1[i] = i;
v2[i] = -i;
}
}
vfy (result, vec1, vec2)
float vec1[], vec2[];
int *result;
{
int i;
float locsum1 = 0, locsum2 = 0, locdot = 0;
*result = 0;
for (i = 0; i < MAX; i++) {
locsum1 += vec1[i];
locsum2 += vec2[i];
locdot += vec1[i] * vec2[i];
}
if (locsum1 != sum1) {
printf ("sum 1 does not agree, local= %f thread= %f\n",
locsum1, sum1);
*result = 1;
}
if (locsum2 != sum2) {
printf ("sum 2 does not agree, local= %f thread= %f\n",
locsum2, sum2);
*result = 1;
}
if (locdot != dot) {
printf ("dots do not agree local= %f thread= %f\n",
locdot, dot);
*result = 1;
}
}
この章では、スレッドで使用する事の出来る同期と排他制御の機構について解説します。
一般的に、どの様な計算機システムであっても、同期命令は必ず実装されていますが、特にマルチプロセッサシステムにおいて同期命令は、システムの性能を左右してしまうほど重要なものです。
通常このような同期命令は、単一のバスサイクルにおいて排他的にメモリの読み書きを実行する test-and-set（ＴＡＳ） か compare-and-swap（ＣＡＰ）命令で実現されています。（例外的にＲ３０００シリーズは持っていません）
この命令を実行している間は、メモリバスはロックがかけられ、他のバスマスタはメモリバスを使用できないようになっています。このような機構により同期命令は正当性を保証される訳ですが、リアルタイムシステムの様に高速に同期処理を行わねばならない場合、この同期にかかる時間は重要です。
なぜなら、頻繁にバスロックが起こると、システムのパフォーマンスが急激に低下するからです。
この問題は、メモリ管理とキャッシュの章で述べた結果と一致します。つまり、ライトスループロトコルでは、書き込みの際にキャッシュとメモリに書き込むため、メモリバスをもロックしますが、コピーバックプロトコルにおいては、複数のＴＡＳ命令が異なったメモリアドレス上で実行されている限り、メモリバスをロックする事がないため、マルチプロセッサやリアルタイム処理において最適であるといえます。
ＲＴＵとＭＡＣＨにおいて、実装されている排他制御機構はたいへん似ています。この同期制御は、あまりにも有名ですので説明は省略しますが、注意して置いて欲しいのは、使用法を誤るとデットロックに陥ってしまう危険があることです。一度デットロックに陥るとシステムは止まってしまいます。
デッドロックの危険は、セマフォを使用して作成したサブルーチンを多重スレッドで使用する場合にも起こります。ＲＴＵでは、このような危険を回避する目的で、再帰的に呼び出せる排他制御機構も実装しています。（内部的にカウンタをもっています）。
┌──────────────────────────────────────────────────┐
│ ＲＴＵ ＭＡＣＨ │
│ 排他制御変数の割当 │
│ mplock mp_lock_alloc(); mutex_t mutex_alloc(); │
│ 排他制御変数の解放 │
│ void mp_lock_free(key); void mutex_free(key) │
│ ロックに失敗するとスリープする │
│ void mp_lock_set(key); void mutex_lock(key); │
│ ロックを解除しスリープしているスレッドがあった場合１つだけ起動する │
│ void mp_lock_clr1(key); void mutex_unlock(key); │
│ ロックを解除しスリープしているスレッドをすべて起動する │
│ void mp_lock_clr(key); │
│ ロックに失敗してもスリープせず帰ってくる │
│ void mp_lock_try(key); void mutex_try_lock(key); │
│ ロックに失敗した場合でもスリープせずライブラリの中でビジーウェイトする │
│ void mp_lock_spin(key); │
│ 排他制御変数型 │
│ mplock key; mutex_t key; │
└──────────────────────────────────────────────────┘
ロック アンロック
↓ ↓ ┌───────┬─────┬─────┐
─┐ ┌───────┐ ┌───── │ ライブラリ名 │ 呼出時間 │ 効果時間 │
│呼出時間│ │呼出時間│ ├───────┼─────┼─────┤
└────┘┌──────┼────┴┐ │mp_lock_alloc │１６７７μ│ ｎａ │
│ │ 効果時間 │ │mp_lock_free │ ４９μ│ ｎａ │
───────┘ │ └──── │mp_lock_clr1 │ ９μ│ ｎａ │
↑スリープ状態 実行可状態 実行状態 │mp_lock_set │ ５μ│ １５４μ│
ロック要求 │mp_lock_try │ ５μ│ １５μ│
│mp_lock_spn │ ４μ│ １５μ│
└───────┴─────┴─────┘
このプログラムは、ファイルを共有して排他的にアクセスする例です。
排他制御機構には、mp_lockファミリを使用しています。
#include
#include
mplock filelock;
#define MAX 1000
int vec[MAX];
void vinit (), sum_file_entries ();
FILE *fp;
main ()
{
int t1, t2, t3, ex1, ex2, ex3, result;
/* Initialize test data and allocate a lock */
vinit ();
filelock = mp_lock_alloc ();
/* Open the file */
fp = fopen ("thrsnmtf.data", "r");
if (fp == NULL)
exit (-1);
/* Run function on 4 threads, including root */
t1 = mp_create (sum_file_entries, 0);
t2 = mp_create (sum_file_entries, 0);
t3 = mp_create (sum_file_entries, 0);
sum_file_entries ();
ex1 = t_wait (t1);
ex2 = t_wait (t2);
ex3 = t_wait (t3);
/* Verify and clean up */
mp_lock_free (filelock);
rewind (fp); /* So vfy() starts at the beginning of the file */
vfy (&result);
exit (result);
}
tprivate int entries[MAX];
void sum_file_entries ()
{
int i, sum, element, number, thread;
/* Set the lock, then do one read before the loop */
mp_lock_set (filelock);
fscanf (fp, "%d %d %d ", &thread, &element, &number);
while (!feof (fp)) { /* While there's still data in the file */
for (i = 0; i < number; i++)
fscanf (fp, "%d ", &entries[i]);
/* Allow just one thread to enter, because the file should be read by
one thread at a time */
mp_lock_clr1 (filelock);
if ((element < 0) || (element >= MAX) || (number > MAX))
exit (2); /* File contains erroneous data */
/* The calculations: add the data in the line and store internally */
sum = 0;
for (i = 0; i < number; i++)
sum += entries[i];
vec[element] = sum;
/* If another thread reaches the end of the file first, this last
fscanf reads garbage. However, the effect is harmless because the
loop does not repeat. */
mp_lock_set (filelock);
fscanf (fp, "%d %d %d ", &thread, &element, &number);
}
条件変数を使った場合、排他制御変数と組合わせることによってキューイングモデルを記述でき、より安全にプログラムすることが可能です。
┌──────────────────────────────────────────────────┐
│ ＲＴＵ ＭＡＣＨ │
│ 条件変数の割当 │
│ mpcond mp_cond_alloc(); condition_t condition_alloc(); │
│ 条件変数の解放 │
│ void mp_cond_free(condition); void condition_free(condition); │
│ これを呼びだすと key をアンロックし condition でスレッドが停止状態になる │
│ void mp_lock_wait(key,condition); void condition_wait(condition,key); │
│ condition で停止状態にあるスレッドを１つだけ実行可能状態にする │
│ void mp_signal(condition); void condition_signal(condition); │
│ condition で停止状態にあるスレッドを全部実行可能状態にする │
│ void mp_broadcast(condition); void condition_broadcast(condition); │
│ │
│ mplock key; mutex_t key; │
│ mpcond condition; condition_t condition; │
└──────────────────────────────────────────────────┘
この同期機構は、使い方が少し複雑ですが下記のようにクリティカルセクションで排他情報を制御します。この例では、マルチクライアント−シングルサーバモデルを記述しています。
┌──────────────────────────────────────────────────┐
│クライアント部 │
│ │
│ mp_lock_set(key); │
│ /* この部分は key により排他制御になっているためだれも入れない */ │
│ mp_lock_wait(key,condition); │
│ /* この部分を実行すると key がアンロックされるため他のスレッドが */ │
│ /* クリティカルセクションに入れるようになるが、自身は condition */ │
│ /* 停止状態におかれるため先に進まず他のスレッドの mp_signal を待つ */ │
│ mp_lock_set(key); │
│ /* 他のスレッドの mp_signal によって起動後直ちに key を確保し */ │
│ /* 排他制御情報をアクセスする */ │
│ mp_lock_clr1(key); │
│ /* 排他制御変数をアンロックする */ │
│ mp_signal(condition); │
│ │
│サーバ部 │
│ │
│ mp_lock_set(key); │
│ /* この部分は key により排他制御になっている */ │
│ mp_clr1(key); │
│ mp_signal(condition); │
└──────────────────────────────────────────────────┘
クライアント set→wait─────────────────────────set→clear───────
↓ │ signal
クライアント set→wait──────────set→ clear ─┬─────┼───────────
↓ │ signal │ │
クライアント set→wait───────┼──────set→clear─┴───────────
↓ │ signal
サーバ set──→clear ─┴────────────────────────
signal
set mp_lock_set ─────実行状態 → クリティカルセクション
wait mp_lock_wait ─────停止状態
clear mp_lock_clr1 │
signal mp_signal ─┘ 他のスレッドをアクティブにする
#include
#include
#include
#define SERVER_THREAD 3 /* Number of threads for servers (enqueuing) */
#define CLIENT_THREAD 2 /* Number of threads for clients (dequeuing) */
#define INTERVAL 1 /* For test, seconds between signals */
#define TIMES 5 /* For test, number of signals to send */
#define TRUE 1 /* Used in while constructs to loop forever */
#define HIGH_PRIME 1074889 /* Seed for random-number generation */
#define QUEUE_DATA_SPACE 4096 /* Arbitrary size (in elements) for queue */
#define TEST_SIGNAL SIGINT /* Signal that triggers the servers */
/* Arbitrary limit to number of items queued simultaneously */
#define QUEUE_POINTER_SPACE (SERVER_THREAD+1)
/* The queue itself, and pointers */
float queue_data[QUEUE_DATA_SPACE];
/* Pointers to per-item information about queue */
struct queue_info
{
int queue_base; /* First element of this queued item */
int num_elem; /* Number of elements to be dequeued */
} queue_pointers[QUEUE_POINTER_SPACE];
/* The indexes are counters so that up-to-date values are available */
mpcounter bottom_queue_index, /* Where each server adds data */
bottom_pointers_index, /* Where each server adds pointer */
top_pointers_index, /* Where each client removes pointer */
queued_items; /* How many items are currently on queue */
/* Locks for atomic access to the _index pointers that control arrays
and a global lock for the queue itself */
mplock server_lock, /* Bottom_queue_index and bottom_pointers_index */
client_lock, /* Top_pointers_index */
queue_lock; /* Queued_items */
/* Conditions to signal between server and client that the queue is available */
mpcond server_done, /* Clients wait, servers signal */
client_done; /* Servers wait, clients signal */
main ()
{
int i, ret;
void server_scheduler (), client (), init_rand (), test_servers ();
sigset (TEST_SIGNAL, SIG_IGN);
bottom_queue_index = mp_counter_alloc ();
bottom_pointers_index = mp_counter_alloc ();
top_pointers_index = mp_counter_alloc ();
queued_items = mp_counter_alloc ();
server_lock = mp_lock_alloc ();
client_lock = mp_lock_alloc ();
queue_lock = mp_lock_alloc ();
client_done = mp_cond_alloc ();
server_done = mp_cond_alloc ();
/* Create an arbitrary number of threads for server and client functions */
for (i = 0; i < SERVER_THREAD; i++)
ret = mp_create (init_rand, HIGH_PRIME);
for (i = 0; i < CLIENT_THREAD; i++)
ret = mp_create (client, 0);
while (mp_counter_fetch (mp_idle_count) < SERVER_THREAD)
;
/* This thread handles the signals */
sigset (TEST_SIGNAL, server_scheduler);
/* This function could automate testing by delivering signals, but the
program hangs because the client and server threads never exit. An
extra level of synchronization would be required to ensure that all
clients and servers finished their work before exiting or being killed. */
/* ret = mp_create(test_servers, t_gettid() ); */
while (TRUE)
;
}
/* Signal handler */
/* This is a crude scheduler, starting up server functions to add data
to the queue in response to interrupts from the user's terminal.
The scheduling algorithm simply starts as many servers as there are
idle threads, on the assumption that every resource is thus being used
to the fullest. */
void server_scheduler (sig)
int sig;
{
int current_idle,
ret;
void fill_buffer (), server ();
/* Signal disabled until this handler can finish its work */
sigset (sig, SIG_IGN);
if (sig != TEST_SIGNAL)
fprintf (stderr, "Unrecognized signal received: %d\n", sig);
else
{
/* Make sure at least one idle thread is available, then run the
server function on all available threads */
while (!(current_idle = mp_counter_fetch (mp_idle_count)))
;
ret = mp_calln (server, 0, current_idle);
}
sigset (sig, server_scheduler);
}
/* Server -- enqueues items */
/* This sample server simply adds an arbitrary number of floating-point numbers,
randomly-generated, to the queue space. In a real application, the
corresponding function would retrieve data from some external source.
No two servers can run simultaneously, because while one is waiting, its data
or pointer space might be overrun by other servers. However, all parts
of the function can run in parallel with a client, except those locked
by the queue_lock. */
void server ()
{
int local_queue_index, local_pointers_index, ret;
/* Arbitrary number of elements per item */
int num_elements = mp_tid * 10;
/* Lock the queue so that other servers cannot update it (clients can
still run). Get current value of pointer to bottom of queue's data
space, and update it for a wrap-arround queue. */
mp_lock_set (server_lock);
/* Retrieve index for this thread's use, and update index for subsequent
threads' use */
local_queue_index = mp_counter_fetch (bottom_queue_index);
ret = mp_counter_init (bottom_queue_index,
(local_queue_index + num_elements) % QUEUE_DATA_SPACE);
/* Get a spot on the bottom of the queue_info structure to put
information needed by a client to dequeue data: the starting point
within the queue's data space and the number of floating-point items
to dequeue. Update pointer to bottom, taking into consideration that
the pointers themselves form a wraparound queue. */
local_pointers_index = mp_counter_fetch (bottom_pointers_index);
ret = mp_counter_init (bottom_pointers_index,
(local_pointers_index + 1) % QUEUE_POINTER_SPACE);
/* Prevent the overwriting of other data on the queue, by checking the
queue counter to see whether the queue is full. In theory, results of
the program could be corrupted because a server overwrites unread
space in the queue_data array; the program does not check for
overruns, but assumes that the array is sufficiently large for
whatever data is waiting to be dequeued. */
mp_lock_set (queue_lock);
/* If the queue is full, wait on client_done for a client. Otherwise,
just proceed. Clear the lock in any case. */
if (mp_counter_fetch (queued_items) >= QUEUE_POINTER_SPACE)
mp_lock_wait (queue_lock, client_done);
mp_lock_clr (queue_lock);
/* Put information needed by a client in the queue_info structure */
queue_pointers[local_pointers_index].queue_base = local_queue_index;
queue_pointers[local_pointers_index].num_elem = num_elements;
/* Place items on the queue. For this example, num_elements causes an
arbitrary number of floating-point values to be placed on the queue,
while the values themselves are simply generated at random. If the
pointer reaches the top of the queue, it wraps around to the bottom.
*/
while (num_elements--)
{
queue_data[local_queue_index] = rand ();
if (++local_queue_index >= QUEUE_DATA_SPACE)
local_queue_index = 0;
}
/* Increment counter to reflect a new element on the queue */
mp_lock_set (queue_lock);
if ((ret = mp_counter_incr (queued_items)) > QUEUE_POINTER_SPACE)
fprintf (stderr, "Server %d: queue pointers overflowed\n", mp_tid);
mp_lock_clr (queue_lock);
/* Clear lock so other servers can enqueue data, and wake up one waiting
client thread, if any are waiting for data */
mp_lock_clr (server_lock);
mp_signal (server_done);
/* Thread goes idle at the end of this function. The same server
function can be run on this thread later */
}
/* Client -- dequeues and processes items */
/* This client loops endlessly, removing data when it exits and otherwise
waiting for a server to signal that new data has arrived.
No two clients can run simultaneously, because while one is waiting, others
could wrap around and conflict with its data or pointer space. However,
all parts of the function can run in parallel with a server, except those
locked by the queue_lock. */
void client ()
{
int local_queue_index, local_pointers_index, num_elements, ret;
void process_item ();
while (TRUE)
{
/* Lock the queue so that other clients cannot update it (servers can
still run). Get current value of pointer to top of queue_info
array, and update pointer, taking into consideration that the
pointers themselves form a wraparound queue. */
mp_lock_set (client_lock);
/* Retrieve index for this thread's use, and update index for
subsequent threads' use */
local_pointers_index = mp_counter_fetch (top_pointers_index);
ret = mp_counter_init (top_pointers_index,
(local_pointers_index + 1) % QUEUE_POINTER_SPACE);
/* To avoid reading from the queue before the servers have had a
chance to enqueue data, check the queue counter to see whether the
queue is empty. */
mp_lock_set (queue_lock);
/* If the queue is empty, wait on server_done for a server. Otherwise,
just proceed. Clear the lock in any case. */
if (mp_counter_fetch (queued_items) <= 0)
mp_lock_wait (queue_lock, server_done);
mp_lock_clr (queue_lock);
/* Retrieve information from queue_info structure */
local_queue_index = queue_pointers[local_pointers_index].queue_base;
num_elements = queue_pointers[local_pointers_index].num_elem;
/* Remove and process items on the queue. For this example,
process_item simply prints out the value retrieved. In a real
application, it would probably read data into a local array, in
preparation for further processing after the locks are released. */
while (num_elements--)
{
process_item (queue_data[local_queue_index]);
if (++local_queue_index >= QUEUE_DATA_SPACE)
local_queue_index = 0;
}
/* Decrement counter because an element has been removed from the
queue */
mp_lock_set (queue_lock);
if ((ret = mp_counter_decr (queued_items)) < 0)
fprintf (stderr, "Client %d: queue pointers underflowed\n", mp_tid);
mp_lock_clr (queue_lock);
/* Clear lock so other clients can read data, and wake up one waiting
server thread, if any are waiting for data */
mp_lock_clr (client_lock);
mp_signal (client_done);
}
}
/* Runs as first function on each thread, so that the thread can later
generate random numbers */
void init_rand (seed)
int seed;
{
srand (seed);
}
void process_item (dequ)
float dequ;
{
printf ("Dequeued item is %f\n", dequ);
}
/* Sends TIMES signals to calling thread at INTERVAL seconds. */
void test_servers (thread_id)
int thread_id;
{
int times = TIMES;
while (times--)
{
t_kill (thread_id, TEST_SIGNAL);
sleep (INTERVAL);
}
mp_exit (0);
}
ＭＡＣＨに実装されている同期プリミティブは以上の２種類ですが、ＲＴＵではその他の同期モデルをプログラムしやすいように、代表的なモデルに合わせて同期プリミティブを用意しています。
これは、ＯＳの教科書で必ず取り上げられるもので、排他制御変数を使えば実現できるものですが、注意深くプログラムしないとデッドロックの危険があります。そこで、信頼性を上げるためにＲＴＵでは、プリミティブとして用意しました。
┌──────────────────────────────────────────────────┐
│ mpcounter mp_counter_alloc(); 動的にカウンタセマフォを割り当て０に初期化する。 │
│ void mp_counter_free(counter) カウンタセマフォをメモリに解放する。 │
│ int mp_counter_init(counter,value); カウンタセマフォを value に初期化する。 │
│ int mp_counter_incr(counter); カウンタセマフォをインクリメントする。 │
│ int mp_counter_decr(counter); カウンタセマフォをデクリメントする。 │
│ int mp_counter_fetch(counter); カウンタセマフォの値を読み出す │
│ mpcounter counter; │
│ int value; │
└──────────────────────────────────────────────────┘
このカウンタセマフォを使用すると、ダブルバッファリングやバリア同期のようなモデルを記述できます。
バリア同期モデルは、下図のように最も単純なスレッドの同期モデルです。
(1) a = sin(x); ◎
(2) b = cos(y) │
(3) c = tan(x+y); (1) (2) (3)
(4) X = (a+b)*c │
(5) Y = (a-b)/c ◎バリア
│
(4) (5)
│
◎バリア
上図のデータフローグラフからも理解できるようにデータ参照関係より［(1)(2)(3)］は並列に
実行できますが［(4)(5)］は［(1)(2)(3)］が終了した後でないと実行できません。
このような、ときにバリア同期を使用すれば簡単に同期をとることができます。
これも非常に有名なプログラミングモデルで、特にデータベースの更新などによく使われます。
このモデルでは、書き込みは常に排他制御されますが、読みだしは並列（あるいは並行）に行われます。
┌──────────────────────────────────────────────────┐
│ mprwlock mp_rwlock_alloc(); 排他書き込み多重読みだし型のセマフォ変数を割り当てる │
│ void mp_rwlock_free(key); 排他書き込み多重読みだし型のセマフォ変数を解放する │
│ void mp_read_set(key); 読み出し権を得る。書き出しスレッドがいる場合封鎖される│
│ int mp_read_try(key); mp_read_set と同様だが獲得に失敗しても封鎖されない │
│ void mp_read_clr(key); 読み出し権を放棄する。この操作により書き出しスレッドの│
│ 封鎖が解除される │
│ void mp_read_conv(key); この操作により現在持っている読み出し権を書き込み権に │
│ 変更する。もし読みだしスレッドが存在すれば封鎖される。│
│ void mp_write_set(key); 書き込み権を得る。もし読みだしと書き出しスレッドがいる│
│ 場合には封鎖される。 │
│ int mp_write_try(key); mp_write_set と同様だが獲得に失敗した場合でも封鎖され │
│ void mp_write_clr(key); 書き出し権を放棄する。この操作により読み込みスレッドの│
│ 封鎖が解除される │
│ void mp_write_conv(key); この操作により現在もっている書き込み権を読みだし権に │
│ 変更する。もし書き出しを待っているスレッドが存在する │
│ 場合は封鎖される │
│ mprwlock key; │
└──────────────────────────────────────────────────┘
ＣとＦＯＲＴＲＡＮの言語間インターフェースの詳細は、FORTRAN Language and Environment Manual の Chapter 11 Using FORTRAN with C に記述があります。
ｆ７７で使用できるプロセス制御関数を以下に示します。
ABORT シグナル SIGIOT("abort")を呼び出したプロセスに送信する。
ALARM 指定された間隔の後に SIGALARM シグナルを送信するようにアラームクロックをセットす る。
FORK 現プロセスをコピーして新しいプロセスを作成する。
KILL シグナルをプロセスに送信する。
PLOCKIN 仮想空間中のメモリセグメントを主記憶にロックする。
PUNLOCK メモリセグメントのロックを解除する。
SIGNAL KILL で送信されるシグナルの処理を定義する。
SLEEP プロセスの実行を中断する。
WAIT 子プロセスの終了を待つ。
これ以外のシステム制御サブルーチンは、ユーザがｆ７７で呼び出せるＣの関数を自作する必要がありますが、以下の手順で簡単に作成する事が出来ます。
１）変数名
ｆ７７から見るＣの変数は最後にアンダースコア"_"を付ける
FORTRAN val
C val_
２）サブルーチン呼びだし
ｆ７７からＣへの呼びだしは、 名前呼びだし（call by name）である。
FORTRAN integer*4 val1,val2,val3
call c_function(val1,val2,val3)
C void c_function_(val1,val2,val3)
int *val1,*val2,*val3;{};
３）変数の対応
FORTRAN C
BYTE or LOGICAL char, unsigned char
INTEGER*2 or LOGICAL*2 short int , unsigned short int
INTEGER*4 or LOGICAL*4 long int , unsigned long int
REAL*4 float
REAL*8 double
COMPLEX*8 struct {float r,i}
COMPLEX*16 struct {double r,i};
文字 int *p
COMMON /xyz/ struct xyz_
ＦＯＲＴＲＡＮ−Ｃ−ＦＯＲＴＲＡＮの言語間リンクを行うプログラム例
コンパイル方法
% f77 -ashard_ -o main main.f subc.c subf.f
####################################################
c FORTRAN main program
common /shared/ rdata
rdata = 10.0
iflg = 0
call subc( iflg )
call subc( iflg )
call subc( iflg )
end
#####################################################
#include
subc_( iflg )
int *iflg ;
{
extern struct{
float rdata ;
} shared_;
(*iflg )++ ;
shared_.rdata *= 2.0 ;
printf( "subc %d %f \n", *iflg, shared_.rdata ) ;
subf_( iflg ) ;
}
###########################################################
c FORTRAN sub program
subroutine subf( iflg )
common /shared/ rdata
iflg = iflg + 1
rdata = rdata * 2.0
print *,"subf ", iflg, rdata
end
この章では、いろいろな問題を複数のプログラム方法で掲載します。
共有メモリに２つの変数カウントおよびフラグを設定する。
１秒ごとにカウントに１を加え、カウント＞＝１０でフラグに１を代入して終了するプロセスＢと、
カウントに１が加わるごとにその値を表示しフラグ＝＝１で終了するプロセスＡを
作成する。
┌────────────────────────────────────────┐
プロセスＡ │ write write write │
└────────────────────────────────────────┘
↓ ↓ ↑ ↑ ↑
カウント←０ 起動 countup countup countup
フラグ ←０ ↓ ↑ ↑ ↑
┌────┬────┬──────────┬────┐
プロセスＢ │←１秒→│←１秒→│──────────│←１秒→│
└────┴────┴──────────┴────┘
実行権を放棄してプロセスＡにゆずる カウント≧１０で
フラグ←１して終了
プロセスＡは、常にカウントとフラグを監視
優先順位 Ｂ＞Ａ
高 低
＜＜ 方法１ ＞＞
このプログラムは、ＵＮＩＸ互換のシステムコールのみで記述します。
従って、共有メモリは SVR4 対応のシェアードメモリを使用して記述し、固定優先度プライオリティには、ｎｉｃｅシステムコールを使用します。
またこの例での共有メモリは、スタティックに宣言した領域を使用しています。
#include
#include
#include
#include
#include
#include
#include
/* protections on ipc stuff */
#define RUSER 0400
#define WUSER 0200
struct {
unsigned int loopflag,
counter;
} shared_memory ;
process_B(upcount)
unsigned int upcount;
{
nice(-20); /* 最大優先度のリアルタイムプライオリテリィにする*/
do {
sleep(1);
(shared_memory.counter)++;
} while ( shared_memory.counter < upcount );
shared_memory.loopflag = 1;
}
main()
{
int pid,count;
int shmid;
char *shmat();
int i,status;
char *shmaddr;
if((shmid = shmget(IPC_PRIVATE,sizeof(shared_memory),RUSER | WUSER )) < 0 ) {
perror("shmmget error");
exit(1);
}
if ((shmaddr = shmat(shmid,&shared_memory,SHM_RND)) == (char *) -1 ){
perror("shmat error");
shmdt(shmaddr);
shmctl (shmid, IPC_RMID, 0);
exit(1);
}
shared_memory.loopflag = 0;
shared_memory.counter = count = 0;
nice(-19); /* 最大優先度-1のリアルタイムプライオリテリィにする*/
if (plock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
if((pid = fork())==NULL) {
process_B(10);
exit(1);
} else {
do {
if ( shared_memory.counter != count ) {
count = shared_memory.counter;
printf(" update counter %d\n",shared_memory.counter);
}
} while ((shared_memory.loopflag)==0);
while ( pid==wait(&status));
shmdt(shmaddr);
shmctl (shmid, IPC_RMID, 0);
fflush(stdout);
}
}
＜＜ 方法２ ＞＞
このプログラムは、SVR4 対応のシェアードメモリを使用し、ＲＴＵのリアルタイム・スケジュリング、メモリロック、astpause等を使用した方法で記述します。
またこの例での共有メモリは、データセグメントを伸張させてダイナミックに割り付けています。
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define RT_CPU_MASK 4
#define roundup(x) (((int)(x) + NBPG -1)& ~(NBPG-1))
#define MAP_SIZE 1*NBPG
/* protections on ipc stuff */
#define RUSER 0400
#define WUSER 0200
struct shared_memory {
unsigned int loopflag,
counter;
};
struct shared_memory *shmaddr = NULL;
process_B(upcount)
unsigned int upcount;
{
int policy, prio;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI; /* 最大優先度のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
do {
astpause(-1,1000);/* １０００ミリ秒、優先度を変化させずに待つ */
(shmaddr->counter)++;
} while ( shmaddr->counter < upcount );
shmaddr->loopflag = 1;
}
main()
{
int policy, prio,pid,count;
int shmid;
char *shmat();
int i,status;
unsigned char *sbrk();
shmaddr = (struct shared_memory*)sbrk(0);
shmaddr = (struct shared_memory*)roundup(shmaddr);
status = brk(shmaddr + MAP_SIZE);
if(status<0){
perror(stderr,"Memory Alocation");
exit(1);
}
if((shmid = shmget(IPC_PRIVATE,sizeof(struct shared_memory),RUSER | WUSER )) < 0 ) {
perror("shmmget error");
exit(1);
}
if (shmat(shmid,shmaddr,SHM_RND) == (char *) -1 ){
perror("shmat error");
shmdt(shmaddr);
shmctl (shmid, IPC_RMID, 0);
exit(1);
}
shmaddr->loopflag = 0;
shmaddr->counter = count = 0;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI-1; /* 最大優先度-1のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
#ifdef SLAVE
if (mpadvise(MPA_RT_SET, RT_CPU_MASK) < 0) {/* セカンドプロセッサの占有をセットする */
perror("mpadvise reserving processor");
exit (1);
}
if (mpadvise(MPA_60HZ_OFF, RT_CPU_MASK) < 0) {/* unix のスケジューリング割り込みをキャンセルする */
perror("mpadvise turning off time-sharing interrupts");
exit (3);
}
if (mpadvise(MPA_TBINV_OFF, RT_CPU_MASK) < 0) {
perror ("MPA_TBINV_OFF");
exit (7);
}
#endif
#ifdef LOCK
if (rtplock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
#endif
if((pid = fork())==NULL) {
process_B(10);
} else {
do {
if ( shmaddr->counter != count ) {
count = shmaddr->counter;
printf(" update counter %d\n",shmaddr->counter);
}
} while ((shmaddr->loopflag)==0);
while ( pid==wait(&status));
shmdt(shmaddr);
shmctl (shmid, IPC_RMID, 0);
fflush(stdout);
}
}
＜＜ 方法３ ＞＞
このプログラムは、RTU におけるリザーブドメモリを使用した方法で記述します。
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define RT_CPU_MASK 4
#define roundup(x) (((int)(x) + NBPG -1)& ~(NBPG-1))
#define MAP_SIZE 1*NBPG
struct shared_memory {
unsigned int loopflag,
counter;
};
struct shared_memory *shmaddr = NULL;
process_B(upcount)
unsigned int upcount;
{
int policy, prio;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI; /* 最大優先度のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
do {
astpause(-1,1000);
(shmaddr->counter)++;
} while ( shmaddr->counter < upcount );
shmaddr->loopflag = 1;
}
main()
{
int policy, prio,pid,count;
int memfd;
int status;
unsigned char *sbrk();
shmaddr = (struct shared_memory*)sbrk(0);
shmaddr = (struct shared_memory*)roundup(shmaddr);
status = brk(shmaddr + MAP_SIZE);
if(status<0){
perror(stderr,"Memory Alocation");
exit(1);
}
if ((memfd=open("/dev/resmem",O_RDWR))==-1) {
perror("reserved memory open failed");
exit(1);
}
status=mmap(shmaddr, MAP_SIZE, PROT_READ|PROT_WRITE, MAP_SHARED, memfd, 0);
if (status < 0 ) {
perror("mmap");
exit(2);
}
shmaddr->loopflag = 0;
shmaddr->counter = count = 0;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI-1; /* 最大優先度-1のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
#ifdef SLAVE
if (mpadvise(MPA_RT_SET, RT_CPU_MASK) < 0) {/* セカンドプロセッサの占有をセットする */
perror("mpadvise reserving processor");
exit (1);
}
if (mpadvise(MPA_60HZ_OFF, RT_CPU_MASK) < 0) {/* unix のスケジューリング割り込みをキャンセルする */
perror("mpadvise turning off time-sharing interrupts");
exit (3);
}
if (mpadvise(MPA_TBINV_OFF, RT_CPU_MASK) < 0) {
perror ("MPA_TBINV_OFF");
exit (7);
}
#endif
#ifdef LOCK
if (rtplock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
#endif
if((pid = fork())==NULL) {
process_B(10);
} else {
do {
if ( shmaddr->counter != count ) {
count = shmaddr->counter;
printf(" update counter %d\n",shmaddr->counter);
}
} while ((shmaddr->loopflag)==0);
while ( pid==wait(&status));
fflush(stdout);
}
}
＜＜ 方法４ ＞＞
このプログラムは、スレッドを使用し、リアルタイムスレーブを使用した方法で記述します。
スレッドは、同一プロセッサに割り付けています。
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define RT_CPU_MASK 4
unsigned int loopflag = 0,
count = 0,counter = 0;
process_B(upcount)
unsigned int upcount;
{
int policy, prio;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI; /* 最大優先度のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
do {
astpause(-1,1000);
counter++;
} while ( counter < upcount );
loopflag = 1;
t_exit(0);
}
main()
{
int policy, prio,tid;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI-1; /* 最大優先度-1のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
#ifdef SLAVE
if (mpadvise(MPA_RT_SET, RT_CPU_MASK) < 0) {/* セカンドプロセッサの占有をセットする */
perror("mpadvise reserving processor");
exit (1);
}
if (mpadvise(MPA_60HZ_OFF, RT_CPU_MASK) < 0) {/* unix のスケジューリング割り込みをキャンセルする */
perror("mpadvise turning off time-sharing interrupts");
exit (3);
}
if (mpadvise(MPA_TBINV_OFF, RT_CPU_MASK) < 0) {
perror ("MPA_TBINV_OFF");
exit (7);
}
#endif
#ifdef LOCK
if (rtplock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
#endif
tid = mp_create(process_B,10);
do {
if ( counter != count ) {
count = counter;
printf(" update counter %d\n",counter);
}
} while (loopflag==0);
t_wait(tid);
fflush(stdout);
}
共有メモリに変数カウントおよびフラグを設定する。
１ミリ秒毎に起動されカウントに１を加え、カウント＞＝１００００でフラグに１を代入して終了するプロセスＢとカウントが１０００で終了するプロセスＡを作成する。
┌────────────────────────────────────────┐
プロセスＡ │ カウントが１０００で割り切れたら write │
└────────────────────────────────────────┘
↓ ↓ ↑ ↑ ↑
カウント←０ 起動 countup countup countup
フラグ ←０ ↓ ↑ ↑ ↑
┌────┐ ┌────┐ ┌────┐
プロセスＢ │ │suspend │ │suspend │←1ms →│
└────┘ └────┘ └────┘
↑ ←1ms→ ↑ ←1ms → ↑
クロック割込み
実行権を放棄してプロセスＡにゆずる カウント≧１００００で
フラグ←１して終了
プロセスＡは、常にカウントとフラグを監視
優先順位 Ｂ＞Ａ
高 低
＜＜ 方法１ ＞＞
スレッドを２つ生成し、クロックＡＳＴを使用する。
この例では使用していないが、２つのスレッドを別のプロセッサに割り付けられる
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define RT_CPU_MASK 4
int cpu_no = 2;
unsigned int loopflag = 0,
count = 0,counter = 0;
#define CLK_TICK 1000 /* マイナーサイクルは1KHz */
#define CLK_HZ 1000 /* 1msで定周期割り込み */
#define CLK_PRI 100 /* AST プライオリティ */
#define GIGA (1000*1000*1000)
#define NANO_CYCLE (GIGA/CLK_HZ)
/* グローバル変数 */
struct clkspec_t interv = {0, NANO_CYCLE, 0, 0};
int ast_id;
function1();
(*cyclic[1])() = { function1 };
extern _asthandler();
/* １ミリ秒ごとにこのスレッドがセカンドプロセッサで起動される*/
function1(overrun)
{
counter ++;
if ((counter%10000)==0) {
loopflag = 1;
}
}
process_B(upcount)
unsigned int upcount;
{
int policy, prio;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI; /* 最大優先度のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
if ( (ast_id = clkast(_asthandler, cyclic[0], CLK_PRI, &interv, cpu_no))== -1) {
perror("Failed to start clock");
exit(1);
}
do{ /* ここで 永久にループして割り込みを待つ */
if ( astpause(0,2000) < 0 ){
perror("ast pause");exit(1);
}
} while(loopflag==0);
t_exit(0);
}
main()
{
int policy, prio,tid;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI-1; /* 最大優先度-1のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
if (mpadvise(MPA_RT_SET, RT_CPU_MASK) < 0) {/* セカンドプロセッサの占有をセットする */
perror("mpadvise reserving processor");
exit (1);
}
if (mpadvise(MPA_60HZ_OFF, RT_CPU_MASK) < 0) {/* unix のスケジューリング割り込みをキャンセルする */
perror("mpadvise turning off time-sharing interrupts");
exit (3);
}
/* 定周期割り込みの周波数をセットする */
if ((mpadvise(MPA_CLK_SET,RT_CPU_MASK, CLK_TICK)) < 0) {
perror("mpadvise setting clock interrupt");
exit (2);
}
if (rtplock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
tid = mp_create(process_B,10000);
do {
if ( (counter%1000)==0 ){
printf("counter %d\n",counter);
}
} while (loopflag==0);
t_wait(tid);
fflush(stdout);
}
＜＜ 方法２ ＞＞
スレッドを使用せず、プロセスと割込み関数のみで記述する。
この例では、ＦａｓｔクロックＡＳＴを使用している。
割込み関数は、割り込まれた関数よりも優先度が高い。
#include
#include
#include
#include
#include
#define CLK_TICK 1000 /* マイナーサイクルは1KHz */
#define CLK_HZ 1000 /* 1000 Hz(1ms)で定周期割り込み */
#define CLK_PRI 100 /* AST プライオリティ */
#define GIGA (1000*1000*1000)
#define NANO_CYCLE (GIGA/CLK_HZ)
#define RT_CPU_MASK 0x4
/* グローバル変数 */
struct clkspec_t interv = {0, NANO_CYCLE, 0, 0};
int ast_id;
function1();
(*cyclic[1])() = { function1};
extern _fasthandler();
int policy = SCHED_FIFO; /* スケジューリングはＦＩＦＯでおこなう */
int prio = MAX_RT_PRI; /* プライオリティは最大プライオリティ */
int loopflag = 0; /* この値を1にするとプロセスが終了する */
int counter = 0;
/* １ミリ秒ごとにこのスレッドが起動される*/
function1(overrun)
{
counter ++;
if ((counter%10000)==0) {
loopflag = 1;
}
}
main()
{
/* FIFO スケジューリングにする 最大優先度のリアルタイムプライオリテリィにする */
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
/* カレントプロセスをロックする */
if (rtplock(PROCLOCK) < 0) {
perror("process lock");
exit(2);
}
/* プロセッサの占有をセットする */
if (mpadvise(MPA_RT_SET, RT_CPU_MASK) < 0) {
perror("mpadvise reserving processor");
exit (1);
}
/* 定周期割り込みの周波数をセットする */
if ((mpadvise(MPA_CLK_SET,RT_CPU_MASK, CLK_TICK)) < 0) {
perror("mpadvise setting clock interrupt");
exit (2);
}
/* unix のスケジューリング割り込みをキャンセルする */
if (mpadvise(MPA_60HZ_OFF, RT_CPU_MASK) < 0) {
perror("mpadvise turning off time-sharing interrupts");
exit (3);
}
/* farst ast ハンドラとcyclic 関数を登録し割り込みで定周期起動がかかるようにする */
if ( (ast_id = fclkast(_fasthandler, cyclic[0], CLK_PRI,
&interv, 2,0))== -1) {
perror("Failed to start clock");
exit(1);
}
do{ /* ここでループして割り込みを待つ */
if ( (counter%1000)==0 ){
printf("counter %d\n",counter);
}
} while(loopflag==0);
}
８．１と同一仕様で、プロセスＡはＦＯＲＴＲＡＮで、プロセスＢは、Ｃで作成する。
以下のプログラムを ex3.f.f ex3.c.c をコンパイル時に -a オプションを指定して、言語間で共有メモリを実現します。この場合、ｆ７７のcommon文で宣言された領域をＣ言語でシェアードメモリシステムコールを使用し共有メモリにしているのがポイントです。
% f77 -o ex3 ex3f.f ex3c.c -ashared_data_
c***********************
c* fortran main ex3f.f *
c***********************
c
integer*4 loopflag,counter
common /shared_memory/ loopflag,counter
integer*4 pid,fork,wait,status,upcount,shmid,shmaddr
c
upcount = 10
call map_memory(shmid,shmaddr)
loopflag = 0
counter = 0
count = 0
call realtime()
pid = fork()
if (pid.eq.0) then
call process_b(upcount)
call exit()
else
do while (loopflag.eq.0)
if (counter.ne.count) then
count = counter
print *,"update counter",counter
end if
end do
status = wait(status)
call unmap_memory(shmid,shmaddr)
end if
end
***********************
* subroutine C ex3c.c *
***********************
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define RT_CPU_MASK 4
#define roundup(x) (((int)(x) + NBPG -1)& ~(NBPG-1))
#define MAP_SIZE 1*NBPG
/* protections on ipc stuff */
#define RUSER 0400
#define WUSER 0200
extern struct {
unsigned int loopflag,
counter;
} shared_memory_;
process_b_(upcount)
int *upcount;
{
int policy, prio;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI; /* 最大優先度のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
do {
astpause(-1,1000);
(shared_memory_.counter)++;
} while ( shared_memory_.counter < *upcount );
shared_memory_.loopflag = 1;
}
map_memory_(shmid,shmaddr)
int *shmid;
char **shmaddr;
{
char *shmat();
if((*shmid = shmget(IPC_PRIVATE,sizeof(shared_memory_),
RUSER | WUSER )) < 0 ) {
perror("shmmget error");
exit(1);
}
if ((*shmaddr = shmat(*shmid,&shared_memory_,SHM_RND)) == (char *) -1 ){
perror("shmat error");
shmdt(*shmaddr);
shmctl (*shmid, IPC_RMID, 0);
exit(1);
}
}
realtime_(){
int i,status;
int policy, prio,pid;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI-1; /* 最大優先度-1のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
#ifdef SLAVE
if (mpadvise(MPA_RT_SET, RT_CPU_MASK) < 0) {/* セカンドプロセッサの占有をセットする */
perror("mpadvise reserving processor");
exit (1);
}
if (mpadvise(MPA_60HZ_OFF, RT_CPU_MASK) < 0) {/* unix のスケジューリング割り込みをキャンセルする */
perror("mpadvise turning off time-sharing interrupts");
exit (3);
}
#endif
#ifdef LOCK
if (rtplock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
#endif
}
unmap_memory_(shmid,shmaddr)
int *shmid;
int **shmaddr;
{
char *shmat();
shmdt(shmaddr);
shmctl (shmid, IPC_RMID, 0);
fflush(stdout);
}
以下の２つのプロセスを作成する。ただし優先度は（低）Ａ＜Ｂ（高）とする。
プロセスＡ
共有メモリのカウントおよびフラグを０に初期化後、プロセスＢを起動し、フラグ＝１を確認し１秒後にプロセスＢを再開する。
その後、カウント値を表示、１秒後プロセスＢを再開する。処理をフラグ＝２まで繰り返す。
プロセスＢ
プロセスＡより起動、フラグに１を代入して停止、その後タスクＡから再開される毎にカウントに１を加え停止する処理をカウント≧１０になるまで繰り返し、フラグに２を代入して終了する。
１秒 １秒 １秒 １秒
┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐
プロセスＡ │ │ │ │ │ │ │ │フラグ＝２で終了
└─────┘ └↓────┘ └↓────┘ └↓────┘
カウント←０ ↓ フラグ＝１ ↓ write ↓ write
フラグ ←０ ↓ 確認 ↓ resume ↓ resume
┌─┐ ┌─┐ ┌─┐
│ │suspend │ │suspend │ │suspend カウント≧１０で
└─┘ └─┘ └─┘ フラグ←２して終了
フラグ←１ カウントアップ
＜＜ 方法１ ＞＞
このプログラムは、SVR4 対応のシェアードメモリとシステムコールを使用した方法で記述します。
#include
#include
#include
#include
#include
#include
#include
/* protections on ipc stuff */
#define RUSER 0400
#define WUSER 0200
struct {
unsigned int loopflag,
counter;
} shared_memory ;
int (*syshandler)(); /* Pointer to old handler. */
function(sig)
int sig;{}
process_B(upcount)
unsigned int upcount;
{
nice(-20); /* 最大優先度のリアルタイムプライオリテリィにする*/
shared_memory.loopflag = 1;
do {
pause();
signal(SIGUSR1,function);
(shared_memory.counter)++;
} while ( shared_memory.counter < upcount );
shared_memory.loopflag = 2;
}
main()
{
int pid,count;
int shmid;
char *shmat();
int i,status;
char *shmaddr;
if((shmid = shmget(IPC_PRIVATE,sizeof(shared_memory),RUSER | WUSER )) < 0 ) {
perror("shmmget error");
exit(1);
}
if ((shmaddr = shmat(shmid,&shared_memory,SHM_RND)) == (char *) -1 ){
perror("shmat error");
shmdt(shmaddr);
shmctl (shmid, IPC_RMID, 0);
exit(1);
}
shared_memory.loopflag = 0;
shared_memory.counter = count = 0;
nice(-19); /* 最大優先度-1のリアルタイムプライオリテリィにする*/
#ifdef LOCK
if (plock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
#endif
if((pid = fork())==NULL) {
syshandler = signal(SIGUSR1,function);
process_B(10);
signal(SIGUSR1,syshandler);
exit(1);
} else {
do {
} while ((shared_memory.loopflag)==0);
do {
sleep(1);
kill(pid,SIGUSR1);
if ( shared_memory.counter != count ) {
count = shared_memory.counter;
printf(" update counter %d\n",shared_memory.counter);
}
} while ((shared_memory.loopflag)==1);
while ( pid==wait(&status));
shmdt(shmaddr);
shmctl (shmid, IPC_RMID, 0);
fflush(stdout);
}
}
＜＜ 方法２ ＞＞
このプログラムは、SVR4 対応のシェアードメモリとＲＴＵのリアルタイムプロセス・スケジュール・システムコールを使用した方法で記述します。
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
/* protections on ipc stuff */
#define RUSER 0400
#define WUSER 0200
struct {
unsigned int loopflag,
counter;
} shared_memory ;
process_B(upcount)
unsigned int upcount;
{
int policy, prio;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI; /* 最大優先度のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
shared_memory.loopflag = 1;
do {
hold();/* プロセスをリアルタイムスケジューリングから外す */
(shared_memory.counter)++;
} while ( shared_memory.counter < upcount );
shared_memory.loopflag = 2;
}
main()
{
int pid,count;
int shmid;
char *shmat();
int i,status;
char *shmaddr;
int policy, prio;
if((shmid = shmget(IPC_PRIVATE,sizeof(shared_memory),RUSER | WUSER )) < 0 ) {
perror("shmmget error");
exit(1);
}
if ((shmaddr = shmat(shmid,&shared_memory,SHM_RND)) == (char *) -1 ){
perror("shmat error");
shmdt(shmaddr);
shmctl (shmid, IPC_RMID, 0);
exit(1);
}
shared_memory.loopflag = 0;
shared_memory.counter = count = 0;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI-1; /* 最大優先度-1のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
#ifdef SLAVE
if (mpadvise(MPA_RT_SET, RT_CPU_MASK) < 0) {/* セカンドプロセッサの占有をセットする */
perror("mpadvise reserving processor");
exit (1);
}
if (mpadvise(MPA_60HZ_OFF, RT_CPU_MASK) < 0) {/* unix のスケジューリング割り込みをキャンセルする */
perror("mpadvise turning off time-sharing interrupts");
exit (3);
}
if (mpadvise(MPA_TBINV_OFF, RT_CPU_MASK) < 0) {
perror ("MPA_TBINV_OFF");
exit (7);
}
#endif
#ifdef LOCK
if (rtplock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
#endif
if((pid = fork())==NULL) {
process_B(10);
exit(1);
} else {
while ((shared_memory.loopflag)==0);
do {
astpause(-1,1000);
release(pid);/* hold しているプロセスをリアルタイムスケジュールキューに戻す */
if ( shared_memory.counter != count ) {
count = shared_memory.counter;
printf(" update counter %d\n",shared_memory.counter);
}
} while ((shared_memory.loopflag)==1);
while ( pid==wait(&status));
shmdt(shmaddr);
shmctl (shmid, IPC_RMID, 0);
fflush(stdout);
}
}
＜＜ 方法３ ＞＞
このプログラムは、スレッドを使用した方法で記述します。
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
struct {
unsigned int loopflag,
counter;
} shared_memory ;
unsigned int condition = 0;
process_B(upcount)
unsigned int upcount;
{
int policy, prio;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI; /* 最大優先度のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
shared_memory.loopflag = 1;
do {
t_cwant(&condition);
t_csleep(&condition);
(shared_memory.counter)++;
} while ( shared_memory.counter < upcount );
shared_memory.loopflag = 2;
t_exit();
}
main()
{
int tid,count;
int i,status;
int policy, prio;
shared_memory.loopflag = 0;
shared_memory.counter = count = 0;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI-1; /* 最大優先度-1のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
#ifdef SLAVE
if (mpadvise(MPA_RT_SET, RT_CPU_MASK) < 0) {/* セカンドプロセッサの占有をセットする */
perror("mpadvise reserving processor");
exit (1);
}
if (mpadvise(MPA_60HZ_OFF, RT_CPU_MASK) < 0) {/* unix のスケジューリング割り込みをキャンセルする */
perror("mpadvise turning off time-sharing interrupts");
exit (3);
}
if (mpadvise(MPA_TBINV_OFF, RT_CPU_MASK) < 0) {
perror ("MPA_TBINV_OFF");
exit (7);
}
#endif
#ifdef LOCK
if (rtplock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
#endif
tid = mp_create(process_B,10);
while ((shared_memory.loopflag)==0);
do {
astpause(-1,1000);
t_twakeup(tid);
if ( shared_memory.counter != count ) {
count = shared_memory.counter;
printf(" update counter %d\n",shared_memory.counter);
}
} while ((shared_memory.loopflag)==1);
t_wait(tid);
}
８．４ の仕様のプロセスＡで、resume 直前の時刻 t1 を読む、同様にプロセスＢで suspend 直後の時刻 t2 を読み、 t2-t1 の値を表示する。（t2-t1 が resume のオーバヘッド）
コンパイルと実行結果
% cc -O3 -o ex5.0 ex5.0.c
% cc -O3 -o ex5.1 ex5.1.c
% cc -O3 -o ex5.2 ex5.2.c
% ex5.0
update counter 1
update counter 2
update counter 3
update counter 4
update counter 5
update counter 6
update counter 7
update counter 8
update counter 9
prosess switch time 622.272 micro sec
% ex5.1
update counter 1
update counter 2
update counter 3
update counter 4
update counter 5
update counter 6
update counter 7
update counter 8
update counter 9
prosess switch time 194.56 micro sec
% ex5.2
update counter 1
update counter 2
update counter 3
update counter 4
update counter 5
update counter 6
update counter 7
update counter 8
update counter 9
prosess switch time 191.936 micro sec
＜＜ 方法１ ＞＞
/* ex5.0.c */
#include
#include
#include
#include
#include
#include
#include
/* protections on ipc stuff */
#define RUSER 0400
#define WUSER 0200
#define LOOPNO 10
struct {
unsigned int loopflag,
counter;
unsigned long int t1[LOOPNO],t2[LOOPNO];
} shared_memory ;
adjust(start,finish,realtime)
unsigned long *start,*finish;
float *realtime;
{
#ifdef mc7000
*realtime = (float)( *finish - *start ) * 640.0 / 1000.0;
#endif
}
readtime(nowtime)
unsigned long *nowtime;
{
#ifdef mc7000
#define HIGH_TIME_ADDRESS 0xfe001000 /* 48 bit realtime counter high byte */
#define LOW_TIME_ADDRESS 0xfe001004 /* 48 bit realtime counter low word */
*nowtime = *(unsigned long *)LOW_TIME_ADDRESS;
#endif
}
int (*syshandler)(); /* Pointer to old handler. */
function(sig)
int sig;
{}
process_B(upcount)
unsigned int upcount;
{
int policy, prio;
nice(-20); /* 最大優先度のリアルタイムプライオリテリィにする*/
shared_memory.loopflag = 1;
do {
pause();
readtime(&(shared_memory.t2[shared_memory.counter]));
signal(SIGUSR1,function);
(shared_memory.counter)++;
} while ( shared_memory.counter < upcount );
shared_memory.loopflag = 2;
}
main()
{
int pid,count;
int shmid;
char *shmat();
int i,status;
char *shmaddr;
int policy, prio;
float realtime=0.0,total = 0.0;
if((shmid = shmget(IPC_PRIVATE,sizeof(shared_memory),RUSER | WUSER )) < 0 ) {
perror("shmmget error");
exit(1);
}
if ((shmaddr = shmat(shmid,&shared_memory,SHM_RND)) == (char *) -1 ){
perror("shmat error");
shmdt(shmaddr);
shmctl (shmid, IPC_RMID, 0);
exit(1);
}
shared_memory.loopflag = 0;
shared_memory.counter = count = 0;
nice(-19); /* 最大優先度-1のリアルタイムプライオリテリィにする*/
#ifdef LOCK
if (plock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
#endif
if((pid = fork())==NULL) {
syshandler = signal(SIGUSR1,function);
process_B(LOOPNO);
signal(SIGUSR1,syshandler);
exit(1);
} else {
do {
} while ((shared_memory.loopflag)==0);
do {
sleep(1);
readtime(&(shared_memory.t1[shared_memory.counter]));
kill(pid,SIGUSR1);
if ( shared_memory.counter != count ) {
count = shared_memory.counter;
printf(" update counter %d\n",shared_memory.counter);
}
} while ((shared_memory.loopflag)==1);
while ( pid==wait(&status));
for (i=0;i
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
/* protections on ipc stuff */
#define RUSER 0400
#define WUSER 0200
#define RT_CPU_MASK 0x04
#define LOOPNO 10
struct {
unsigned int loopflag,
counter;
unsigned long int t1[LOOPNO],t2[LOOPNO];
} shared_memory ;
adjust(start,finish,realtime)
unsigned long *start,*finish;
float *realtime;
{
#ifdef mc7000
*realtime = (float)( *finish - *start ) * 640.0 / 1000.0;
#endif
#ifdef mc8000
*realtime = (float)( *finish - *start ) * 62.50 / 1000.0;
#endif
}
readtime(nowtime)
unsigned long *nowtime;
{
#ifdef mc7000
#define HIGH_TIME_ADDRESS 0xfe001000 /* 48 bit realtime counter high byte */
#define LOW_TIME_ADDRESS 0xfe001004 /* 48 bit realtime counter low word */
*nowtime = *(unsigned long *)LOW_TIME_ADDRESS;
#endif
#ifdef mc8000
unsigned long int *clk_ptr = (unsigned long int *)0x7FFFA000;
*nowtime = *((unsigned long int *)0x7FFFA000);
#endif
}
process_B(upcount)
unsigned int upcount;
{
int policy, prio;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI; /* 最大優先度のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
shared_memory.loopflag = 1;
do {
hold();
readtime(&(shared_memory.t2[shared_memory.counter]));
(shared_memory.counter)++;
} while ( shared_memory.counter < upcount );
shared_memory.loopflag = 2;
}
main()
{
int pid,count;
int shmid;
char *shmat();
int i,status;
char *shmaddr;
int policy, prio;
float realtime=0.0,total = 0.0;
if((shmid = shmget(IPC_PRIVATE,sizeof(shared_memory),RUSER | WUSER )) < 0 ) {
perror("shmmget error");
exit(1);
}
if ((shmaddr = shmat(shmid,&shared_memory,SHM_RND)) == (char *) -1 ){
perror("shmat error");
shmdt(shmaddr);
shmctl (shmid, IPC_RMID, 0);
exit(1);
}
shared_memory.loopflag = 0;
shared_memory.counter = count = 0;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI-1; /* 最大優先度-1のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
#ifdef SLAVE
if (mpadvise(MPA_RT_SET, RT_CPU_MASK) < 0) {/* セカンドプロセッサの占有をセットする */
perror("mpadvise reserving processor");
exit (1);
}
if (mpadvise(MPA_60HZ_OFF,RT_CPU_MASK)<0){/*unixのスケジューリング割り込みをキャンセルする*/
perror("mpadvise turning off time-sharing interrupts");
exit (3);
}
#endif
#ifdef LOCK
if (rtplock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
#endif
if((pid = fork())==NULL) {
process_B(LOOPNO);
exit(1);
} else {
do {
} while ((shared_memory.loopflag)==0);
do {
astpause(-1,1000);
readtime(&(shared_memory.t1[shared_memory.counter]));
release(pid);
if ( shared_memory.counter != count ) {
count = shared_memory.counter;
printf(" update counter %d\n",shared_memory.counter);
}
} while ((shared_memory.loopflag)==1);
while ( pid==wait(&status));
for (i=0;i
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define LOOPNO 10
struct {
unsigned int loopflag,
counter;
unsigned long int t1[LOOPNO],t2[LOOPNO];
} shared_memory ;
unsigned int condition = 0;
adjust(start,finish,realtime)
unsigned long *start,*finish;
float *realtime;
{
#ifdef mc7000
*realtime = (float)( *finish - *start ) * 640.0 / 1000.0;
#endif
#ifdef mc8000
*realtime = (float)( *finish - *start ) * 62.50 / 1000.0;
#endif
}
readtime(nowtime)
unsigned long *nowtime;
{
#ifdef mc7000
#define HIGH_TIME_ADDRESS 0xfe001000 /* 48 bit realtime counter high word */
#define LOW_TIME_ADDRESS 0xfe001004 /* 48 bit realtime counter low word */
*nowtime = *(unsigned long *)LOW_TIME_ADDRESS;
#endif
#ifdef mc8000
unsigned long int *clk_ptr = (unsigned long int *)0x7FFFA000;
*nowtime = *((unsigned long int *)0x7FFFA000);
#endif
}
process_B(upcount)
unsigned int upcount;
{
int policy, prio;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI; /* 最大優先度のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
shared_memory.loopflag = 1;
do {
t_cwant(&condition);
t_csleep(&condition);
readtime(&(shared_memory.t2[shared_memory.counter]));
(shared_memory.counter)++;
} while ( shared_memory.counter < upcount );
shared_memory.loopflag = 2;
t_exit();
}
main()
{
int tid,count;
int i,status;
int policy, prio;
float realtime=0.0,total = 0.0;
shared_memory.loopflag = 0;
shared_memory.counter = count = 0;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MAX_RT_PRI-1; /* 最大優先度-1のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
#ifdef SLAVE
if (mpadvise(MPA_RT_SET, RT_CPU_MASK) < 0) {/* セカンドプロセッサの占有をセットする */
perror("mpadvise reserving processor");
exit (1);
}
if (mpadvise(MPA_60HZ_OFF, RT_CPU_MASK) < 0) {/* unix のスケジューリング割り込みをキャンセルする */
perror("mpadvise turning off time-sharing interrupts");
exit (3);
}
if (mpadvise(MPA_TBINV_OFF, RT_CPU_MASK) < 0) {
perror ("MPA_TBINV_OFF");
exit (7);
}
#endif
#ifdef LOCK
if (rtplock(PROCLOCK) < 0) {/* カレントプロセスをロックする */
perror("process lock");
exit(2);
}
#endif
tid = mp_create(process_B,10);
while ((shared_memory.loopflag)==0);
do {
astpause(-1,1000);
readtime(&(shared_memory.t1[shared_memory.counter]));
t_twakeup(tid);
if ( shared_memory.counter != count ) {
count = shared_memory.counter;
printf(" update counter %d\n",shared_memory.counter);
}
} while ((shared_memory.loopflag)==1);
t_wait(tid);
for (i=0;i
共有メモリに flag1,falg2 を設定、以下にに示すプロセスＡ、Ｂ、Ｃをそれぞれ別プロセッサで実行する。
┌───────────┬────────────────┬─────────────┐
プロセスＡ │flag1←0 │flag1 == 1 かつ flag2 == 1 なら │flag1==3 かつflag2==3 なら│
│flag2←0│BとCを起動 │時刻t1取得 flag1 ← 2 flag ← 2 │時刻t2取得 t2-t1表示、終了│
└────┼──────┴────────────────┴─────────────┘
├────┬──────────────────┬───────┬────┐
プロセスＢ │flag1←1│ │flag1=2なら │flag1←3│
├────┴──────────────────┴───────┴────┘
│ data(i)=2.0*i 終了
│ i=1,10000000
├────┬──────────────────┬───────┬────┐
プロセスＣ │flag2←1│ │flag2=2なら │flag2←3│
└────┴──────────────────┴───────┴────┘
data(i)=2.0*i 終了
i=10000001,2000000
ここでは、同期変数に計数型排他制御変数（カウンタセマフォ）を使用します。
コンパイルと実行結果
% cc -O3 ex7.c
prosess execution time 1.7994e+06 micro sec
/* ex7.c */
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define DATA_SIZE 2000000
struct {
mpcounter flag1,flag2;
float data[DATA_SIZE];
} shared_memory ;
unsigned long int t1,t2;
unsigned int condition = 0;
adjust(start,finish,realtime)
unsigned long *start,*finish;
float *realtime;
{
#ifdef mc7000
*realtime = (float)( *finish - *start ) * 640.0 / 1000.0;
#endif
#ifdef mc8000
*realtime = (float)( *finish - *start ) * 62.50 / 1000.0;
#endif
}
readtime(nowtime)
unsigned long *nowtime;
{
#ifdef mc7000
#define HIGH_TIME_ADDRESS 0xfe001000 /* 48 bit realtime counter high word */
#define LOW_TIME_ADDRESS 0xfe001004 /* 48 bit realtime counter low word */
*nowtime = *(unsigned long *)LOW_TIME_ADDRESS;
#endif
#ifdef mc8000
unsigned long int *clk_ptr = (unsigned long int *)0x7FFFA000;
*nowtime = *((unsigned long int *)0x7FFFA000);
#endif
}
process_B(mask)
unsigned int mask;
{
int policy, prio,i;
policy = SCHED_FIFO;/* FIFO スケジューリングにする */
prio = MIN_RT_PRI; /* 最大優先度のリアルタイムプライオリテリィにする*/
if (schedset(NULL, &policy, &prio, NULL) < 0) {
perror("schedset");
exit(2);
}
if (mpadvise(MPA_RT_SET,mask) < 0) {/* プロセッサの占有をセットする */
perror("mpadvise reserving processor");
exit (1);
}
if (mpadvise(MPA_60HZ_OFF,mask) < 0) {
/* unix のスケジューリング割り込みをキャンセルする */
perror("mpadvise turning off time-sharing interrupts");
exit (3);
}
mp_counter_incr(shared_memory.flag1);
do { astpause(-1,10);}
while ( (mp_counter_fetch(shared_memory.flag1)!=2) ||
(mp_counter_fetch(shared_memory.flag2)!=2));
for ( i=0;i
このプログラム例は、スレッドによるダブルバッファ法を使用し、非同期入出力を記述しています。
繰返しループ 繰返しループ
┌───┐┌───┐┌───┐┌───┐┌───┐┌───┐┌───┐┌───┐┌───┐
入出力スレッド│read1 ││read2 ││write1││read1 ││write2││read2 ││write1││read1 ││write2│
└───┘└───┘└───┘└───┘└───┘└───┘└───┘└───┘└───┘
┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────
計算スレッド │ calc1 │ │ calc2 │ │ calc1 │ │ calc2 │ │
└─────┘ └─────┘ └─────┘ └─────┘ └─────
繰返しループ 繰返しループ
#include
#include
#define VECNUM 9
#define BIG_VECTOR 10
#define INPUT_FILENAME "thrdbf.input"
mpcounter c1, c2;
float buffer1[BIG_VECTOR], buffer2[BIG_VECTOR];
FILE *finp;
main ()
{
void vinit (), parallel_func ();
/* read_buf() and write_buf() handle I/O to and from a specified buffer */
int i, ret, read_buf (), write_buf ();
int end_reached = 0;
c1 = mp_counter_alloc ();
c2 = mp_counter_alloc ();
/* Create initial data */
vinit ();
finp = fopen (INPUT_FILENAME, "r");
/* Initializing the counters is a useful precaution that is not really
needed in this example, because when the counters are allocated they
have the value 0. But if the entire loop were to be repeated,
re-initializing the counters here would be critical. */
ret = mp_counter_init (c1, 0);
ret = mp_counter_init (c2, 0);
/* Start other thread to perform calculations, and begin main loop in
root thread */
ret = mp_create (parallel_func, 0);
/* Double-buffering algorithm requires one buffer to be read once before
the main loop */
if (read_buf (buffer1))
exit (-2);
/* Let calculating thread start on buffer 1 */
ret = mp_counter_incr (c1);
while (!end_reached) /* *** Start of main loop *** */
{
/* Read data into second buffer, ending loop if EOF or an error occurs.
Increment counter to let calculating thread start on buffer. */
if (!(ret = read_buf (buffer2)))
ret = mp_counter_incr (c2);
else
{
end_reached = 2;
break;
}
/* Odd counter means the calculating thread is still handling the buffer */
while (mp_counter_fetch (c1) & 1)
;
/* When the calculating thread finishes the first buffer, write it and
read new data into it, ending loop if end-of-file reached.
Increment counter to let calculating thread start on buffer. */
write_buf (buffer1);
if (!(ret = read_buf (buffer1)))
ret = mp_counter_incr (c1);
else
{
end_reached = 1;
break;
}
/* Write second buffer when calculating thread is finished with it */
while (mp_counter_fetch (c2) & 1)
;
write_buf (buffer2);
} /* *** End of Main loop *** */
/* Double-buffering algorithm requires a buffer to be written once after
the main loop. Wait on counter so that last calculation can finish
before the write. */
ret = 0;
switch (end_reached)
{
case 1:
while (mp_counter_fetch (c2) & 1)
;
write_buf (buffer2);
break;
case 2:
while (mp_counter_fetch (c1) & 1)
;
write_buf (buffer1);
break;
default:
fprintf (stderr, "Incorrect return value %d from W/R function\n",
end_reached);
ret++;
break;
}
/* Clean up */
mp_counter_free (c1);
mp_counter_free (c2);
fclose (finp);
exit (ret);
}
void parallel_func ()
{
void calculate ();
int ret;
while (1)
{
/* Even counter means the I/O thread is still handling the buffer */
while (!(mp_counter_fetch (c1) & 1))
;
calculate (buffer1);
ret = mp_counter_incr (c1);
while (!(mp_counter_fetch (c2) & 1))
;
calculate (buffer2);
ret = mp_counter_incr (c2);
}
}
int write_buf (buf)
float buf[];
{
int i;
for (i = 0; i < BIG_VECTOR; i++)
if (fprintf (stdout, "%f ", buf[i]) <= 0)
{
fprintf (stderr, "Error writing to file\n");
exit (-3);
}
fprintf (stdout, "\n");
}
int read_buf (buf)
float buf[];
{
int i;
for (i = 0; i < BIG_VECTOR; i++)
if (fscanf (finp, "%f", &buf[i]) <= 0)
return (1);
return (0);
}
/* For this example, a trivial calculation is used: replacing each
element n with the sum of itself and the element at BIG_VECTOR-n. If
BIG_VECTOR is odd, the middle element is untouched. */
void calculate (buf)
float buf[];
{
float tmp;
int i;
for (i = 0; i < BIG_VECTOR / 2; i++)
{
tmp = buf[i] + buf[BIG_VECTOR - i - 1];
buf[i] = tmp;
buf[BIG_VECTOR - i - 1] = tmp;
}
}
void vinit ()
{
int i, j;
finp = fopen (INPUT_FILENAME, "w");
for (i = 0; i < VECNUM; i++)
{
for (j = 0; j < BIG_VECTOR; j++)
{
float tmp;
tmp = ((j + 1.0) * (BIG_VECTOR - j) - 1.0) /
(VECNUM - VECNUM % (i + 1)) + 1.0;
if (fprintf (finp, "%f ", tmp) <= 0)
{
fprintf (stderr, "Error creating input file\n");
exit (-1);
}
}
fprintf (finp, "\n");
}
fclose (finp);
}
現在、ＰＯＳＩＸ Ｐ１００３．４やＲｅａｌｔｉｍｅＭＡＣＨ、ＬｙｎｘＯＳ、ＶｘＷｏｒｋｓなど、リアルタイムＯＳもずいぶんメジャーな話題として伝えられるようになりました。これは、コンカレント社がＲＴＵを世界で最初にＵＮＩＸをリアルタイム化した（当時はマスコンプ社でした）当時からは考えられないことです。
けれども、いちばん重要なことは、信頼性のあるリアルタイムシステムを構築するためには、十分な経験を積まねばなりません。
たとえば、コンカレント７０００シリーズは、ＲＴＵを最も効率よく実行するためにＲＴＵ用に設計した初めてのシリーズですが、ここまでくるのに１０年間に渡って実際のフィールドで経験を積まなくてはなりませんでした。
特に並列処理の技術はその究極にあるものでそのノウハウは、一朝一夕に得難いものです。
その経験をもとに、今後重要になるリアルタイム処理と多重スレッドについて、標準ＵＮＩＸとの違いを説明してきました。
おそらく、今後リアルタイムと並列処理の技術は、ネットワーク技術についでＵＮＩＸエンジニアにとって必須の技術となると思われます。そのような意味で、これが少しでも役にたてば幸いです。
１９９２年８月 コンカレント日本株式会社 技術部
"Real-Time and Multipocessing Manual"
"Realtime Programming Manual"
"Writing a UNIX Device Driver"
以上 Concurrent Computer Corporation.
"ＵＮＩＸデバイスドライバ" 野中浩一 大西照代共訳 アスキー出版局（この本は 3) の抄訳である）
"２枚のＶＭＥ６Ｕボードでメモリ共有型３プロセッサシステムを実現したリアルタイムコンピュータ７０００シリーズ" 大島龍博／田中直樹 １９９２年電子情報通信学会春期大会Ｄ−１３９
"リアルタイムシステムにおけるＵＮＩＸと並列処理" 大島龍博／堀口幹友 ｊｕｓ設立１０周年記念ＵＮＩＸ国際シンポジウム論文集 ｊａｐａｎ ＵＮＩＸ Ｓｏｃｉｅｔｙ
"３個の６８０４０を搭載したＶＭＥｂｕｓボードなど" 日経エレクトロニクス
１９９２年３月２日号 No548 pp 231
"C Threads" Eric C.Cooper Richard P.Draves Department of Computer Sciencd Carnegie Mellon University
Draft of 20 July 1987
"リアルタイムＵＮＩＸシステム" 成田光彰訳
日経ＢＰ社
リアルタイムとＵＮＩＸ特集 ＵＮＩＸマガジン
１９９１年９月号 pp68-87
"リアルタイムＵｎｉｘＦＡ市場へ発進" 日経エレクトロニクス １９８２年１２月２６日号 No.463 pp117-124
"リアルタイムＯＳの基本技術を見渡す" 日経エレクトロニクス １９８８年４月４日号 No.444 pp161-177
"ＵＮＩＸ ４．３ＢＳＤの設計と実装" 中西明／相田仁／小池汎平共訳 丸善株式会社
"ＵＮＩＸシステムチューニング" Mike Loukides 著／砂原秀樹監訳 アスキー出版局
POSIX P1003.4/D12 February 1992 IEEE STD Dept.
"ＵＮＩＸ原典" 石田晴久監修 パーソナルメディア
"The Interaction of Architecture and Operating System Design" Thomas E.Anderson,Henry M.Levy,Brian N.Bershad and Edward D.Lazowska [ASPLOS-IV Proceeding pp.108-120(Apr.1991)]
"リアルタイム処理のＰＯＳＩＸ標準が定まる、多くのベンダが準拠を表明" 日経エレクトロニクス １９９１年９月３０日号 No.537 pp127-133
"米サンのＳｏｌａｒｉｓ、パソコンとＷＳの共通ＯＳを目指す" 日経エレクトロニクス １９９２年２月３日号 No.546 pp157-164
"並行プログラミングの原理" リチャード．Ｃ．ホルト著／大野豊監訳 啓学出版
"並行処理とＵＮＩＸ" Ｍ．ベン−アリ著／渡辺栄一訳 啓学出版
"並列処理マシン" 富田真治／末吉敏則共著 オーム社
"並列処理技術" 笠原博徳著 コロナ社
"新しいＯＳ" 石田晴久／土居範久共同編集 共立出版株式会社
Back