\section*{Acknowledgement}
This work was mainly supported by the Compute-Flow-Architecture project under PNNL's DMC LDRD Initiative. It was also supported by the SO$(\text{DA})^2$ and FALLACY projects under DMC. The evaluation platforms were supported by the U.S. DOE Office of Science, Office of Advanced Scientific Computing Research, under award 66150: "CENATE - Center for Advanced Architecture Evaluation". The Pacific Northwest National Laboratory is operated by Battelle for the U.S. Department of Energy under contract DE-AC05-76RL01830.

\bibliographystyle{cbxabbrv}

\section{ARENA CGRA-Cluster Architecture}\label{sec:architecture}
\sloppy

Our ARENA prototype in this work is built upon a CGRA cluster interconnected by a fast ring network. Figure~\ref{fig:framework}(a) shows the overall design --- multiple reconfigurable nodes are connected in a ring topology. Each node incorporates a micro-controller (e.g., a simplified CPU), a task dispatcher, a CGRA, a network interface (NIC), a DMA unit, and local memory storage. At runtime, task tokens are circulating along the ring. The dispatcher can split, offload, and forward a task token based on the data range as already discussed. We adopt the ring network topology to simplify the routing strategy and provide sufficient bandwidth for delivering the small-sized task tokens ($\sim$21 bytes) with up to 16 nodes evaluated in this paper. The ring network can also provide near optimal bandwidth for most collective communication \cite{nvidia2017dgx1}, and can be easily built upon various physical network topology. We leave the exploration of alternative multi-node topology as a future work.

\begin{figure}[htb!]
	\centering
	\includegraphics[width=0.99\columnwidth]{graphs/overview.pdf}
	\caption{ARENA cluster overview and task token format.}
	\label{fig:framework}
\end{figure}

\subsection{Task Token}\label{subsec:token}

A task is represented by a task token in ARENA, which can be dynamically spawned, executed, delivered, and split. Figure~\ref{fig:framework}(b) shows the general format of the task token which comprises 7 fields: $TASK_{id}$ indicates the task will be executed, which is registered by the user (using \emph{ARENA\_task\_register()}) before launching the ARENA runtime. During execution, the reconfigurable node will be dynamically configured (see Section~\ref{subsec:cgra}) based on $TASK_{id}$. $TASK_{start}$ and $TASK_{end}$ together describe the data range for the task. $PARAM$ refers to a token-carried return value that is typically initialized by its parent task. This field is useful when performing collective operations (e.g., \emph{reduction} and \emph{accumulation}). For unavoidable remote data access, we use $REMOTE_{start}$ and $REMOTE_{end}$ to indicate the starting and ending addresses. $FROM_{node}$ labels the node where its parent task locates. Each task token thus requires 21 bytes in our prototype architecture (i.e., 4-bit each for $TASK_{id}$ and $FROM_{node}$; 4-byte each for the other fields).

\subsection{Task Dispatcher}

ARENA's task dispatcher mainly includes a task \texttt{Filter Logic} and three queues as shown in Figure~\ref{fig:arena_node}. The \texttt{Filter Logic} can offload, split, and convey a task token based on the data requirement (see Section \ref{subsec:runtime}). The NIC handles remote data requests from the task tokens in the \texttt{WaitQueue}. The \texttt{WaitQueue} will be acknowledged when the required remote data arrives at the data memory. The CGRA controller will then pop the acknowledged task token from the head of \texttt{WaitQueue} and reconfigure the CGRAs accordingly.

\begin{figure}[htb]
	\centering
	\includegraphics[width=0.9\columnwidth]{graphs/arena_node.pdf}
	\caption{{Architecture of an ARENA node.}}
	\label{fig:arena_node}
\end{figure}

\subsection{Reconfigurable CGRA Nodes and Toolchain}\label{subsec:cgra}

To achieve rapid dynamically reconfiguration, an ARENA node is prototyped with CGRA. The on-chip configuration memory is used for conserving the control signals for each task. The intra-node CGRA consists of 64 tiles connected in a mesh network. A scratchpad data memory conserves the data required for the computation. Note that both the control signals and the data are pre-loaded by the CPU/micro-controller through the DMA unit before launching the ARENA runtime. The CGRA communicates with the \texttt{Task Dispatcher} through the CGRA controller. The controller can offload tasks and coalesce spawned tokens through the \texttt{Coalescing Unit}.

\textbf{CGRA Tile --} As shown in Figure~\ref{fig:arena_node}, each CGRA tile contains a functional unit, a scratchpad control memory, a crossbar switch, and three sets of registers. The functional unit supports all the basic operations (e.g., $add$, $mul$, $shift$, $select$, $branch$, $load$, $store$, etc). Control-divergence (i.e., the existence of multiple control flow paths) inside the loop kernel is supported through partial predication~\cite{hamzeh2014branch}. The functional unit also supports the $spawn$ operation (i.e., generate a new task token and issue to the CGRA controller), which is unique in ARENA. If sufficient information is available ($TASK_{id}$, $TASK_{start}$ and $TASK_{end}$), a new token can be spawn in a single cycle; otherwise, two cycles are required to encode additional information (i.e., $PARAM$, $REMOTE_{start}$, and $REMOTE_{end}$). The $FROM_{node}$ filed will be automatically filled by the CGRA Controller. In Figure~\ref{fig:arena_node}, there are 4 tiles being able to spawn new tasks (marked in green). The leftmost tiles are connected to two 4-port scratchpad data memory banks. The functional unit can be configured to perform different operations at each cycle based on the control signals from the control memory. The CGRA tiles are granular to support the simultaneous execution of multiple tasks. Specifically, all the tiles are partitioned into 4 groups and a task can be executed by 1, 2, and 4 groups, dynamically managed by the CGRA controller.

\textbf{Control Memory --} The control signals of all the tasks are initially pre-loaded into the control memory. At runtime, tiles iterate over a subset of the control signals to execute specific tasks based on the $TASK_{id}$ in the task token. Each tile requires a 480-byte control memory in our prototype to support all application tasks evaluated in this paper (Section~\ref{sec:evaluation}). Each task has three execution modes powered by different tile groups. It takes only 8 cycles for the CGRA controller to reconfigure specific tile groups by using the data network to forward the $TASK_{id}$ systolically through the array from right to left.

\begin{figure}[htb!]
	\centering
	\includegraphics[width=0.99\columnwidth]{graphs/toolchain1.pdf}
	\caption{ARENA data-centric execution model development procedure with compiler toolchain.}
	\label{fig:procedure}
\end{figure}

\textbf{CGRA Controller --} The CGRA Controller can launch a task (using the task token at the head of \texttt{WaitQueue}) to be executed by different groups of the CGRA tiles. Based on the current CGRA utilization status and the data requirement of the target task, the CGRA controller allocate an appropriate number of groups for a waiting task. For example, if the data range required by the target task is less than a quarter of the local data range (i.e., $\text{TASK}_{\text{end}}$-$\text{TASK}_{\text{start}}<(\text{LOCAL}_{\text{end}}$-$\text{LOCAL}_{\text{start}}$)/4), only one available group (i.e., 2x8 CGRA tiles) will be allocated to the task. When the target task works on more than half of the local data range (i.e., $\text{TASK}_{\text{end}}$-$\text{TASK}_{\text{start}}>(\text{LOCAL}_{\text{end}}$-$\text{LOCAL}_{\text{start}}$)/2), the CGRA controller attempts to allocate all the four groups (i.e., entire 8x8 CGRA tiles) when available (otherwise, two groups are allocated). In addition, there are four queues in the controller to temporarily hold the spawned task tokens, which would be coalesced by the \texttt{Coalescing Unit} if any two of them imply continuous task data range and share the same $TASK_{id}$ and $PARAM$. When there are insufficient slots in the queues, the CGRA controller stops fetching tokens from the \texttt{WaitQueue} in the Task Dispatcher. Deadlock can be avoided by providing a memory attached to the CGRA controller for storing the over-spawned task tokens.

\textbf{Compiler Toolchain for CGRA --} Figure~\ref{fig:procedure} shows the development procedure for ARENA using CGRA cluster as the backend. As already mentioned, the CGRA cluster is just one design choice; the ARENA execution model can be realized on alternative back-ends in case the HAF APIs (Table~\ref{tab:program_interface}) are realized. For example, on a CPU cluster backend, we can adopt MPI non-blocking-send primitives~\cite{gabriel2004open} to realize HAF APIs. We use this as one of our baselines in the evaluation. Here, to support the CGRA-cluster backend, an LLVM \cite{lattner2004llvm} based design automation toolchain is developed. In particular, a kernel that typically includes a multi-level nested loop is described as a task. As shown in Figure~\ref{fig:procedure}, to synthesize an appropriate mapping for a task on the allocated CGRA tiles, the nested loop is first vectorized with a factor that can fully leverage relatively larger CGRA tiles (e.g., 8x8 tiles) in the vectorization pass. Then, the remaining loops are flattened, generating the Control-Data Flow Graph (CDFG) representation, which is an extension of DFG with control dependence edges. We implemented a heuristic method~\cite{karunaratne2017hycube} to map the CDFG on various combinations of the tiles (i.e., 2x8, 4x8, and 8x8 tiles) and produce their control signals.

