160 lines
17 KiB
TeX
160 lines
17 KiB
TeX
|
|
\section{MWRE Workflow Enactment}\label{sec:MWRE-EnactmentModel}
|
|
This section describes the enactment model of our workflow engine. In order to be able to meet the requirements of shared-memory manycore systems, we designed it significantly different from traditional scientific workflow engines for DCIs by paying special attention to the efficient use of resources, low overheads, and arbitrarily complex data types for input and output activity ports.
|
|
|
|
\begin{wrapfigure}{R}{0pt}
|
|
\centering
|
|
\includegraphics[keepaspectratio=true, width=.5\textwidth]{./mwre/figures/engine-schematics.pdf}
|
|
\caption{MWRE workflow execution model.}
|
|
\label{fig:engine-schematics}
|
|
\end{wrapfigure}
|
|
|
|
\subsection{Workflow enactment model}
|
|
Workflow engines for DCIs usually rely on reflection and type introspection to generate the complete workflow structure and retrieve detailed activity information. Based on this information, the engine executes the workflow and handles the dependency resolution and the data transfers. One problem with this approach is its high overhead and increased resource use. Furthermore, to reduce the complexity and simplify the engine implementation, only a restricted set of data types for ports is usually supported. While this is sufficient for DCIs where complex data structures are usually transmitted via files, it is more efficient on shared memory systems to directly transfer them via shared memory requiring support for arbitrarily complex data types. To circumvent these problems, we use a novel design which, instead of relying on reflection and type introspection, employs a callback mechanism to resolve dependencies, transfer data, and handle composite activities.
|
|
|
|
Figure~\ref{fig:engine-schematics} shows the execution model of our engine. First (\circled{1}), the engine reads the workflow structure from the workflow program. Listing~\ref{alg:povray_structure} shows the structure of the C++ workflow program generated from the POV-Ray workflow representation. Lines~\ref{alg:povray_structure:SceneFileImpls:start}~--~\ref{alg:povray_structure:SceneFileImpls:end} list the available activity implementations of the \texttt{SceneFile} atomic activity, lines~\ref{alg:povray_structure:RenderImpls:start}~--~\ref{alg:povray_structure:RenderImpls:end} the implementations of the \texttt{Render} atomic activity, and lines~\ref{alg:povray_structure:ConvertImpls:start}~--~\ref{alg:povray_structure:ConvertImpls:end} the implementations of the \texttt{Convert} atomic activity. Afterwards, lines~\ref{alg:povray_structure:activities:start}~--~\ref{alg:povray_structure:activities:end} encode a table containing all workflow activities. Each line in this table represents a single activity and contains the activity name, the parent activity which represents the composite activity the activity is contained within, the successors and predecessors of the activity representing the data- and control-flow dependencies, the callbacks associated with the activity, and other activity dependent information. Finally, line~\ref{alg:povray_structure:workflow} defines the workflow itself and defines the initial start activity.
|
|
|
|
\begin{listing}[t]
|
|
\scriptsize
|
|
\begin{algorithmic}[1]
|
|
\State ActivityImplemenation SceneFileImpls[] = \{ \label{alg:povray_structure:SceneFileImpls:start}\\
|
|
\ \ \ \ /* \{type, function-pointer, meta-data\} */ \\
|
|
\ \ \ \ PThread, \&renderImplFunc, \{\{``SSE'', ``SSE2''\}\} \\
|
|
\};
|
|
\Statex \label{alg:povray_structure:SceneFileImpls:end}
|
|
|
|
\State ActivityImplemenation RenderImpls[] = \{ \label{alg:povray_structure:RenderImpls:start}\\
|
|
\ \ \ \ PThread, \&renderImplFunc, \{\{``SSE'', ``SSE3''\}, \{``key2'', ``value2''\}\} \label{alg:povray_structure:meta-data-example}\\
|
|
\};
|
|
\Statex \label{alg:povray_structure:RenderImpls:end}
|
|
|
|
\State ActivityImplemenation ConvertImpls[] = \{ \label{alg:povray_structure:ConvertImpls:start}\\
|
|
\ \ \ \ PThread, \&convertImplFunc1, \{\{``SSE'', ``SSE''\}, \{``key3'', ``value3''\}\} \\
|
|
\ \ \ \ OpenCl, \&convertImplFunc2, \{\{``OpenCL'', ``2.1''\}\} \\
|
|
\};
|
|
\Statex \label{alg:povray_structure:ConvertImpls:end}
|
|
|
|
\State ActivityTemplate Activities[] = \{ \label{alg:povray_structure:activities:start}\\
|
|
\ \ \ \ /* (name, ID, parentID, branchID, succs, preds, callbacks, activity-specific) */ \\
|
|
\ \ \ \ AT::Container(``toplevel'', 1, 0, 0, \{\}, \{\}, callbacks, ...), \label{alg:povray_structure:activities:0}\\
|
|
\ \ \ \ AT::Atomic(``SceneFile'', 2, 1, 0, \{3\}, \{\}, callbacks, SceneFileImpls, 1, ...), \\
|
|
\ \ \ \ AT::PForLoop(``PFor'', 3, 1, 0, \{5\}, \{2\}, callbacks, ...), \\
|
|
\ \ \ \ AT::Atomic(``Render'', 4, 2, 0, \{\}, \{\}, callbacks, RenderImpls, 1, ...), \\
|
|
\ \ \ \ AT::Atomic(``Convert'', 5, 1, 0, \{\}, \{3\}, callbacks, ConvertImpls, 2, ...), \\
|
|
\};
|
|
\Statex \label{alg:povray_structure:activities:end}
|
|
|
|
|
|
\State /* (workflow name, activities, id of start activity) */ \\
|
|
Workflow workflow(``povray'', Activities, 1);\label{alg:povray_structure:workflow}
|
|
\Statex
|
|
|
|
\State int main() \{ \label{alg:povray_structure:main:start}\\
|
|
\ \ \ wf\_input wf\_int = ...; \\
|
|
\ \ \ wf\_output wf\_out = WorkflowEngine.startWorkflow(workflow, wf\_in); \\
|
|
\}\label{alg:povray_structure:main:end}
|
|
\end{algorithmic}
|
|
\caption{POV-Ray workflow program snippet.}
|
|
\label{alg:povray_structure}
|
|
\end{listing}
|
|
|
|
\begin{listing}[t]
|
|
\scriptsize
|
|
\begin{algorithmic}[1]
|
|
\Function{visitPFor}{activity instance $AI_i$}
|
|
\If{$\lnot AI_i$.\Call{allInputsAvailable}{}}
|
|
\State \Call{$AI_i$.\Call{inputCallback}{}}{}\label{alg:visitpfor:inputCallback}
|
|
\EndIf
|
|
\If{$AI_i$.state $<$ \texttt{ReadyForExecution}}
|
|
\State \Call{$AI_i$.\Call{forCounterCallback}{}}{}\label{alg:visitpfor:forCallback}
|
|
\EndIf
|
|
\If{$AI_i$.state = \texttt{ReadyForExecution}}
|
|
\If{$\lnot AI_i$.\Call{isUnrolled}{}}
|
|
\State \Call{UnrollLoop}{$AI_i$}\label{alg:visitpfor:unrollLoop}
|
|
\EndIf
|
|
\State \Call{visitChildren}{$AI_i$}\label{alg:visitpfor:visitChildren}
|
|
\EndIf
|
|
\If{$AI_i$.state $\ge$ \texttt{ReadyForExecution} $\land \lnot AI_i$.\Call{allOutputsAvailable}{}}
|
|
\State \Call{$AI_i$.\Call{outputCallback}{}}{}\label{alg:visitpfor:outputCallback}
|
|
\EndIf
|
|
\If{$AI_i$.state $<$ \texttt{Finished} $\land$ all children are finished}
|
|
\State $AI_i$.\Call{setState}{\texttt{Finished}}\label{alg:visitpfor:finished}
|
|
\EndIf
|
|
\EndFunction
|
|
\end{algorithmic}
|
|
\caption{Visitor function of a \texttt{parallel for} loop.}
|
|
\label{alg:visitpfor}
|
|
\end{listing}
|
|
|
|
\begin{listing}[t]
|
|
\scriptsize
|
|
\begin{algorithmic}[1]
|
|
\Function{forLoopInputCallback}{ForLoopInstance $for$}\label{alg:povray_callbacks:inputcallback}
|
|
\State $for\_in \gets$ \Call{retrieveInStruct}{$for$}\Comment{Initialisation of $for\_in$ input port}
|
|
\State $parent\_in \gets$ \Call{retrieveInStruct}{\Call{getParentInstance}{$for$}}
|
|
\If{\Call{isValid}{$parent\_in.povFile$}}
|
|
\State $for\_in.povFile \gets parent\_in.povFile$
|
|
\EndIf
|
|
\Statex
|
|
\State [\ldots] \Comment{Similar code for initializing $totalFrames$, $framesPerActivity$ and $scene$ ports}
|
|
\Statex
|
|
\If{\Call{isValid}{$for\_in.povFile$, $for\_in.totalFrames$, $for\_in.framesPerActivity$, $for\_in.scene$}}
|
|
\State $for$.\Call{setAllInputsAvailable}{\texttt{true}}
|
|
\EndIf
|
|
\EndFunction
|
|
\Statex
|
|
|
|
\Function{forLoopOutputCallback}{ForLoopInstance $for$}\label{alg:povray_callbacks:outputcallback}
|
|
\State $for\_out \gets$ \Call{retrieveOutStruct}{$for$}
|
|
\For{$i \gets 0$ \textbf{to} $for$.getIterationCount()}
|
|
\State $child\_out \gets$ \Call{retrieveOutStruct}{\Call{getChildInstance}{$for, i, 0$}}
|
|
\If{\Call{isValid}{$child\_out.frames$}}
|
|
\State $for\_out.frames$.set(i, $child\_out.frames$)
|
|
\EndIf
|
|
\EndFor
|
|
\If{\Call{containsNValidEntries}{\mbox{$for$.\Call{getIterationCount}{}, $for\_out.frames$}}}
|
|
\State $for$.\Call{setAllOutputsAvailable}{\texttt{true}}
|
|
\EndIf
|
|
\EndFunction
|
|
\Statex
|
|
|
|
\Function{forLoopCounterCallback}{ForLoopInstance $for$}\label{alg:povray_callbacks:countercallback}
|
|
\State $for\_in \gets$ \Call{retrieveInStruct}{$for$}
|
|
\If{isValid($for\_in.totalFrames, for\_in.framesPerActivity$)}
|
|
\State $for$.\Call{setCounter}{\mbox{0, $for\_in.totalFrames / for\_in.framesPerActivity$, 1}}
|
|
\State $for$.\Call{setState}{\texttt{ReadyForExecution}}
|
|
\EndIf
|
|
\EndFunction
|
|
\end{algorithmic}
|
|
\caption{Callback functions of the POV-Ray \texttt{ParallelFor} loop activity.}
|
|
\label{alg:povray_callbacks}
|
|
\end{listing}
|
|
|
|
The engine maintains bookkeeping record information for the workflow and each activity instance. A record entry includes, among other things, the current state of the activity instance and references to the input and output data buffers. By amending the engine with an API that allows accessing these records, third-party applications can easily monitor the workflow progress. Furthermore, workflow steering can be easily implemented by allowing manipulation of the input and output data structures by third-party applications. However, due to its architecture, the engine cannot provide data type information inside input/output data buffers and, therefore, third-party applications used for workflow steering need to retrieve this information from other sources.
|
|
|
|
From the workflow structure, the engine constructs the WEP (\circled{2}) and repeatedly traverses it during the enactment. When traversing the WEP, the engine executes so-called \emph{visitor functions} for each node. In traditional engines, the visitor functions are directly responsible for the dependency resolution and data transfers. In contrast, the visitor functions in our engine are only responsible for orchestrating the execution of the associated pre-compiled callback functions (\circled{3}) that directly modify the state of the associated workflow activities (\circled{4}). The callback functions are part of the workflow specification, and it is the responsibility of the source-to-source compiler to generate them. Each callback function implements a specific functionality, such as transferring the input data, collecting the output data from children for composite activities, or resolving the condition of a \texttt{while} loop. This approach has the advantage of keeping the workflow engine lightweight and efficient. Instead of knowing all the details about the workflow, the engine only knows what is necessary to traverse the workflow structure and to keep track of the execution status, with no need to implement generic functionality for dependency resolution, data transfer and other tasks that introduce performance overheads. Another advantage is that it allows arbitrary data type support and facilitates extensibility. Every functionality that needs information about data types is encapsulated in a purposely-tailored callback function used in a particular workflow. New functionality can be easily implemented in callback functions too without needing to modify the engine.
|
|
|
|
\begin{wrapfigure}{R}{0pt}
|
|
\centering
|
|
\includegraphics[keepaspectratio=true, width=.5\textwidth]{./mwre/figures/activityStateDiagram.pdf}
|
|
\caption{Simplified state diagram of a \texttt{parallel for} activity.}
|
|
\label{fig:state-diagram}
|
|
\end{wrapfigure}
|
|
|
|
|
|
Listing~\ref{alg:visitpfor} shows the generic visitor function used to visit a \texttt{parallel for} loop (Figure~\ref{fig:state-diagram} shows a simplified state diagram of the \texttt{parallel for} loop). At first, it calls the input callback function in line~\ref{alg:visitpfor:inputCallback} to fetch the input data from its predecessors, if they are not already available. If the \texttt{for} loop is not ready for execution (meaning that the for loop counter has not been yet evaluated), the evaluating \texttt{for} counter callback gets executed in line~\ref{alg:visitpfor:forCallback}. When the for loop is ready for execution (meaning that the iteration space of the loop counter is known), the loop gets unrolled and an activity instance is created for each child and each iteration in line~\ref{alg:visitpfor:unrollLoop}. Afterwards, the visitor functions for the children activities are called in line~\ref{alg:visitpfor:visitChildren}. In our case, the atomic activity \texttt{render} is the only child. The visitor function of an atomic activity executes the input callback function and, when all inputs are satisfied, the activity gets executed and the output data is available in the output buffer of the atomic activity. The visitor function of the \texttt{for} loop activity executes the output callback function in line~\ref{alg:visitpfor:outputCallback}, which fetches data from the output buffers of the child activities and stores them into the output buffer of the loop activity. After all child activities have finished their execution, the activity state is set to \texttt{Finished} in line \ref{alg:visitpfor:finished}.
|
|
|
|
As an example, Listing~\ref{alg:povray_callbacks} shows the implementations of the callback functions of the \texttt{parallel for} loop (\texttt{ParallelFor}) of the POV-Ray workflow. The first callback (line~\ref{alg:povray_callbacks:inputcallback}) is the \emph{input callback} responsible for initializing the activity's input data buffer. For this, it accesses the input data buffer of the parent activity, checks whether it contains valid data, saves it into the input buffer of the \texttt{for} activity, and sets the \texttt{allInputsAvailable} flag to \texttt{true} if all input data fields contain valid data to avoid unnecessary function invocations. The second callback (line~\ref{alg:povray_callbacks:outputcallback}) is the \emph{output callback} responsible for filling in the output data buffers by accessing the output data buffers of its children, checking the validity of the data, storing the data into the output buffer of the \texttt{for} loop, and setting the \texttt{allOutputsAvailable} flag to \texttt{true} if all output data fields contain valid data. The third callback (line~\ref{alg:povray_callbacks:countercallback}) is the \emph{for counter callback} responsible for initializing the loop counter of the \texttt{for} loop activity \texttt{PForLoop} by checking if all required data is available, setting the counter, and declaring the activity as ready-for-execution by modifying its state.
|
|
|
|
|
|
\subsection{Scheduling and activity execution}
|
|
|
|
After the WEP has been evaluated for the first time and after every change, the workflow engine sends the new WEP to the scheduler. The scheduler then maps atomic activities to computing resources. Traditional workflow engines for DCIs often only lazily evaluate the WEP as far as it is required for workflow enactment to enact the ready to execute activities. However, full-ahead schedulers require a WEP that is as complete as possible to achieve a better schedule and, therefore, need to compute the WEP on their own. While this was not a major issue in traditional engines, as they are more tolerant to high overheads and usually run on own external dedicated systems, the performance overhead are of a greater concern for workflows on shared memory manycore systems. Because of this, we designed the WEP maintained by the workflow engine as complete as possible and such that all the relevant information is shared with the scheduler to avoid unnecessary recalculations.
|
|
|
|
Each activity implementation comes with meta-data in form of key-value pairs that can be used by the scheduler to select an appropriate implementation for execution. There is a set of well-defined key-value pairs such as the key \texttt{OpenCL} used to specify the OpenCL version of a specific activity implementation, or \texttt{SSE} to specify the required version of the Streaming SIMD Extension. Moreover, users can also define their own arbitrary key-value pairs. For example, line~\ref{alg:povray_structure:meta-data-example} in Listing~\ref{alg:povray_structure} specifies that the activity implementation requires \emph{SSE3} and there is also a user-defined key-value pair \texttt{key2=value2}.
|
|
|
|
After the scheduler has selected appropriate activity implementations and generated a schedule, the individual scheduling decisions are sent to the appropriate execution subsystems (e.g. \texttt{pthread} subsystem for \texttt{x86} code, \texttt{CUDA}/\texttt{OpenCL} subsystem for GPU kernels). For this purpose, MWRE provides a common extensible interface for the execution subsystems designed to support a wide variety of computing resources and to allow the scheduler interact with a specific execution subsystem without knowing the details of the underlying implementation. The execution results are then send back to the scheduler, which then handles them accordingly and notifies the workflow engine of any activity state changes.
|
|
|
|
Whenever an activity execution fails, the scheduler decides how the fault is resolved. We offer an API for implementing third-party schedulers and using this API the scheduler can resubmit the activity as it is, select a different activity implementation, map the activity onto a different resource, or simply fail the entire workflow. For example, our scheduler implementations try to resubmit a failed activity three times and after that the entire workflow is put into failed state, but other implementations may come up with more sophisticated fault handling methods.
|
|
|