% Todo
% 並列処理 の全体的な説明
% 構成のセクションいるかしら?
% par goto をコンパイルタイミングでflowを解析し(モデル検査で)処理が軽い場合は並列ではなく call にするみたいな最適化
% taskManager の初期化のコードはいらない

\chapter{Gears OSの並列処理}
Gears OS では実行の Task を Code Gear と Input/Output Data Gear の組で表現する。
Input/Output Data Gear によって依存関係が決定し、それにそって並列実行を行う。

本章では、Gears OS の並列処理の構成、機能について説明する。

\section{Task}
Gears OS では 並列実行する Task を Context で表現する\cite{ikkun-sigos}。
Context には Task 用の情報として、実行される Code Gear、Input/Output Data Gear の格納場所、待っている Input Data Gear のカウンタ等を持っている。
Task の Input Data Gear が揃っているかを TaskManager で判断し、 実行可能な Task を Worker に送信する。
Worker は送信された Task が指定した Code Gear を実行し、Output Data Gear を書き出す。

実行される Code Gear の例を \coderef{codeGearExample} に示す。
\coderef{codeGearExample} は Integer 型 の Input Data Gear を2つ受け取り、加算処理を行い、Integer 型 の Output Data Gear に書き出す。
並列処理を行う Code Gear は Interface の Code Gear と同じく、引数に Input Data Gear、処理が終了した後に継続する Code Gear、引数の Code Gear の中に Output Data Gear を記述する(\coderef{codeGearExample} 1行目)。
この Code Gear 実行後は通常 Output Data Gear を書き出す Code Gear に継続する。
実際に Output Data Gear を書き出す場合、goto 文に Output Data Gear を引数に渡す(\coderef{codeGearExample} 3行目)。

\lstinputlisting[caption=並列実行される Code Gear の例, label=code:codeGearExample]{./src/codeGearExample.cbc}

% Context の話を書く
% 生成されるstub は Context から input/output に index を指定して生成する。
Task のInput/Output Data Gear の格納場所は Context の Task 情報が持っている。
その為、Task に対応する Code Gear の stub Code Gear は context が持っている Input/Output Data Gear用のインデックスから Data Gear を取り出す。
\coderef{metaCodeGearExample} に \coderef{codeGearExample} の stub Code Gear を示す。
この stub Code Gear も Interface の stub Code Gear と同等にスクリプトによって自動生成される。

\lstinputlisting[caption=並列実行される Code Gear のstub Code Gear, label=code:metaCodeGearExample]{./src/metaCodeGearExample.cbc}

\section{TaskManager}
Gears OS の TaskManager は Task を実行する Worker の生成、管理、Task の送信を行う。
\coderef{taskManagerInterface} に TaskManager の Interface を示す。

\lstinputlisting[caption=TaskManager の Interface, label=code:taskManagerInterface]{./src/taskManagerInterface.cbc}

TaskManager は 以下の API を持っている。

\begin{itemize}
    \item Task の実行(spawn、spawnTasks)
    \item Task の依存関係の設定(setWaitTask) 
    \item TaskManager が管理している Task 数のインクリメントとデクリメント(increment/decrementTaskCount)
    \item TaskManager(shutdown)の終了処理
\end{itemize}

TaskManager は初期化の際に、指定した数の Worekr を生成する。
その際CPU、GPU の数を指定することができ、指定した分の CPUWorker と GPUWorker が生成される。

TaskManager は \figref{sendTask}に示すように spawn を呼び出した際、実行する Task の Input Data Gear が用意されているかを判断する。
Input Data Gear が全て用意されている場合、その Task を Worker の Queue に送信する。
 送信する Worker は Task を実行する環境(CPU、GPU) によって決定する。

\begin{figure}[htbp]
    \begin{center}
        \includegraphics[scale=0.6]{./fig/sendTask.pdf}
    \end{center}
    \caption{Workerへの Task 送信}
    \label{fig:sendTask}
\end{figure}

\section{Worker}
Worker は自身の Queue から Task を取得し、Task の Code Gear を実行し、Output Data Gear の書き出しを行っている。