\section{Background}\label{sec:background}

\section{Motivation}\label{sec:background}
\sloppy

\begin{figure*}[htb]
	\centering
	\includegraphics[width=0.9\textwidth]{graphs/motivating1.pdf}
	\caption{Motivating example comparing traditional compute-centric execution model and ARENA data-centric execution model.}
	\label{fig:motivating}
\end{figure*}

Scientific simulation, where linear solvers iterate on data organized in dense (structured sparse) matrices or tensors, typically easy to divide in equally sized tiles, represents the premier HPC workload \cite{NWChem, GROMACS, quantummech, ExaCT, Carbon}. However, emerging HPC applications, targeting areas such as power grid dynamics~\cite{gridlab}, seismic risk assessment~\cite{Seismic}, urban systems simulation, and microbiome analysis~\cite{Genome}, will likely combine together traditional scientific simulation with advanced data analytics and machine learning. The datasets for these applications are much less structured, and thus more difficult to organize in \textbf{regular} and \textbf{partitionable} data structures. Applications will alternate phases of scientific simulations with \textbf{regular} behaviors, to phases where the computation happens on sparse data structures (e.g., sparse matrices, graph traversal) that induce unpredictable fine-grained data accesses and \textbf{irregular} behaviors. This scenario provides a clear opportunity for adaptability to diverse behaviors with reconfigurable hardware, while at the same time makes the current HPC programming models inadequate. In the following, we describe three major existing multi-node computing paradigms to motivate the ARENA design.

\subsection{Baseline-1: Compute-Centric BSP Execution Model}

HPC systems typically rely on the classical Bulk Synchronous Parallel (BSP) programming model in which a process is assigned to a processor or an entire node, and communication typically happens through message passing with libraries such as MPI. In the BSP model, the computation proceeds as a series of global supersteps: concurrent computation, where every participating nodes perform parallel computations on local data; communication, where nodes exchange data among them (with various, algorithm dependent, patterns); and barrier synchronization, to align execution of nodes.

The BSP model assumes that data are partitioned and distributed across nodes and rarely moves, to facilitate the local computation super-steps. Otherwise, the message-passing based communication super-steps would dominant the execution time. While this model works well for applications with easily partitionable data, regular computation, and limited, structured communication, it starts to experience significant limitations when workloads exhibit irregular behaviors (skewed data distributions, high synchronization intensity, irregular communication patterns). For these reasons, we consider the BSP model as \emph{Compute-Centric}.

Consider as an example an application with the (hierarchical) task graph in Figure~\ref{fig:motivating}(a), where the computation is split into 4 high-level tasks, each one executing task-partitioned computational kernels where the subtasks require data from other nodes. When employing a BSP model, both the data allocation and the distribution of the high-level tasks are fixed for the entire application execution. Hence, if a subtask needs data available in another node, it needs to initiate a communication phase, load the remote data, and synchronize to avoid hazards. Despite the latest high-performance designs can exploit mechanisms such as \emph{remote-direct-memory-access} (one-sided communication, not requiring a blocking receive with implicit synchronization from the remote node), prefetching and data migration, when these remote accessing are frequent, the bandwidth gap between local memory and remote memory (which needs to be accessed through the network) still remains the major performance (and consequently energy) concern. For example, if, as typical in HPC applications, tasks are executed in a loop, and they contend on the same data blocks (e.g., Tasks A, B, and C in Figure \ref{fig:motivating}), data migration may trigger even more data movement and synchronization, as the actual data distribution and access patterns are unknown before runtime.

\textbf{Limitations --} While compute-centric BSP long remains the standard model in HPC, its adoption in emerging HPC applications may be limited by data movement and synchronization. HPC practitioners have introduced asynchronous multi-task runtimes to tackle the limitations of the BSP model. Besides migrating data blocks where the computation occurs, these runtimes often allow tasks to migrate where the data reside, via approaches such as active messages and remote procedure calls, following the \emph{data-centric} models.

\comm{
\begin{figure*}[htb]
\minipage{0.45\textwidth}
  \includegraphics[width=\linewidth]{graphs/nonessential_data_movement.pdf}
  \caption{Nonessential data movement.}\label{fig:awesome_image2}
\endminipage\hfill
\minipage{0.45\textwidth}%
  \includegraphics[width=\linewidth]{graphs/category.pdf}
  \caption{Prgramming model and runtime.}\label{fig:awesome_image3}
\endminipage
\end{figure*}
}

\subsection{Baseline-2: Reconfigurable accelerators in HPC}

Reconfigurable accelerators have been deployed at a massive scale in data centers to provide application-specific acceleration with improved power-/area-efficiency~\cite{putnam2014reconfigurable, prabhakar2017plasticine, geng2018fpdeep}. Some institutions have also started hosting clusters with FPGA to perform research in HPC \cite{Paderborn}.

However, accelerators in HPC installations still generally adopt the BSP model, where a part of the local computation is offloaded to the accelerator itself. This also requires gathering desired data by the running tasks offloaded for execution. While FPGAs potentially allow acceleration of workloads more diverse than conventional accelerators such as GPUs, their current usage in HPC application often entails static configuration for accelerating a small amount of kernels (e.g., FFT, GEMM). This approach matches with an offload accelerator model - tasks and computational kernels do not move in the system. On the other hand, a reconfigurable architecture potentially allows to dynamically adjust the configuration at runtime, being able to accelerate divergent tasks with the computation proceeds. Despite the potential, the excessive reconfiguration overheads (typically in milliseconds) makes such an approach very costly.

\textbf{Limitations --} Reconfigurable accelerators make it possible to accelerate a more diverse workload, but current practice in heterogeneous HPC still leverages the offload model. The lack of low-latency runtime reconfiguration also limits the chances of large-scale task migration across the whole system. On the other hand, the data-centric execution model allows different tasks to work concurrently, despite using the same set of data in a node. CGRAs, as an alternative reconfigurable solution \cite{prabhakar2017plasticine} to FPGAs~\cite{de2019coarse}, offer significantly reduced reconfiguration time with coarser-grained reconfiguration, making flexible runtime architecture adjustment becomes possible.

\subsection{Baseline-3: Data-Centric Software Approaches}

Many programming models have been leveraged or designed to allow data-centric execution on multi-node systems. SSMP~\cite{perez2010handling} can operate on shared memory machines and support dynamic detection of dependencies between tasks. The implicitly shared memory management and dependence detection improve the programmability at the cost of increased synchronization overhead. \emph{Remote procedure call} (RPC) achieves near-data-computation based on prior knowledge about the exact distribution of data. X10~\cite{charles2005x10} and Chapel~\cite{chamberlain2007parallel} allow the programmers to control where to place the data and tasks. Similarly, Legion~\cite{bauer2012legion} enables explicit, programmer-controlled movement of data and placement of asynchronously spawned tasks, based on locality information. Legion employs a Cilk-like~\cite{blumofe1996cilk} algorithm for locality-aware task stealing.

Towards data-centric programming, MapReduce~\cite{dean2008mapreduce} programmers think in a data-centric fashion: they focus more on handling the sets of data records, rather than managing fine-grained threads, processes, communication, and coordination~\cite{alvaro2009boom}. However, MapReduce constrains its usage to batch-processing tasks, which falls in the BSP scope. To accommodate the emerging data-driven applications with irregular and unpredictable data access patterns, data-centric execution with asynchronous task-spawn should be enabled with hardware support. Meanwhile, the synchronization and task dependencies should be specified by the programmers to eliminate the unnecessary performance and energy overhead rather than forced by the programming model (e.g., remote procedure need to return to local in RPC, all the spawned tasks of the same ancestor need to join eventually in Legion). Besides the data locality, the runtime should also consider the computing resource utilization when reconfigurable accelerators are deployed and shared by multi-users in HPC/data-center environments.

\textbf{Limitations --} The high-level software framework and runtime facilitate the asynchronous execution of tasks and attempt to take advantage of data locality. Unfortunately, existing frameworks based on software solutions incur considerable overhead. The lack of hardware reconfiguration also limits the benefit from application-specific design and heterogeneity.

