数据流最优化限制 - 2021.2 Chinese

Vitis 高层次综合用户指南 (UG1399)

Document ID

UG1399

Release Date

2021-12-15

Version

2021.2 Chinese

DATAFLOW 最优化可对任务（函数和循环）之间的数据流以及按理想方式流水打拍的函数和循环进行最优化，从而最大限度提升性能。这些任务无需逐一链接，但数据传输方式存在某些限制。

以下行为可能阻止或限制 Vitis HLS 可通过 DATAFLOW 最优化执行的重叠。

在数据流区域中间读取函数输入或写入函数输出
单一生产者使用者违例
绕过任务和通道大小调整
任务间的反馈
任务的有条件执行
含多个退出条件的循环

重要：如果存在上述任一编码样式，Vitis HLS 就会发出一条描述状况的消息。

注释：您可在Analysis透视图中使用“Dataflow viewer”（数据流查看器）来查看应用 DATAFLOW 指令时的结构。

读取输入/写入输出

读取函数输入应在数据流区域起始位置执行，写入输出则应在数据流区域结束位置执行。读取/写入函数端口可能导致各进程按顺序执行，而非按重叠方式执行，从而对性能产生不利影响。

单一生产者使用者违例

为便于 Vitis HLS 执行 DATAFLOW 最优化，任务间传递的所有元素都必须遵循单一生产者使用者模型。每个变量都必须从单一任务驱动，并且仅限供单一任务使用。在以下代码示例中，temp1 将扇出并供 Loop2 和 Loop3 使用。这违背了单一生产者使用者模型的要求。

void foo(int data_in[N], int scale, int data_out1[N], int data_out2[N]) {
   int temp1[N];

   Loop1: for(int i = 0; i < N; i++) {
     temp1[i] = data_in[i] * scale;   
   }
   Loop2: for(int j = 0; j < N; j++) {
     data_out1[j] = temp1[j] * 123;   
   }
   Loop3: for(int k = 0; k < N; k++) {
    data_out2[k] = temp1[k] * 456;
   }
}

此代码的修改后版本使用 Split 函数来创建单一生产者使用者设计。以下代码块示例显示了通过 Split 对带有函数的数据流执行拆分的方式。现在数据在全部 4 个任务之间流动，这样 Vitis HLS 即可执行 DATAFLOW 最优化。

void Split (in[N], out1[N], out2[N]) {
// Duplicated data
 L1:for(int i=1;i<N;i++) {
 out1[i] = in[i]; 
 out2[i] = in[i];     
 }
}
void foo(int data_in[N], int scale, int data_out1[N], int data_out2[N]) {

 int temp1[N], temp2[N]. temp3[N]; 
 Loop1: for(int i = 0; i < N; i++) {
 temp1[i] = data_in[i] * scale;
 }
 Split(temp1, temp2, temp3);
 Loop2: for(int j = 0; j < N; j++) {
 data_out1[j] = temp2[j] * 123;
 }
 Loop3: for(int k = 0; k < N; k++) {
 data_out2[k] = temp3[k] * 456;
 }
}

绕过任务和通道大小调整

此外，通常数据应在不同任务间流动。如果您绕过任务，则可能降低 DATAFLOW 最优化的性能。在以下示例中，Loop1 会为 temp1 和 temp2 生成值。但下一项任务 Loop2 仅使用 temp1 的值。直至 Loop2 后才会使用 temp2 的值。因此，temp2 会绕过序列中的下一项任务，这可能限制 DATAFLOW 最优化的性能。

void foo(int data_in[N], int scale, int data_out1[N], int data_out2[N]) {
  int temp1[N], temp2[N]. temp3[N];
  Loop1: for(int i = 0; i < N; i++) {
  temp1[i] = data_in[i] * scale;
  temp2[i] = data_in[i] >> scale;
  }
  Loop2: for(int j = 0; j < N; j++) {
  temp3[j] = temp1[j] + 123;
  }
  Loop3: for(int k = 0; k < N; k++) {
  data_out[k] = temp2[k] + temp3[k];
  }
}

在此情况下，应将用于存储 temp2 的 PIPO 缓冲器的深度增大为 3，而不是采用默认深度 2。这样即可允许该缓冲器在执行 Loop2 的同时，存储用于 Loop3 的值。同样，绕过 2 个进程的 PIPO 深度应设为 4。使用 STREAM 编译指示或指令设置缓冲器深度：

#pragma HLS STREAM type=pipo variable=temp2 depth=3

重要：通道大小调整也会对性能产生类似的影响。FIFO/PIPO 深度不匹配可能因 FIFO/PIPO 的反压而导致数据流区域内无意间出现同步点。

任务间的反馈

当任务输出供 DATAFLOW 区域中前一个任务使用时，就会发生反馈。在 DATAFLOW 区域中不建议任务间反馈。当 Vitis HLS 检测到反馈时，它会根据情况发出警告，并且可能不执行 DATAFLOW 最优化。

但是，DATAFLOW 搭配 hls::streams 使用即可支持反馈。以下示例演示了此例外。

#include "ap_axi_sdata.h"
#include "hls_stream.h"