\coderef{createCPUWorker} に Task を CPU で実行する CPUWorker の初期化部分を示す。
CPUWorker は初期化の際に スレッドを生成する(\coderef{createCPUWorker} 10行目)。
生成されたスレッドはまず startWorker 関数(\coderef{createCPUWorker} 14-21行目)を呼び出し、このスレッド用の Context を生成する。
Context をスレッド毎に生成することで、メモリ空間をスレッドごとに持てるため Gearef マクロ で Interface の引数を取得する際の競合、メモリ確保の処理での他のスレッドの停止を防ぐ事ができる。

\lstinputlisting[caption=CPUWorker の初期化, label=code:createCPUWorker]{./src/createCPUWorker.cbc}

Context の生成後は Queue から Task を取得する Code Gear へ継続する。
Task は Context なので、Worker は Context を入れ替えて Task の実行を行う。

CPUWorker での Task の実行を\coderef{workerRun}、 \figref{workerRun} に示す。
\coderef{workerRun} は Context の入れ替えを行うため、getTaskCPUWorker(\coderef{workerRun} 1-9行目)の引数に入れ替え後のTask(Context)を記述している。
Worker は中身が NULL の task を取得すると Worker の終了処理を行う(\coderef{workerRun} 2-4 行目)。
Task が取得できた場合 Task の実行後に継続する Code Gear を格納し(\coderef{workerRun} 7行目)、Task を Context としてCode Gear に継続する(\coderef{workerRun} 8行目)。
Task の実行後に継続する Code Gear は Data Gear の書き出しと依存関係の解決を行う。
Data Gear 書き出し後は Task の Context を Worker の Context に入れ替え、再び Task を取得する Code Gear に継続する。

\lstinputlisting[caption=CPUWorker でのTaskの実行, label=code:workerRun]{./src/workerRun.cbc}

\begin{figure}[htbp]
    \begin{center}
        \includegraphics[scale=0.6]{./fig/workerRun.pdf}
    \end{center}
    \caption{Workerでの Task 実行}
    \label{fig:workerRun}
\end{figure}

\section{SynchronizedQueue}
SynchronizedQueue は Worker の Queue として使用される。
Worker の Queue は TaskManager を経由して Task を送信するスレッドと Task を取得する Worker 自身のスレッドで扱われる。
そのため SynchronizeQueue はマルチスレッドでもデータの一貫性を保証する Queue を実装する必要がある。

データの一貫性を保証する解決例としての1つとしてロックを使った解決方法がある。
しかし、ロックを行ってデータを更新した場合、同じ Queue に対して操作を行う際に待ち合わせが発生し、全体の並列度が下がってしまう。
そこで、Gears OS ではデータの一貫性を保証するために CAS(Check and Set、Compare and Swap) を利用した Queue\cite{queue} を実装している。
CAS は値の比較、更新をアトミックに行う命令である。
CAS を使う際は更新前の値と更新後の値を渡し、渡された更新前の値を実際に保存されているメモリ番地の値と比較し、同じならデータ競合がないため、データの更新に成功する。
異なる場合は他に書き込みがあったとみなされ、値の更新に失敗する。

Gears OS ではこの CAS を行うための Interface を定義している(\coderef{atomicInterface})。
この Interface では、Data Gear 全てを内包している Data 共用体のポインタの値を更新する CAS を定義している(\coderef{atomicInterface} 6行目)。

\lstinputlisting[caption=AtomicInterface, label=code:atomicInterface]{./src/atomicInterface.h}

AtomicInterface での CAS の実際の実装を \coderef{atomicImpl} に示す。
実際の実装では \_\_sync\_bool\_compare\_and\_swap 関数を呼び出すことで CAS を行う(\coderef{atomicImpl} 2行目)。
この関数は第一引数に渡されたアドレスに対して第二引数の値から第三引数の値ヘ CAS を行う。
CAS に成功した場合、true を返し、失敗した場合は false を返す。
\coderef{atomicImpl} では CAS に成功した場合と失敗した場合それぞれに対応した Code Gear へ継続する。