\subsection{Compute-Flow ARENA Execution Model}

We propose ARENA to address the limitation of the three aforementioned baseline. ARENA includes a novel programming model targeting asynchronous data-centric execution paradigm. As shown in Figure~\ref{fig:motivating}, all the configurable nodes in ARENA are connected by a ring network to bring the specialized computation to the data rather than the reverse to minimize data movement. Each reconfigurable node mainly contains a CGRA (detailed in Section~\ref{subsec:cgra}) that supports real-time reconfiguration and simultaneous execution of multi-tasks.

\section{Case Study}\label{sec:casestudy}
\sloppy
We present a case study to illustrate the features and impacts of the ARENA system architecture employed in an HPC cluster. We choose \textbf{breadth-first search} (BFS) kernel that solves the single source shortest path problem as our driving application kernel. The kernel traverses necessary vertices until the shortest paths from a source node to all the other nodes are found. In this case study, we represent a graph as an adjacency matrix that is distributed on an HPC cluster with a distributed memory system. Without loss of generality, we let each HPC node own $SIZE/NODES$ vertices (rows) of the entire graph (adjacency matrix of $SIZE$ x $SIZE$, $\infty$ indicates an edge while 0 indicates no connection).

\begin{figure}[!htb]
\begin{lstlisting}[basicstyle={\scriptsize\fontencoding{T1}\footnotesize\fontfamily{lmtt}\fontseries{c}\selectfont},escapechar=!]
int** local_M;
int num_visit;
...
int BFS_kernel(int vertex, int level) {
  for(int i=0; i<SIZE; ++i) {
    if(local_M[vertex][i] > level) {
      local_M[vertex][i] = level;
      visit_vertices[rank][num_visit] = i;
      num_visit += 1;
    }
  }
}
void run() {
  // Initialize local variables (e.g., local
  // partition of adjacency matrix: local_M)
  init();
  // Source vertex is launched on certain node
  if(local_start <= src and src < local_end) {
    num_visit = 0;
    visit_vertices[rank][num_visit] = src;
    num_visit += 1;
    level = 1;
  }
  while(1) {
    // Indicate #values will be broadcast
    Allgather(&num_visit, 1, num_all, 1);
    for(int i=0; i<NODES; ++i)
      if(num_all[i] > 0)
        // Broadcast the visited vertices
        Bcast(visit_vertices[i],num_all[i],i);

    for(int i=0; i<NODES; ++i)
      for(int j=0; j<num_all[i]; ++j)
        recv_q.push(visit_vertices[i][j]);

    num_visit = 0;
    if(recv_q.empty())
      break;
    while(recv_q.size()>0) {
      int v = recv_q.front();
      recv_q.pop();
      if(local_start <= v and v < local_end)
        BFS_kernel(v-local_bound, level);
      ++level;
    }
  }
}
\end{lstlisting}
\caption{A typical compute-centric BFS implementation.}
\label{fig:bfs_convention_impl}
\end{figure}

Conventional compute-centric BFS implementation requires considerable data communication and synchronization~\cite{bulucc2011parallel}. A typical implementation can be seen in Figure~\ref{fig:bfs_convention_impl}. Each HPC node only maintains the status (visited or not) of vertices it owns. Only the owner node of a vertice can identify whether it is newly visited or not. Thus, all the adjacencies of the vertices in the current frontier on a node need to be sent to their corresponding owner node, which incurs a set of all-to-all communications especially when each node is not aware of the global vertex distribution.

In contrast, ARENA enables an asynchronous data-centric execution model. Neither all-to-all data communication nor synchronization is needed. Task tokens are delivered via ARENA's ring topology and a task token will be picked by the node that contains the data range indicated by $TASK_{start}$ and $TASK_{end}$. Each HPC node then invokes the CGRA accelerator to perform the kernel execution. A typical data-centric implementation of BFS kernel is illustrated in Figure~\ref{fig:bfs_arena_impl}. \textit{ARENA\_spawn} can spawn a new task token, which will be dispatched by the current node itself or delivered to the next node on ARENA's ring. Once a task token is picked and executed by a node, multiple new tokens might be spawned. In this way, the entire graph will be traversed. We first evaluate the performance of BFS kernels in different execution models running on a CPU-only HPC cluster. Figure~\ref{fig:bfs_perf} shows that the data-centric execution model improves the performance with better scalability than the conventional compute-centric one.

\begin{figure}[htb]
	\centering
	\includegraphics[width=\columnwidth]{graphs/evaluate_bfs_scale.pdf}
	\caption{Performance of BFS kernels running on different architectures.}
	\label{fig:bfs_perf}
\end{figure}

We also evaluate the ARENA with the hardware support of the dispatcher and CGRA. Our compiler toolchain can generate control signals of the ARENA task for different CGRAs. In addition, the compiler can lower the \textit{ARENA\_spawn} function as a $spawn$ operation supported by our CGRA's functional unit (four nodes in green color at the right most column as shown in Figure~\ref{fig:arena_node}).

CGRA can accelerates the execution of BFS by exacting the data parallelism and pipelining loop iterations. The 8x8 CGRA can be partitioned into four 2x8 CGRA groups, which enables different normalized speedups for BFS kernel supported by our compiler: 8.2X for a 2x8 CGRA, 16.4X for a 4x8 CGRA, and 21.8X for a 8x8 CGRA compared with an in-order execution without any acceleration (i.e., single cycle per operation). Note that the speedup of single 8x8 CGRA are less than the speedup of four 2x8 CGRAs (i.e., 21.8 $<$ 4 $\times$ 8.2). In conventional compute-centric model, entire 8x8 CGRA should be used to maximize the speedup of the level-synchronous BFS kernel (Figure~\ref{fig:bfs_convention_impl}). However, ARENA's asynchronous data-centric model efficiently utilizes the CGRA computing resources based on the required data range of the target task (discussed in Section~\ref{subsec:cgra}). To be specific, one group of 2x8 CGRA tiles are allocated to each data-centric BFS kernel (Figure~\ref{fig:bfs_arena_impl}). Since new tasks are spawned asynchronously at runtime, at most four groups can accelerate four BFS kernels simultaneously, which improves the overall speedup (as shown in Figure~\ref{fig:bfs_perf}) and the CGRA's area-efficiency.

\todo[inline]{briefly mention DNA, GCN user cases.}

\todo[inline]{mention multi-user application scenario.}
\section{Conclusion}\label{sec:conclusion}
\sloppy

In this paper, we propose an asynchronous-reconfigurable-accelerator-ring architecture for next generation data-driven high-performance computing. Through the co-design of architecture and programming model, ARENA is able to bring computation tasks in the form of CGRA-specialized hardware accelerators to the data, rather than the reverse as in contemporary compute-centric and dataflow architectures, significantly improving performance and reducing data-movement.
\section{Evaluation}\label{sec:evaluation}
\sloppy

\subsection{Environment Setup}

The ARENA runtime is evaluated on traditional CPU HPC clusters and our proposed CGRA-based ARENA cluster. For the latter, we extend the \emph{Structural Simulation Toolkit} (SST)~\cite{rodrigues2016structural} to model a multi-node cluster based on MPI. We model the network topology and package transmit switch in SST using the MACRELS Analytic Model~\cite{sstnetwork}. The token transmit network is modeled as 1D Torus Ring. We implement the task dispatcher, CGRA, and CGRA controller in PyMTL~\cite{lockhart2014pymtl}, which can report cycle-accurate simulation results for a single node. To obtain the cycle-level simulation result, we implement the dispatcher interface in SST to handle the task tokens. Finally, we feed the single-node result to SST and generates synthesizable Verilog for power, area, and timing analysis. The detailed simulation parameters are listed in Table \ref{tab:simulation_parameter}.