void firstProc(hls::stream<int> &forwardOUT, hls::stream<int> &backwardIN) {
  static bool first = true;
  int fromSecond;

  //Initialize stream
  if (first) 
    fromSecond = 10; // Initial stream value
  else
    //Read from stream
    fromSecond = backwardIN.read(); //Feedback value
  first = false;

  //Write to stream
  forwardOUT.write(fromSecond*2);
}

void secondProc(hls::stream<int> &forwardIN, hls::stream<int> &backwardOUT) {
  backwardOUT.write(forwardIN.read() + 1);
}

void top(...) {
#pragma HLS dataflow
  hls::stream<int> forward, backward;
  firstProc(forward, backward);
  secondProc(forward, backward);
}

在此简单设计中，执行 firstProc 时，它使用 10 作为输入的初始值。由于 hls::streams 不支持初始值，此方法可用于在不违反单一生产者使用者规则的前提下提供初始值。在后续迭代中，firstProc 会通过 backwardIN 接口读取 hls::stream。

firstProc 会处理该值，并通过串流将其发送给 secondProc，此串流将按原始 C++ 函数执行顺序正向传输。secondProc 会读取 forwardIN 上的值，对其加 1，并通过反馈串流将其发回 firstProc，此反馈串流按执行顺序反向传输。

从第二次执行开始，firstProc 会使用从串流中读取的值来执行其计算，两个进程可通过正向和反向通信来保持永久运行，针对首次执行则使用初始值即可。

任务的有条件执行

DATAFLOW 最优化不会对有条件执行的任务进行最优化。以下示例着重演示了此限制。在此示例中，Loop1 和 Loop2 的有条件执行会阻止 Vitis HLS 对这些循环之间的数据流进行最优化，因为在循环之间不发生数据流动。

void foo(int data_in1[N], int data_out[N], int sel) {

 int temp1[N], temp2[N];

 if (sel) {
 Loop1: for(int i = 0; i < N; i++) {
 temp1[i] = data_in[i] * 123;
 temp2[i] = data_in[i];
 }
 } else {
 Loop2: for(int j = 0; j < N; j++) {
 temp1[j] = data_in[j] * 321;
 temp2[j] = data_in[j];
 }
 }
 Loop3: for(int k = 0; k < N; k++) {
 data_out[k] = temp1[k] * temp2[k];
 }
}

要确保在所有情况下都能执行每个循环，必须按以下示例中所示方式对代码进行转换。在此示例中，条件语句已移至首个循环内。这两个循环都始终执行，数据始终在循环间进行流传输。

void foo(int data_in[N], int data_out[N], int sel) {

 int temp1[N], temp2[N];

 Loop1: for(int i = 0; i < N; i++) {
 if (sel) {
 temp1[i] = data_in[i] * 123;
 } else {
 temp1[i] = data_in[i] * 321;
 }
 }
 Loop2: for(int j = 0; j < N; j++) {
 temp2[j] = data_in[j];
 }
 Loop3: for(int k = 0; k < N; k++) {
 data_out[k] = temp1[k] * temp2[k];
 }
}

含多个退出条件的循环

在 DATAFLOW 区域中无法使用含多个出口点的循环。在以下示例中，Loop2 具有 3 项退出条件：

由 N 的值定义的出口：当 k>=N 时循环将退出。
由 break 语句定义的出口。

由 continue 语句定义的出口。

#include "ap_int.h"
#define N 16

typedef ap_int<8> din_t;
typedef ap_int<15> dout_t;
typedef ap_uint<8> dsc_t;
typedef ap_uint<1> dsel_t;

void multi_exit(din_t data_in[N], dsc_t scale, dsel_t select, dout_t data_out[N]) {
 dout_t temp1[N], temp2[N];
 int i,k;

 Loop1: for(i = 0; i < N; i++) {
 temp1[i] = data_in[i] * scale;
 temp2[i] = data_in[i] >> scale;
 }

 Loop2: for(k = 0; k < N; k++) {
 switch(select) {
        case  0: data_out[k] = temp1[k] + temp2[k];
        case  1: continue;
        default: break;
 }
 }
}

由于循环的退出条件始终由循环边界来定义，因此使用 break 或 continue 语句将禁止在 DATAFLOW 区域内使用循环。

最后，DATAFLOW 最优化不含任何分层实现。如果子函数或循环包含可能受益于 DATAFLOW 最优化的其它任务，那么必须对该循环、子函数或子函数的内联应用 DATAFLOW 最优化。

您也可以在 DATAFLOW 区域内使用 std::complex。但应搭配 __attribute__((no_ctor)) 一起使用，如下示例所示：

void proc_1(std::complex<float> (&buffer)[50], const std::complex<float> *in);
void proc_2(hls::Stream<std::complex<float>> &fifo, const std::complex<float> (&buffer)[50], std::complex<float> &acc);
void proc_3(std::complex<float> *out, hls::Stream<std::complex<float>> &fifo, const std::complex<float> acc);

void top(std::complex<float> *out, const std::complex<float> *in) {
#pragma HLS DATAFLOW

  std::complex<float> acc __attribute((no_ctor)); // Here
  std::complex<float> buffer[50] __attribute__((no_ctor)); // Here
  hls::Stream<std::complex<float>, 5> fifo; // Not here

  proc_1(buffer, in);
  proc_2(fifo, buffer, acc);
  proc_3(out, fifo, acc);
}