\lstinputlisting[caption=CAS の実装, label=code:atomicImpl]{./src/atomicImpl.cbc}

SynchronizedQueue の Data Gear の定義を \coderef{synchronizedQueue} に示す。
SynchronizedQueue はデータのリストの先頭と、終端のポインタを持っている。
Queue を操作する際はこのポインタに対して CAS をすることでデータの挿入と取り出しを行う。

\lstinputlisting[caption=SynchronizedQueue の定義, label=code:synchronizedQueue]{./src/synchronizedQueue.h}

\figref{takeSynchronizedQueue1} は要素の取り出し(dequeue) を行った流れを示している。
データを取り出す際はリストの先頭を次の要素へ CAS することででデータを取得する。
この Queue では先頭に挿している要素はダミー扱いとする。
その為、実際に取り出される値は CAS に成功した後の先頭の値となる。

\begin{figure}[htbp]
    \begin{center}
        \includegraphics[scale=0.6]{./fig/takeSynchronizedQueue1.pdf}
    \end{center}
    \caption{SynchronizedQueueによる要素の取り出し}
    \label{fig:takeSynchronizedQueue1}
\end{figure}

\figref{putSynchronizedQueue1} は要素の挿入(inqueue) を行った流れを示している。
データを挿入する際は2度の CAS を行う。
まず、末尾の要素の次の要素を新しい要素に CAS を行う。
その後、末尾の要素が差している要素を挿入に成功した新しい要素に CAS を行う。

\begin{figure}[htbp]
    \begin{center}
        \includegraphics[scale=0.6]{./fig/putSynchronizedQueue1.pdf}
    \end{center}
    \caption{SynchronizedQueueによる要素の挿入}
    \label{fig:putSynchronizedQueue1}
\end{figure}

しかし、データの挿入する際は2度の CAS の間に他のスレッドの操作が入り、Queueの構造が破綻する場合がある。
例えば\figref{putSynchronizedQueue2} に示すように、あるスレッドが末尾を更新した際に、他のスレッドが更新処理を行うとQueue が破綻する。

\begin{figure}[htbp]
    \begin{center}
        \includegraphics[scale=0.6]{./fig/putSynchronizedQueue2.pdf}
    \end{center}
    \caption{SynchronizedQueue によるデータの挿入時の破綻例}
    \label{fig:putSynchronizedQueue2}
\end{figure}

\newpage

\figref{putSynchronizedQueue2}は2つのスレッドが以下の順番で処理を行っている。

\begin{enumerate}
    \item threadA: 末尾の要素(data3)の次の要素を新しい要素(data4)に CAS を実行する
    \item threadB: threadA での末尾更新をする前に末尾の要素(data3)の次の要素を新しい要素(data5)にCASを実行する。
          この時の末尾が指す要素は、threadA が要素挿入を行う前の状態と同じなため、この操作を行うとリストが破綻する。
    \item threadA: 末尾の要素(data3)を挿入に成功した要素(data4)に CAS を行う。
    \item threadB: 末尾の要素(data3)を挿入に成功した要素(data5)に CAS を行う。
          threadB が挿入の操作を行ったときの末尾はthreadA が末尾更新する前の末尾要素なので、このCAS は失敗する。
\end{enumerate}

この破綻は末尾の要素が必ず末尾を示していると仮定しているためである。
しかし、データ挿入の際は2度の CAS の間に他のスレッドが割り込む場合がある。
そこで、末尾の要素は必ずしも末尾を示さないと仮定して要素の取出しと挿入の処理を記述する。
\coderef{putSynchronizedQueue} は要素の挿入の処理の一部である。
末尾の要素が末尾を示さない場合の処理は\coderef{putSynchronizedQueue} の 13-16 行目に記述している。
変数 nextElement は 末尾要素の次の要素を示しており、NULL ではない場合は末尾を差していない状態ということになる。
その場合は、末尾の要素をnextElement に CAS を行う。
この処理は要素の取り出しを行う際にも行われる。