\begin{table}[htb!]
\centering
\footnotesize
\begin{tabular}{cl}
\hline\hline
  \textbf{Technology} &
  45nm \\ \hline
  \textbf{\begin{tabular}[c]{@{}c@{}}Network Interface\end{tabular}} &
  80 Gb/s \\ \hline
  \textbf{\begin{tabular}[c]{@{}c@{}}Network Topology\end{tabular}} &
  1D Torus Ring \\ \hline
   \textbf{\begin{tabular}[c]{@{}c@{}}Network Switch\end{tabular}} &
  \begin{tabular}[c]{@{}l@{}}1 per node, 1us hop latency\end{tabular} \\ \hline
  \textbf{Dispatcher} &
  \begin{tabular}[c]{@{}l@{}}Filter logic, 8-entry receive queue,
  8-entry \\wait queue, 8-entry send queue\end{tabular} \\ \hline
  \begin{tabular}[c]{@{}c@{}}\textbf{CPU} (baseline)\end{tabular} & \begin{tabular}[c]{@{}l@{}}2.6GHz, 20MB 3-level Cache, Out-of-order, x86\end{tabular} \\ \hline
  \textbf{CGRA} &
  \begin{tabular}[c]{@{}l@{}}8 $\times$ 8 CGRA, 480-byte control memory per tile, \\2-bank 4-port 32KB scratchpad data memory,\end{tabular} \\ \hline
  \textbf{\begin{tabular}[c]{@{}l@{}}CGRA Controller\end{tabular}} &
  \begin{tabular}[c]{@{}l@{}}4 $\times$ 4-entry queue for spawned task tokens, \\Coalescing unit\end{tabular} \\
  \hline\hline
\end{tabular}
\caption{RTL and simulation parameters for ARENA.}
\label{tab:simulation_parameter}
\end{table}

\textbf{Applications --} We evaluate ARENA using representative HPC and data-analytics workloads: The single-source shortest paths (\textbf{SSSP}) problem (see Section~\ref{subsec:programming}) is a key subroutine in many data-intensive graph computations.  General Matrix Multiply (\textbf{GEMM}) is the core function of linear algebra and deep learning workloads. We assume the matrices are distributed among nodes. Sparse-matrix-vector multiplication (\textbf{SPMV}) is the fundamental kernel in many scientific \& data applications. Here, the distributed matrix is in the Compressed Sparse Row (CSR) format. DNA sequence alignment (\textbf{DNA}) \cite{NBCIBLAST} leverages Needleman-Wunsch (NW) algorithm to search the best-matched protein sequences with respect to the target pattern. A Graph convolutional network \textbf{GCN}~\cite{fey2019fast} inference application on the Cora dataset is also evaluated, representing emerging irregular machine learning workloads. We assume the adjacency and feature matrices are distributed among nodes. Finally, an \textbf{N-body} simulation application \cite{green2010particle} for simulating dynamical particle system is demonstrated, representing traditional scientific simulation workloads. Again, the particle information (e.g., accelerations, velocity, position, collision, etc) are distributively stored and required to be updated per iteration at runtime.

The conventional compute-centric parallel implementations of all the evaluated applications are developed based on state-of-the-art algorithms or derived from the widely-used benchmark suites~\cite{pouchet2012polybench, che2009rodinia}. For example, the SSSP application is implemented based on~\cite{bulucc2011parallel}. The GCN model is extracted from PyTorch Geometric~\cite{fey2019fast}. The DNA application leverages the NW algorithm from Rodinia~\cite{che2009rodinia}. Regarding the ARENA implementation, all the applications are programmed following the ARENA programming models for data-centric asynchronous execution with runtime hardware specialization. Note that GCN and NBody contain numeric distinct functional tasks.

\begin{figure*}[htb!]
	\centering
	\includegraphics[width=0.95\textwidth]{graphs/evaluation_data_centric_speedup1.pdf}
	\caption{Normalized speedup for compute-centric and ARENA's data-centric execution models running on different multi-CPUs cluster w.r.t. a serial implementation on a single node.}
	\label{fig:data_centric_speedup}
\end{figure*}

\begin{figure}[htb!]
	\centering
	\includegraphics[width=0.95\columnwidth]{graphs/data_movement1.pdf}
	\caption{{Normalized data movement breakdown in data-centric model w.r.t. the compute-centric model.}}
	\label{fig:arena_utilization}
\end{figure}

\subsection{Evaluation Results}\label{subsec:results}

We show the benefits of the data-centric programming model, CGRA hardware acceleration, and the entire ARENA system.

\textbf{Programming Model --} We first show the performance effectiveness of ARENA's data-centric programming model. Figure~\ref{fig:data_centric_speedup} illustrates the normalized speedups for conventional compute-centric and ARENA's models (Both are software implemented based on MPI) with respect to a serial implementation on a single CPU node (i.e., baseline). As can be seen, ARENA's data-centric execution model shows higher speedups and better scalability in general. This is mainly due to the elimination of synchronization, and the minimization of data communication. On average, ARENA's data-centric model outperforms the compute-centric counterpart by 1.61$\times$ (i.e., 7.82/4.87) in a 16-node cluster. Specifically, the kernels with higher data parallelism (e.g. SSSP, GEMM, and SPMV) can gain better scalability for both models; for kernels with limited data parallelism such as DNA, the compute-centric model exhibits lower scalability due to massive data dependency and costly remote communication. In this condition, ARENA achieves better scalability by streaming task tokens over the nodes to minimize data movement.

\begin{figure*}[htb!]
	\centering
	\includegraphics[width=0.95\textwidth]{graphs/evaluation_ARENA_speedup1.pdf}
	\caption{Normalized speedup for compute-centric and ARENA's data-centric execution models running on different multi-CGRAs cluster w.r.t. a serial implementation on a single node without any acceleration.}
	\label{fig:arena_speedup}
\end{figure*}

Figure~\ref{fig:arena_utilization} illustrates the normalized data movement breakdown for ARENA's data-centric model with respect to the compute-centric model for all applications in a 4-node cluster. Compared with the compute-centric HPC cluster, ARENA can eliminate on average 53.9\% data movement without any prior knowledge about the data distribution, leading to substantial improvement in energy efficiency. We also observe different data movement patterns across applications. For example, SSSP posts considerable task movement, as it spawns massive fine-grained tasks with discrete data-ranges, which are hard to coalesce. Regarding DNA, the compute-centric implementation is based on OpenMP where all threads are sharing the same copy in global memory~\cite{che2009rodinia}. The sub-blocks workload distribution to threads following a zig-zag manner incurs frequent data movement. In ARENA, the data dependency only exits on the edge of the sub-block, which can be explicitly labeled by the parent tasks using the ARENA User-APIs (i.e., $REMOTE_{start}$ and $REMOTE_{end}$ in ARENA\_task\_spawn()), therefore minimizing data movement.
Finally, GEMM and NBody comprise coarse-grained tasks and the task-flows require data streaming among the nodes, leading to little task movement or essential data movement as shown in the figure.

\textbf{CGRA Speedup --} Figure~\ref{fig:cgra_speedup} shows the normalized speedup of the evaluated applications running on different configurations or combinations of ARENA's CGRA tile groups (each group is a 2x8 CGRA) with respect to the single node CPU baseline. In general, a larger CGRA tile configuration leads to higher speedups. For DNA, however, the loop-carried data dependency limits the data parallelism, as well as the obtainable speedups (1.7$\times$ speedup at most). On average, 1.3$\times$, 2.4$\times$, and 3.5$\times$ speedups are achieved by ARENA's 2x8, 4x8, and 8x8 CGRA across all the applications and kernels. The 2x8 CGRA exhibits the optimal area-efficiency, showcasing the advantages of runtime hardware specialization.

\begin{figure}[htb!]
	\centering
	\includegraphics[width=0.95\columnwidth]{graphs/evaluation_cgra_speedup1.pdf}
	\caption{Normalized CGRA speedup w.r.t. the single node baseline CPU execution without any acceleration.}
	\label{fig:cgra_speedup}
\end{figure}

\textbf{Overall System --} The normalized speedup of ARENA is shown in Figure~\ref{fig:arena_speedup}. As can be seen, ARENA maximizes the overall performance by leveraging a data-centric execution model and CGRA in a synergistic and integrated fashion. Instead of fixing the CGRA configuration for each workload, ARENA dynamically allocates and configures the CGRA tiles specifically for a particular task based on the task-carried specification (obtained based on data requirement of the task), as well as the current CGRA resource availability. On average, the compute-centric execution model using the entire CGRAs for each kernel obtains 10.06$\times$ speedup on a 16-node cluster, whereas  ARENA achieves 21.29$\times$ speedup. In other words, ARENA is 2.17$\times$ better than the compute-centric with CGRA support). This implies ARENA can leverage the CGRAs in a more efficient way. Compared with Figure~\ref{fig:data_centric_speedup}, we can see that ARENA with CGRAs also gains better scalability (from 1.61$\times$ to 2.17$\times$ on a 16-node cluster). The performance of DNA does not improve much due to limited acceleration from CGRA. Finally, the compute-centric execution of GEMM does not scale well because synchronization over a larger amount of data creates serious performance bottlenecks.

\subsection{RTL Timing, Area and Power}

We evaluate the timing, area, and power consumption of ARENA's CGRA-cluster (e.g., CGRAs, CGRA controllers and task dispatchers) using the synthesized Verilog HDL code from PyMTL. We use Synopsys Design Compiler, Cadence Innovus, and Synopsys PrimeTime PX in order to synthesize, place, route, and estimate the power consumption of the designs. We use FreePDK45 with the Nangate standard cell library. Figure~\ref{fig:layout} illustrates the obtained chip layout. The area and power of the 32KB scratchpad data memory are estimated based on CACTI-6.5. The chip area is 2.19mm x 1.24mm and the operating frequency is 800MHz @ 45nm with on average 759.8mW power consumption.
\begin{figure}[htb!]
	\centering
	\includegraphics[width=0.99\columnwidth]{graphs/layout_data.pdf}
	\caption{{Chip layout of a single ARENA node.}}
	\label{fig:layout}
\end{figure}

\section{Introduction}\label{sec:intro}
\vspace{-0.1cm}
\sloppy

With the slowing down of Moore's Law \cite{moore1965cramming}, future computer systems will need to resort to domain-specific accelerators for continuous performance scaling \cite{shalf2020future, dongarra2020numerical} under the same power envelope. This is especially the case for HPC and data centers, as we are quickly entering an era of extreme heterogeneity \cite{vetter2018extreme}, characterized by cluster nodes integrating a multitude of cooperating accelerators \cite{putnam2014reconfigurable, hazelwood2018applied, jouppi2017datacenter, lee2017introducing}.

While integrating domain-specific accelerators (DSAs) into HPC and data centers provides considerable efficiency gains \cite{hazelwood2018applied, jouppi2017datacenter, shaw2014anton, ohmura2014mdgrape}, it leads to enormous complexities as well. First, DSAs can only be economically designed for ubiquitous computational patterns in applications, while the workloads currently running in HPC and data centers are converging towards mixed workflows that include scientific simulation, machine learning, data analytics, etc. Second, managing various (typically loosely coupled) accelerators across many nodes significantly complicates programming models and the software infrastructure. In both HPC and data centers, the accelerators typically need to be shared among multiple users, often using multiple nodes for applications with divergent characteristics. Since we could not design hardware accelerators for all seen and unseen kernels, reconfigurable architecture, which allows specialization after system deployment and even during system execution, promises to be a wise solution. While Field Programmable Gate Arrays (FPGAs) have already been deployed at large scale in some data centers \cite{putnam2014reconfigurable}, they may suffer from limited frequency and energy-efficiency compared with ASICs, as well as long reconfiguration time (e.g. in milliseconds) due to bit-level reconfigurability. Coarse-grained reconfigurable arrays (CGRAs), which integrate highly optimized functional units (rather than fundamental lookup tables or LUTs) and offering reconfigurability at the word-level, emerge as a promising alternative choice \cite{prabhakar2017plasticine}. The rapid reconfiguration \cite{de2019coarse} makes dynamic formalization of hardware accelerators at runtime becomes feasible, and even plausible.

Conventional large-scale HPC clusters implicitly assuming homogeneous node-configuration and bulk-synchronous-parallel (BSP) execution model suffer from three low-utilization challenges regarding per-node data locality: (1) unbalanced data distribution among homogeneous nodes may lead to unbalanced workload and poor utilization; (2) If data is not locally available, during the long-time remote data fetching, the compute units can be idle; (2) Even worse, these idle units or idle nodes cannot be reclaimed for other tasks despite the node may hold their desired data. As emerging workloads become more dynamic and data-driven, decentralized asynchronous task management is highly desired, while data locality becomes a crucial factor for the system design \cite{boroumand2018google, shalf2020future, dongarra2020numerical}. This is largely due to the observation that the energy cost of data-movement significantly overweights the energy cost of computing them \cite{horowitz20141, kestor2013quantifying, molka2010characterizing}, which is particularly the case when migrating data through the interconnect network (e.g., the power budget is $\sim$5.5 watts per full bi-directional NVLink port \cite{nvlink_power}). While software solutions such as \emph{active messages} \cite{bonachea2017gasnet, bonachea2018gasnet} and remote procedure call (RPC) mitigate the problem by pushing computation to the nodes where data resides, they suffer from considerable overhead due to the lack of architectural support. Existing applications however still widely adopt the BSP model \cite{valiant1990bridging, bisseling1993scientific} alternating phases of (parallel) local computation with phases of global communication. Note that the BSP model implicitly assumes that the majority of the time can be spent in easily parallelizable computation phases, with limited data movement. However, the exponential growth in the availability of data and the emergence of new applications, radically changed the balance.

\begin{figure}[htb!]
	\centering
	\includegraphics[width=0.8\columnwidth]{graphs/stack.pdf}
	\caption{Overall ARENA design stack.}
	\label{fig:stack}
\end{figure}

Bridging coarse-grained reconfigurable architecture with locality-oriented asynchronous parallel task execution, in this paper, we propose ARENA --- an \underline{A}synchronous \underline{RE}co\underline{N}figurable \underline{A}ccelerator ring architecture \& runtime, to enable data-centric computation flow paradigm aiming at significantly reducing unnecessary inter-node data movement. ARENA comprises multiple CGRAs interconnected in a ring to provide quick reconfiguration at runtime. The tasks of the application, tagged as task tokens, are injected into the ring when dependencies are satisfied. Task-specialized CGRA accelerators can be constructed at runtime using the specification embedded in the task token. When task tokens are streaming around the ring, each node can verify whether a task should be executed locally (based on data locality and resource availability). In other words, ARENA is dynamically sharing hardware resource according to data locality. This is in contrast with conventional systems sharing time-slots according to hardware availability. In summary, this paper thus makes the following contributions:

\begin{itemize}
    \item \textbf{Architecture:} we proposed a multi-CGRA cluster architecture (Section~\ref{sec:architecture}) that can be dynamically reconfigured at runtime, enabling asynchronous parallel execution for multi-users with very little overhead.
    \item \textbf{Runtime:} we proposed a flexible runtime (Section~\ref{subsec:runtime}) enabling task-token delivery along the ring network based on data locality. Compared to compute-centric execution models, the proposed runtime can significantly mitigate data movement and improve performance.
    \item \textbf{Programming Model:} we proposed a data-centric programming model (Section~\ref{subsec:programming}) with easy-to-use APIs to facilitate the programming for ARENA architecture and runtime. An LLVM-based compiler toolchain is constructed to support the programming model.
    \item \textbf{Evaluation:} RTL simulation results using practical HPC and data analytics applications (Section~\ref{sec:evaluation}) show that ARENA can provide better parallel scalability with reduced data movement (53.9\%) and bring 2.17$\times$ and 4.37$\times$ speedup over traditional compute-centric approach with and without CGRA acceleration. The CGRA-based ARENA prototype only costs 2.93$mm^2$ chip area in 45nm and can work at 800MHz with 759.8mW power consumption per node, showing significant advantages over the present architectural design in the current HPC and data centers.
\end{itemize}

\section{ARENA Programming Model}\label{sec:programming}

Being the interface between software and hardware, the ARENA programming model defines a list of API functions in Table~\ref{tab:program_interface}. On one hand, in order to program an ARENA abstract machine, a software programmer has to define their user-logic as task functions, and rely on the \emph{User APIs} to operate the abstract machine. Please note that although in this work we use CGRAs as the hardware testbed, it is only one of the the possible instantiations of the ARENA abstract machine model (AMM). On the other hand, in order to support ARENA software and run ARENA program, an alternative architecture has to support the \emph{Hardware Abstract Functions} here.

\begin{table}[htb!]
\scriptsize
\centering
\begin{tabular}{l|l}
\bottomrule\bottomrule
\textbf{Function \& Construct} & \textbf{Description} \\ \toprule
\multicolumn{2}{c}{\textbf{User Defined Function}} \\ \bottomrule
\begin{tabular}[c]{@{}l@{}}void \textbf{my\_task}(\\\hspace{0.7cm}\textbf{Address} \textit{TASKstart},\\\hspace{0.7cm}\textbf{Address} \textit{TASKend},\\\hspace{0.7cm}\textbf{float} \textit{PARAM})\end{tabular}& \begin{tabular}[c]{@{}l@{}}A user can implement multiple different \\tasks to compose a single or multiple \\applications. \end{tabular}\\ \toprule
\multicolumn{2}{c}{\textbf{User APIs}} \\ \bottomrule
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_task\_register}(\\\hspace{0.7cm}\textbf{int} \textit{TASKid},\\\hspace{0.7cm}\textbf{Address} \&\textit{my\_kernel},\\\hspace{0.7cm}\textbf{bool} \textit{isRoot})\end{tabular}& \begin{tabular}[c]{@{}l@{}} Registers a kernel (e.g. \textit{my\_kernel}) with\\ \textit{TASKid}. The root task is launched by a \\CPU or a microcontroller once the system \\starts to run. \end{tabular}\\ \hline
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_task\_spawn}(\\\hspace{0.7cm}\textbf{int} \textit{TASKid},\\\hspace{0.7cm}\textbf{Address} \textit{TASKstart},\\\hspace{0.7cm}\textbf{Address} \textit{TASKend}, \\\hspace{0.7cm}\textbf{float} \textit{PARAM}, \\\hspace{0.7cm}\textbf{Address} \textit{REMOTEstart}, \\\hspace{0.7cm}\textbf{Address} \textit{REMOTEend})\end{tabular}& \begin{tabular}[c]{@{}l@{}}Dynamically spawns a new \textbf{TaskToken} \\(\textit{FROMnode} is automatically applied) that \\will be issued to the \textbf{CoalesceUnit}. The \\fields of a task token is explained in detail \\in Section~\ref{subsec:token}. \end{tabular}\\  \toprule
\multicolumn{2}{c}{\textbf{Hardware Abstract Functions}} \\ \bottomrule
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_init}(\\\hspace{0.7cm}\textbf{Address*} \textit{local\_start},\\\hspace{0.7cm}\textbf{Address*} \textit{local\_end})\end{tabular}& \begin{tabular}[c]{@{}l@{}}Initializes \textit{local\_start} and \textit{local\_end} based \\on local data information. \end{tabular}\\ \hline
\begin{tabular}[c]{@{}l@{}}{TaskToken} \textbf{ARENA\_arrive}()\end{tabular}& \begin{tabular}[c]{@{}l@{}}Receives an incoming task from a remote \\node.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_filter}(\\\hspace{0.7cm}\textbf{TaskToken} \textit{token},\\\hspace{0.7cm}\textbf{Address*} \textit{local\_start},\\\hspace{0.7cm}\textbf{Address*} \textit{local\_end},\\\hspace{0.7cm}\textbf{TaskQueue*} \textit{SendQueue},\\\hspace{0.7cm}\textbf{TaskQueue*} \textit{WaitQueue})\end{tabular}& \begin{tabular}[c]{@{}l@{}}Detaches, Splits or passes tasks based on \\the \textit{token}'s required data addresses (by \\comparing them with \textit{local\_start} and \\\textit{local\_end}).\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}bool \textbf{ARENA\_ready}(\\\hspace{0.7cm}\textbf{TaskToken} \textit{token})\end{tabular}& \begin{tabular}[c]{@{}l@{}}Checks if the computing resources are \\available for executing the task \textit{token}.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_launch}(\\\hspace{0.7cm}\textbf{TaskToken} \textit{token},\\\hspace{0.7cm}\textbf{TaskQueue*} \textit{CoalesceUnit})\end{tabular}& \begin{tabular}[c]{@{}l@{}}Issues a task denoted by \textit{token} either to a \\CPU or to a reconfigurable accelerator (e.g., \\CGRA). The spawned new tasks will be \\pushed into CoalesceUnit.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_data\_acquire}(\\\hspace{0.7cm}\textbf{TaskToken} \textit{token}) \end{tabular}& \begin{tabular}[c]{@{}l@{}}Acquires additional data from remote node \\via NIC.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}TaskToken \textbf{ARENA\_coalesce}(\\\hspace{0.7cm}\textbf{TaskQueue*} \textit{CoalesceUnit})\end{tabular}& \begin{tabular}[c]{@{}l@{}}Coalesces spawned tasks with continuous \\data range, identical required remote data \\range, and identical PARAM.\end{tabular}
\\ \toprule
\multicolumn{2}{c}{\textbf{Base Constructs}} \\ \bottomrule
\begin{tabular}[c]{@{}l@{}}\textbf{TaskToken} \end{tabular} & \begin{tabular}[c]{@{}l@{}}encapsulates a task.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}\textbf{Address} \end{tabular} & \begin{tabular}[c]{@{}l@{}}Local address of data.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}\textbf{TaskQueue} \end{tabular} & \begin{tabular}[c]{@{}l@{}}Buffers for task tokens.\end{tabular} \\ \toprule\toprule
\end{tabular}
\caption{ARENA programming and hardware APIs.}
\label{tab:program_interface}
\end{table}