\lstinputlisting[caption=SynchronizedQueue による要素の挿入, label=code:putSynchronizedQueue]{./src/putSynchronizedQueue.cbc}

\section{依存関係の解決}
Gears OS は並列処理の依存関係を Input と Output の Data Gear と Code Gear の関係で解決する。
Code Gear は Input に指定した Data Gear が全て書き込まれると実行され、実行した結果を Output に指定した Data Gear に書き出しを行う。

Data Gear はメタレベルで依存関係解決のための Queue を持っている。
この Queue にはその Data Gear を Input Data Gear として使用する Task(Context)が入っている。

依存関係の解決の流れを\figref{dependency} に示す。
Worker は Task の Code Gear を実行後、Output Data Gear の 書き出し処理(Commit)を行う。
書き出し処理は Data Gear の Queue から、依存関係にある Task を参照する。
参照した Task には実行に必要な Input Data Gear のカウンタをもっているので、そのカウンタのデクリメントを行う。
カウンタが $0$ になったら Task が待っている Input Data Gear が揃ったことになるので、その Task を TaskManager 経由で 実行される Worker に送信する。

\begin{figure}[htbp]
    \begin{center}
        \includegraphics[scale=0.7]{./fig/dependency.pdf}
    \end{center}
    \caption{依存関係の解決処理}
    \label{fig:dependency}
\end{figure}

\section{並列構文}
Gears OS では並列実行する Task の設定をメタレベルで \coderef{metaCreateTask} のように行っている。
\coderef{metaCreateTask} では実行する Code Gear、待ち合わせ中の Input Data Gear の数、Input/Output Data Gear への参照等の設定を記述している。
これらの記述は Context などを含む煩雑な記述であるが、並列実行されることを除けば通常の CbC の goto 文と同等である。
そこで、Context を直接用いないノーマルレベルの並列構文である par goto 文を用意した。

\lstinputlisting[caption=メタレベルによる Task の生成, label=code:metaCreateTask]{./src/metaCreateTask.cbc}

\coderef{parGotoCreateTask}に par goto文による記述例を示す。
この記述はスクリプトにより、Task で実行される Code Gear の Input/Output の数を解析し、\coderef{metaCreateTask} に変換される。

\lstinputlisting[caption=par goto による並列実行, label=code:parGotoCreateTask]{./src/parGotoCreateTask.cbc}

par goto の引数には Input/Output Data Gear と 実行後に継続する Code Gear を渡す。
par goto で生成された Task は \_\_exit に継続することで終了する。
Gears OS の Task は Output Data Gear を生成した時点で終了するので、par goto では直接 \_\_exit に継続するのではなく、Output Data Gear への書き出し処理に継続される。
これにより Code Gear と Data Gear の依存関係をノーマルレベルで記述できるようになる。
この par goto 文は 通常のプログラミングの関数呼び出しのように扱える。

\section{Task(Context) 間の同期処理}
Gears OS では複数の Task(Context) から同じ Output Data Gear を修正する場合がある。
その際に適切な同期処理を行わずそのまま実行すると Output Data Gear の整合性が取れない場合がある。

そこで 複数のTask 間の同期処理を行うために Semaphore の実装を行った。
Semaphore の Interface を \coderef{semaphoreInterface} に示す。

\lstinputlisting[caption=Semaphore Interface, label=code:semaphoreInterface]{./src/semaphoreInterface.h}

Semaphore はある資源に対してアクセスできるスレッドの数を制限するものであり、P命令 と V命令がある。
P命令は資源の消費に相当し、V命令が資源の開放に相当する命令である。
P命令を行う際、資源がなければ V命令で開放されるまでそのスレッドは処理を停止する。

Gears OS の Context はスレッドに相当するため、 Gears OS 上で Semaphore を実装することは Context の停止を実装する必要がある。
Gears OS の Semaphore は Context の停止を停止用の待ち Queue を使って行う。

\figref{semaphoreSequence} に資源が1つのSemaphore に2つのContext が P命令を実行しているシーケンス図を示す。