\comm{
\begin{table}[htb!]
\small
\centering
\begin{tabular}{l|l}
\bottomrule\bottomrule
\textbf{Function \& Construct} & \textbf{Description} \\ \toprule
\multicolumn{2}{c}{\textbf{User Defined Function}} \\ \bottomrule
\begin{tabular}[c]{@{}l@{}}void \textbf{my\_task}(\\\hspace{0.7cm}\textbf{Address} \textit{TASKstart},\\\hspace{0.7cm}\textbf{Address} \textit{TASKend},\\\hspace{0.7cm}\textbf{float} \textit{PARAM})\end{tabular}& \begin{tabular}[c]{@{}l@{}}A user can implement \\ multiple tasks to compose \\a single or multiple \\applications. \end{tabular}\\ \toprule
\multicolumn{2}{c}{\textbf{User APIs}} \\ \bottomrule
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_task\_register}(\\\hspace{0.7cm}\textbf{int} \textit{TASKid},\\\hspace{0.7cm}\textbf{Address} \&\textit{my\_kernel},\\\hspace{0.7cm}\textbf{bool} \textit{isRoot})\end{tabular}& \begin{tabular}[c]{@{}l@{}} Registers a kernel (e.g. \\ \textit{my\_kernel}) with \textit{TASKid}.\\ Root task is launched once\\ the system starts to run. \end{tabular}\\ \hline
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_task\_spawn}(\\\hspace{0.7cm}\textbf{int} \textit{TASKid},\\\hspace{0.7cm}\textbf{Address} \textit{TASKstart},\\\hspace{0.7cm}\textbf{Address} \textit{TASKend}, \\\hspace{0.7cm}\textbf{float} \textit{PARAM}, \\\hspace{0.7cm}\textbf{Address} \textit{REMOTEstart}, \\\hspace{0.7cm}\textbf{Address} \textit{REMOTEend})\end{tabular}& \begin{tabular}[c]{@{}l@{}}Dynamically spawns a new \\\textbf{TaskToken} (\textit{FROMnode} \\is automatically applied)\\ that will be issued to the \\ \textbf{CoalesceUnit}. The fields \\of a task token is detailed \\in Section~\ref{subsec:token}. \end{tabular}\\  \toprule
\multicolumn{2}{c}{\textbf{Hardware Abstract Functions}} \\ \bottomrule
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_init}(\\\hspace{0.7cm}\textbf{Address*} \textit{local\_start},\\\hspace{0.7cm}\textbf{Address*} \textit{local\_end})\end{tabular}& \begin{tabular}[c]{@{}l@{}}Initializes \textit{local\_start} and \\\textit{local\_end} based on local \\data information. \end{tabular}\\ \hline
\begin{tabular}[c]{@{}l@{}}{TaskToken} \textbf{ARENA\_arrive}()\end{tabular}& \begin{tabular}[c]{@{}l@{}}Receives an incoming task \\from a remote node.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_filter}(\\\hspace{0.7cm}\textbf{TaskToken} \textit{token},\\\hspace{0.7cm}\textbf{Address*} \textit{local\_start},\\\hspace{0.7cm}\textbf{Address*} \textit{local\_end},\\\hspace{0.7cm}\textbf{TaskQueue*} \textit{SendQueue},\\\hspace{0.7cm}\textbf{TaskQueue*} \textit{WaitQueue})\end{tabular}& \begin{tabular}[c]{@{}l@{}}detaches, Splits or passes \\tasks based on the \textit{token}'s\\ required data addresses (by \\comparing them with \\\textit{local\_start} and \textit{local\_end}).\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}bool \textbf{ARENA\_ready}(\\\hspace{0.7cm}\textbf{TaskToken} \textit{token})\end{tabular}& \begin{tabular}[c]{@{}l@{}}Checks if the computing \\resources are available for\\executing the task \textit{token}.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_launch}(\\\hspace{0.7cm}\textbf{TaskToken} \textit{token},\\\hspace{0.7cm}\textbf{TaskQueue*} \textit{CoalesceUnit})\end{tabular}& \begin{tabular}[c]{@{}l@{}}Issues a task denoted by \\\textit{token} either to a CPU or to\\a reconfigurable accelerator \\(e.g., CGRA). The spawned \\new tasks will be pushed\\ into CoalesceUnit.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}void \textbf{ARENA\_data\_acquire}(\\\hspace{0.7cm}\textbf{TaskToken} \textit{token}) \end{tabular}& \begin{tabular}[c]{@{}l@{}}Acquires additional data \\from remote node via NIC.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}TaskToken \textbf{ARENA\_coalesce}(\\\hspace{0.7cm}\textbf{TaskQueue*} \textit{CoalesceUnit})\end{tabular}& \begin{tabular}[c]{@{}l@{}}Coalesces spawned tasks \\with continuous data range,\\same required remote data \\range, and same PARAM.\end{tabular}
\\ \toprule
\multicolumn{2}{c}{\textbf{Base Constructs}} \\ \bottomrule
\begin{tabular}[c]{@{}l@{}}\textbf{TaskToken} \end{tabular} & \begin{tabular}[c]{@{}l@{}}encapsulates a task.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}\textbf{Address} \end{tabular} & \begin{tabular}[c]{@{}l@{}}Local address of data.\end{tabular} \\ \hline
\begin{tabular}[c]{@{}l@{}}\textbf{TaskQueue} \end{tabular} & \begin{tabular}[c]{@{}l@{}}Buffers for task tokens.\end{tabular} \\ \toprule\toprule
\end{tabular}
\caption{ARENA Programming and Hardware APIs.}
\label{tab:program_interface}
\end{table}
}
\subsection{ARENA Programming Interface}\label{subsec:programming}

To program an ARENA abstract machine, the programmers first partition their application into tasks and register the defined tasks to the ARENA runtime. Ideally, the partition can separate the working set into a bunch of continuous data segments, where each task accounts for a segment. ARENA does not limit the granularity of a task, which can be extremely fine-grained or coarse-grained. While ARENA works perfectly when the data is locally available for a task, we understand this is not always feasible. When remote data access is inevitable, the application can either spawn a new task for the remote data, or explicitly initiate the data-movement through the \emph{data-transfer-network}.

\begin{figure}[htb]
\begin{lstlisting}[basicstyle={\scriptsize\fontencoding{T1}\footnotesize\fontfamily{lmtt}\fontseries{c}\selectfont},escapechar=!]
// Users can define multiple different kernels,
// each with a specific task token ID.
#define BFS_TOKEN 1
int** local_M;
...
void BFS_kernel(int TASKstart, int TASKend,
                int PARAM) {
  int level = PARAM;
  for(int i=TASKstart; i<TASKend; ++i) {
    for(int j=0; j<SIZE; ++j) {
      if(local_M[i][j] > level) {
        local_M[i][j] = level;
        ARENA_task_spawn(BFS_TOKEN, j, j+1,
                         level+1); }}}}
void run() {
  // Register kernels.
  ARENA_task_register(BFS_TOKEN, &BFS_kernel, true);
  // Launch ARENA runtime.
  ARENA_runtime(); }
\end{lstlisting}
\caption{Example of programming SSSP in ARENA.}
\label{fig:bfs_arena_impl}
\end{figure}

Figure~\ref{fig:bfs_arena_impl} shows an example on how to solve the single-source shortest path (SSSP) problem using a breadth-first search (BFS) kernel. The design traverses associated vertices until the shortest path(s) from a source node to all the other nodes are found. Without losing generality, we assume the graph, represented as an adjacency matrix, is distributively stored on all nodes and each node holds $SIZE/NODES$ vertices (rows) of the entire graph (the adjacency matrix is in $SIZE$ x $SIZE$, an initial value of $\infty$ indicates a connected edge while 0 implies no connection).

ARENA enables asynchronous data-centric execution by dynamically spawning new task-tokens among the nodes. Currently, all tasks need to be registered at the beginning. During runtime, task-tokens are circulating among all nodes in the ring. In case a node confirms it has the data required by a task-token (indicated by the starting \& ending addresses $TASK_{start}$ and $TASK_{end}$) as well as sufficient hardware resources for runtime hardware specialization implied by the task-token, it takes out the token from the task-token-stream and executes it. A task will be executed by the CPU if no hardware specialization is provided. New tasks for remote nodes can be generated or spawned at any node. We currently rely on the programmer to determine the granularity of a spawned task (in other words, the data-range a spawned task designated). Fine-grained tasks facilitate asynchronous execution but increase scheduling overhead in the runtime. We discuss this tradeoff in detail in Section~\ref{subsec:runtime}.

This is in contrast with the conventional compute-centric approach demanding frequent data communication and synchronization~\cite{bulucc2011parallel}. As each node maintains the vertex status locally, and no prior knowledge about vertex distribution is asserted, repeated all-to-all communications are essentially desired for broadcasting vertex updating information to associated nodes on the present frontier. Figure~\ref{fig:data_centric_speedup} shows the performance gain of ARENA for the SSSP application.

\section{Related Work}\label{sec:related}

We summarize related work regarding the ring network, clusters of reconfigurable architecture, and the task-based execution model in a ring network.

\textbf{Ring Network.}
The ring network has been adopted in real multi-processors design \cite{barroso1993performance} (e.g., Intel Xeon-Phi \cite{chrysos2012intel}) and logically studied for efficient collective communication \cite{liu2015imr,awan2016efficient,alazemi2018routerless, li2019evaluating,wang2019achieving,lin2020deep}. On the one hand, the ring network provide simple routing mechanism to better-utilized the link bandwidth for fast communication \cite{barroso1993performance}. on the other hand, the ring network has been criticized for easy saturation due to linear increased latency with more node \cite{ainsworth2007characterizing, jerger2009chip, liu2015imr}. Previous works \cite{liu2015imr,alazemi2018routerless, wang2019achieving,lin2020deep} extend the ring concept to form local-rings network among a subset of nodes, known as routerless network. In ARENA, we avoid the saturation problem through: (a) a routerless task execution model among nodes; and (b) the dynamically task allocation and dispatching mechanism.

\textbf{Reconfigurable Hardware Cluster.}
Clusters incorporating reconfigurable devices such as FPGAs \cite{putnam2014reconfigurable, dondo2015integrating, knodel2016rc3e, tarafdar2017enabling} and CGRAs \cite{prabhakar2017plasticine, jin2012implementation, HammerBlade} have already been showcased by existing works \cite{lyke2015introduction, tessier2015reconfigurable}. On one hand, Putnam et al. from Microsoft propose the reconfigurable Catapult fabric \cite{putnam2014reconfigurable}, where each instantiation consists of a 6x8 2D torus of Xilinx Stratix-V FPGAs. Every FPGA is connected to a CPU server via PCI-e and directly links to other FPGAs through the SAS cables. This 1632-node FPGA cluster has been adopted for document ranking of the Bing search engine. Nearly 95\% performance increase has been demonstrated with only a 10\% extra power budget. However, each FPGA in Catapult is specialized to a single application kernel during runtime. Reconfiguration takes several milliseconds. Even reloading models without changing the computation logic can take up to 250 microseconds. On the other hand, the data-centric execution model allowing different tasks working on the same set of data in an HPC node requires rapid dynamic reconfiguration. Gazzano et al. propose R-Grid, a complete grid infrastructure for distributed high-performance computing using dynamically reconfigurable FPGAs \cite{dondo2015integrating}. Knodel et al. virtualize the FPGA resources and propose adapted service models in a cloud context \cite{knodel2016rc3e}. Tarafdar et al. discussed how FPGAs of a cloud data center can be flexibly connected based on a logical kernel and a mapping file \cite{tarafdar2017enabling}. Zhang et al. adopt a cluster of six Xilinx VC709 FPGA for cooperative convolutional neural network inference \cite{zhang2016energy}.
Regarding to the CGRAs, the SambaNova DataScale system incorporates 8 reconfigurable-dataflow-units (RDUs) derived from Plasticine \cite{prabhakar2017plasticine}, claiming higher performance than a thousand GPUs for training extremely large deep-learning models. PPA~\cite{park2009polymorphic} exploits pipeline parallelism in streaming applications to create a CGRA-like pipeline to execute streaming media applications. Samsung proposes to adopt the CGRA cluster for medical volume image rendering \cite{jin2012implementation}. HammerBlade \cite{HammerBlade} aims at designing a rack-scale cluster for ML and Graphs. Their ASIC is composed of general-purpose cores and specialized CGRAs (e.g., Chimera \cite{ye2000chimaera}).