\begin{figure}[htbp]
    \begin{center}
        \includegraphics[scale=0.4]{./fig/semaphoreSequence.pdf}
    \end{center}
    \caption{Gears OS 上 Semaphore}
    \label{fig:semaphoreSequence}
\end{figure}

\figref{semaphoreSequence} の処理の流れを以下に示す。

\begin{enumerate}
    \item Context1 が Semaphore に対して P命令を実行する。
          Semaphore には資源が残っているので資源を消費する
    \item Context2 が Semaphore に対して P命令を実行する。
          この時、Semaphore に資源が残っていないので Context を Seamphore が持っている待ち Queue に追加する。
          その後はContext2 を取得していた Worker に処理を移譲する。
    \item Context1 が Semaphore に対して V命令を実行する。
          Semaphore の資源を開放しつつ、 待ち Queue に Context があるかの確認を行う。
          Context があった場合 1つ Context を 待ち Queue  から取得し、TaskManager へ Context の実行を行う命令を実効する。
\end{enumerate}

TaskManager に送られた Context は Worker で取得される。
取得された Context は停止した時の状態を記録しているため、停止した Code Gear である P命令を再び実行する。

\section{データ並列}
並列プログラミングを行う際、並列化の方式としてタスク並列とデータ並列の2つがある。
Gears OS の並列処理は Task(Context) を使用したタスク並列により実現されている。

タスク並列は処理をタスクに分割し、各タスク間に依存関係のないものを集め、それを並列化する。
Gears OS では依存関係を Input/Output Data Gear から解析を行い、依存関係が解決された Code Gear から実行される。
一方でデータ並列は処理対象のデータが十分な数のサブデータへ分割することが可能で、各サブデータに行う処理が同じ場合に有効な並列処理手法である。
このデータ並列は GPGPU と相性が良く、GPU 環境でも実行できる Gears OS でもサポートを行った。

Gears OS でデータ並列実行を行う場合、\coderef{iteratePargoto} のようにpar goto 文の引数にデータ並列用の構文として iterate を入れて実行する。
iterate は複数の数値を引数とし、数値の値がデータの分割数、数値の個数が次元数になる。

\lstinputlisting[caption=par goto によるデータ並列, label=code:iteratePargoto]{./src/iteratePargoto.cbc}

Gears OS は データ並列用の par goto を実行した場合、 データ並列用に Task が生成される。
この Task には \coderef{iteratorInterface} の Iterator Interface を実装した Data Gear を持たせる。
このデータ並列用の Task は Input Data Gear が揃うまでは通常の Task 同等として扱う。
依存関係が解決され、実行可能な Task になった際に、 Iterator Interface の exec を呼ぶ。
exec では  par goto で渡された次元、数値分 Task を コピーし、インデックスを割り当てる処理を行う。
この index は コピーされた Task の Input Data Gear として扱われ、Code Gear 内では通常の Data Gear として利用される。
\figref{iterateTaskExec} は1次元で 数値$4$ を渡した場合の Task 実行を示している。
コピーされた Task は通常の Task と同じように TaskManager を通して Worker に送信される。

\lstinputlisting[caption=Iterator Interface の定義, label=code:iteratorInterface]{./src/iteratorInterface.h}

\begin{figure}[htbp]
    \begin{center}
        \includegraphics[scale=0.6]{./fig/iterateTaskExec.pdf}
    \end{center}
    \caption{1次元、数値$4$のデータ並列用 Task の実行}
    \label{fig:iterateTaskExec}
\end{figure}

通常の Task であれば、実行後に Output Data Gear を書き出す処理に入るが、 データ並列用の Task はコピーされた全てのTask 実行後に Output Data Gear の書き出しを行う。
その判断と処理を行うのが Iterator Interface の barrier である。
barrier は コピーされた Task 実行後に呼ばれ、Output Data Gear が書き出せる状態なら書き出し処理を行う Code Gear に継続し、書き出せる状態でないなら Worker に操作を移譲するCode Gear に継続する。