\textbf{Asynchronous Task Execution.}
The asynchronous tasks execution has been studied in many graph processing frameworks \cite{nasre2013data,nguyen2013lightweight,ben2017groute}.
They propose software-level data-centric approaches to implement irregular graph kernels on multi-node clusters \cite{nguyen2013lightweight} and GPUs platform \cite{nasre2013data}. The work that is most relevant to ARENA is Groute \cite{ben2017groute}, which is an asynchronous runtime environment for processing irregular graphs. In Groute, the GPUs form a logical ring network with each GPU conserves a worklist. Tasks are encapsulated as messages passing along the ring. ARENA is motivated by Groute. However, Groute is a pure software implementation based on general-purpose GPUs, thus cannot benefit from hardware specialization. Additionally, for Groute, the routing policy, which specifies the action when input is received, as well as memory consistency and ownership, are all defined and maintained by the users, which can be complicated, tedious and error-prone (e.g., imagine if data dependency occurs at runtime, users have to manually locate them and fetch in corresponding tasks). In ARENA, these are designed and supported by hardware. Furthermore, global coordination and work counting in Groute are centralized and managed by the CPU whereas in ARENA, all of them are distributedly processed.

\subsection{ARENA Runtime Support}\label{subsec:runtime}

Figure~\ref{fig:runtime_demo} and \ref{fig:runtime_code} illustrate the workflow of ARENA runtime executed per-node and the pseudo-code of the workflow, respectively. As can be seen, multiple tasks (marked with different colors) can be asynchronously executed in parallel. Note, the runtime can be supported by CPUs, GPUs, DSPs, or any other fixed or reconfigurable hardware substrate given the substrate realize the \emph{Hardware Abstract Functions} and support the \emph{Base Constructs}.

\begin{figure}[htb!]
\centering
\includegraphics[width=0.9\columnwidth]{graphs/runtime1.pdf}
\caption{ARENA runtime workflow.}
\label{fig:runtime_demo}
\end{figure}

\begin{figure}[htb!]
\begin{lstlisting}[basicstyle={\scriptsize\fontencoding{T1}\footnotesize\fontfamily{lmtt}\fontseries{c}\selectfont},escapechar=!]
void ARENA_runtime() {
  bool terminate = false;
  TaskQueue WaitQueue, RecvQueue;!\label{algo:runtime:queue:start}!
  TaskQueue SendQueue, CoalesceUnit;!\label{algo:runtime:queue:end}!
  TaskToken token;
  Address local_start, local_end;
  ARENA_init(&local_start, &local_end);!\label{algo:runtime:init}!
  while(true) {!\label{algo:runtime:while}!
    // Enqueue an arriving task token.
    token = ARENA_arrive();
    RecvQueue.enqueue(token);!\label{algo:runtime:arrive}!
    // Check termination for all tasks.!\label{algo:runtime:terminate:start}!
    token = RecvQueue.dequeue();
    if(token.TASKid == TERMINATE and
       WaitQueue.empty()) {
      SendQueue.enqueue(token);
      if(terminate) break;
      terminate = true;
      continue; }
    terminate = false;!\label{algo:runtime:terminate:end}!
    // Offload, split, or convey a task.!\label{algo:runtime:filter:start}!
    ARENA_filter(token, local_start, local_end,
                 &SendQueue, &WaitQueue);!\label{algo:runtime:filter:end}!
    // Check availability for task execution.
    token = WaitQueue.peek();
    if(ARENA_ready(token)) {!\label{algo:runtime:ready}!
      WaitQueue.dequeue();
      // Acquire data if necessary.
      if(token.REMOTEend > token.REMOTEstart) {
        ARENA_data_acquire(token);}!\label{algo:runtime:acquire}!
      // Issue a task for execution and return
      // without waiting for completion.
      ARENA_launch(token, CoalesceUnit);{!\label{algo:runtime:launch}!
      // Task coalescing if possible.
      token = ARENA_coalesce(CoalesceUnit);{!\label{algo:runtime:coalesce}!
      if(token)
        RecvQueue.enqueue(token); }}}
\end{lstlisting}
\caption{ARENA runtime pseudocode.}
\label{fig:runtime_code}
\end{figure}

\newcommand{\RNum}[1]{\uppercase\expandafter{\romannumeral #1\relax}}

We describe the runtime process below: Primarily, the task queues (line~\ref{algo:runtime:queue:start}-\ref{algo:runtime:queue:end}) and local data range (line~\ref{algo:runtime:init}) are initialized. We then proceed with 6 steps: \textbf{Step-(1):} All the incoming task tokens from the proceeding node will be appended to the \texttt{RecvQueue} (line~\ref{algo:runtime:while}-\ref{algo:runtime:arrive}). \textbf{Step-(2):} A token popped from the \texttt{RecvQueue} will be processed in the \texttt{Filter}, where a task can be split into multiple tasks, which are either buffered in the \texttt{WaitQueue} for local execution, or forwarded to the \texttt{SendQueue}, wait for being conveyed to the next node (line~\ref{algo:runtime:filter:start}-\ref{algo:runtime:filter:end}). The logic is: (\RNum{1}) if the task data range is irrelevant to the node's local data range, it is forwarded to \texttt{SendQueue} as it is; (\RNum{2}) if the task data range is a subset of the node's local data range ($local_{start} \leq TASK_{start} \leq TASK_{end} \leq local_{end}$), implying the local availability of all the needed data for the task, the token will be pushed into \texttt{WaitQueue} for future execution; (\RNum{3}) if the data range indicated by the task is a superset of the node's local data range ($TASK_{start} \leq local_{start} \leq local_{end} \leq TASK_{end}$), it suggests that the task might be too coarse-grained. We split the task into three portions and spawn three new tasks. The one with data range $local_{start}$ to $local\_{end}$ will be buffered in \texttt{WaitQueue} for local processing; the other two tasks are redirected to \texttt{SendQueue}; (\RNum{4}) finally, if the task data range is partially aligned with the node's local data range, two new tasks will be spawned. The aligned part is buffered in \texttt{WaitQueue} while the mismatch part is forwarded to \texttt{SendQueue}. \textbf{Step-(3):} The runtime checks whether there is available resources for the token at the top of the \texttt{WaitQueue} to be executed (line~\ref{algo:runtime:ready}). \textbf{Step-(4):} If so, the runtime verifies whether the task needs to incur any inevitable remote data access. If yes, the data will be acquired from the objective remote nodes (line~\ref{algo:runtime:acquire}) through the \texttt{Data-Transfer-Network}. \textbf{Step-(5):} when all demanding data are available, the task is issued to the computing resources for execution (line~\ref{algo:runtime:launch}). \textbf{Step-(6)} As new tasks can be generated locally in a node during the task execution, to avoid too many tasks flooding the system, a \texttt{CoalesceUnit} (line~\ref{algo:runtime:coalesce}) is designed to aggregate the newly generated tasks if the boundaries of their data-ranges coincide each other, and whether alternative key parameters ((e.g., task-token carried partial-reduction variables) are the same. This can avoid the scenario that too many generated fine-grained tasks saturate the task-token network and the associated buffers. Finally, the runtime on a particular node terminates when the TERMINATE token has been continuously received, and there is no pending tasks in the local \texttt{WaitQueue} (line~\ref{algo:runtime:terminate:start}-\ref{algo:runtime:terminate:end}).

\comm{
\SetKw{Continue}{continue}
\SetKw{Break}{break}

\begin{algorithm}[h]
\SetAlgoLined
  \textbf{TaskQueue} WaitQueue, RecvQueue, SendQueue;\\
  \textbf{bool} once = False;\\
  \textbf{TaskToken} token;\\
  \textbf{Address} local\_start, local\_end;\\
  \textbf{ARENA\_init}(&local_start, &local_end);\\
  \While{True} {
    // Enqueue an arriving task token\\
    token = \textbf{ARENA\_arrive}();\\
    RecvQueue.enqueue(token);\\
    token = RecvQueue.dequeue();\\
    \If{token.TASKid == TERMINATE} {
      \If{once} {
        \Break;
      }
      once = True;\\
      SendQueue.enqueue(token);\\
      \Continue;
    }
    once = false;\\
    // Task dispatching.\\
    \textbf{ARENA\_filter}(token, local\_start, local\_end,\\
                 &SendQueue, &WaitQueue);\\

    token = WaitQueue.peek();\\
    \If{\textbf{ARENA\_ready}(token)} {
      WaitQueue.dequeue();\\

      // Acquire data if necessary.\\
      \If{token.REMOTEend-token.REMOTEstart>0} {
        \textbf{ARENA\_data\_acquire}(token);\\
      }
      // Issue a task for execution without waiting for completion.\\
      \textbf{ARENA\_launch}(token);\\

      // Task coalescing.\\
      token = \textbf{ARENA_coalesce}();\\
      \If{token}{
        RecvQueue.enqueue(token);
      }
    }
  }
 \caption{ARENA\_runtime() Pseudocode.}
\end{algorithm}
}