这是您第一次登陆到 developerWorks，已经自动为您创建了您的概要文件。 选择您概要文件中可以公开的信息的信息（如姓名、国家/地区，以及公司），这些信息同时也会与您所发布的内容相关联。 您可以随时更新您的 IBM 账号。

所有提交的信息确保安全。

当您初次登录到 developerWorks 时，将会为您创建一份概要信息，您需要指定一个昵称。您的昵称将和您在 developerWorks 发布的内容显示在一起。

昵称长度在 3 至 31 个字符之间。您的昵称在 developerWorks 社区中必须是唯一的，并且出于隐私保护的原因，不能是您的电子邮件地址。

所有提交的信息确保安全。

My developerWorks:

我的通知：

退出

选择语言：

在 Cell BE 处理器上编写高性能的应用程序，第 6 部分: 用 DMA 传输进行智能缓冲区管理

利用双缓冲和多重缓冲支持 SPU

Jonathan Bartlett (johnnyb@eskimo.com), 技术总监, New Medio

Jonathan Bartlett 是 Programming from the Ground Up 一书的作者，这本书对使用 Linux 汇编语言的编程进行了简介。他是 New Medio 的开发技术总监，为客户开发 Web 应用程序、视频应用程序、kiosk 应用程序和桌面应用程序。

简介： 研究双缓冲和多重缓冲概念，从而通过处理和数据传输的并行化来提高代码速度，并让 SPE 的内存流控制器（MFC）产生最佳的装载和存储操作次序。

查看本系列更多内容

发布日期： 2007 年 5 月 24 日
级别： 中级
访问情况： 2398 次浏览
评论：

本系列以前的文章中编写的代码采用以下基本模式：

SPU 对 DMA GET 进行排队，从而将涉及的数据集的一部分从主内存传输到一个缓冲区。
SPU 等待缓冲区被填充。
SPU 处理缓冲区。
SPU 对 DMA PUT 进行排队，从而将缓冲区传输回主内存。
SPU 等待缓冲区完成传输。
如果还有数据，就重复这个过程。

这个过程的问题是，它会浪费大量处理器时间。两个传输步骤根本没有涉及 SPU，只涉及 MFC（这是比较大的 SPE 的一部分）。在目前编写的代码中，SPU 仅仅等待 MFC 完成传输，然后才能处理其他东西。在它等待时，我们肯定能够为它找到可以做的事情。

双缓冲

这里的情况很像医生办公室中的情况 —— 我预料到会在候诊室中等候很长时间。因此，总是带着一些工作，这样就可以在等候时有事情可做。同样的原理也应用于编程。不要浪费宝贵的处理器周期来等待数据传输。代码可以在等候时处理下一个缓冲区。因此，在等待传输一组数据时，可以处理另一组数据。新的处理算法像下面这样：

SPU 对 DMA GET 进行排队，从而将涉及的数据集的一部分从主内存传输到一号缓冲区。
SPU 对 DMA GET 进行排队，从而将涉及的数据集的另一部分从主内存传输到二号缓冲区。
SPU 等待一号缓冲区被填充。
SPU 处理一号缓冲区。
SPU (a) 对 DMA PUT 进行排队来传输一号缓冲区的内容，然后 (b) 对 DMA GETB 进行排队，从而在 PUT 之后用主内存中的下一部分数据重新填充这个缓冲区。
SPU 等待二号缓冲区被填充。
SPU 处理二号缓冲区。
SPU (a) 对 DMA PUT 进行排队来传输二号缓冲区的内容，然后 (b) 对 DMA GETB 进行排队，从而在 PUT 之后用主内存中的下一部分数据重新填充这个缓冲区。
从步骤 3 开始重复以上过程，直到所有数据处理完毕。
等待所有缓冲区传输完毕。

当然，与这个算法解决的问题相比，它导致的新问题可能更多。首先注意，当缓冲区用尽时，可能要做许多不必要的工作，因为对于每个循环迭代都要处理两个缓冲区。可以使用几个 if 语句提前退出，以及在处理完数据时停止缓冲区重新填充过程。但是，对于这个程序，我没有采用这种方法，因为它会在每个迭代 中引入许多额外的处理。如果代码要处理大型数据集，那么每个迭代的成本要比开始和结束时的成本重要得多。因此，为了避免出现分支，我将尽可能多的有条件工作分载到 SPE 中。对于缓冲区处理，MFC 将大小为零的数据请求当作无操作来对待，所以即使没有数据可读取了，也可以继续发出请求。对于实际的缓冲区处理，函数只需直接返回，就能够很好地处理大小为零的缓冲区。所以所有情况都已经得到了处理，清除额外的结束步骤的任何分支只会减慢对默认情况的处理。

另一个问题是如何在同一缓冲区 上调度 PUT 和 GET 而不造成冲突。在每个数据处理步骤之后，设置一个 PUT（将数据传输到主内存）和一个 GET（获得下一组数据）。因为在默认情况下 MFC 按照它选择的任意次序处理请求，所以如何确保特定的次序呢？正如在最后一篇文章中要讨论的，解决方法是 barrier 和 fence。在请求上放一个 fence，就会迫使同一标记组中以前发出的所有 MFC 请求在当前请求之前得到处理。但是，它并没有为未来的传输指定次序。barrier 与 fence 相似，但是它对以前的请求和后续请求的次序都进行限制。因此，通过发送带 fence 或 barrier 的第二个请求，可以迫使 MFC 以正确的次序处理请求；而且因为它们在同一个标记组中，所以在使用缓冲区时，只需等待整个标记组完成。对于单一缓冲区，GETB、PUTB、GETF 和 PUTF 是与 fence 和 barrier 相关的主要 DMA 命令。

现在，我们考虑一下如何将这个算法应用于当前的大写转换代码。为了便于参考，下面给出 convert_driver_c.c 中原来的代码：

清单 1. 原来的单一缓冲区 MFC 传输程序

                
#include <spu_intrinsics.h>
#include <spu_mfcio.h> /* constant declarations for the MFC */
typedef unsigned long long uint64;
typedef unsigned int uint32;

void convert_buffer_to_upper(char *conversion_buffer, int current_transfer_size);

#define MAX_TRANSFER_SIZE 16384
char conversion_buffer[MAX_TRANSFER_SIZE];

typedef struct {
	uint32 length __attribute__((aligned(16)));
	uint64 data __attribute__((aligned(16)));
} conversion_structure;

int main(uint64 spe_id, uint64 conversion_info_ea) {
	conversion_structure conversion_info; 
	/* Information about the data from the PPE */

	/* New variables to keep track of where we are in the data */
	uint32 remaining_data; /* How much data is left in the whole string */
	uint64 current_ea_pointer; /* Where we are in system memory */
	uint32 current_transfer_size; /* How big the current transfer is (may
	                               * be smaller than MAX_TRANSFER_SIZE) */

	/* We are only using one tag in this program */
	mfc_write_tag_mask(1<<0);

	/* Grab the conversion information */
	mfc_get(&conversion_info, conversion_info_ea, 
	   sizeof(conversion_info), 0, 0, 0);
	spu_mfcstat(MFC_TAG_UPDATE_ALL); /* Wait for Completion */

	/* Setup the loop */
	remaining_data = conversion_info.length;
	current_ea_pointer = conversion_info.data;

	while(remaining_data > 0) {
		/* Determine how much data is left to transfer */
		if(remaining_data < MAX_TRANSFER_SIZE)
			current_transfer_size = remaining_data;
		else
			current_transfer_size = MAX_TRANSFER_SIZE;

		/* Get the actual data */
		mfc_getb(conversion_buffer, current_ea_pointer, 
		   current_transfer_size, 0, 0, 0);
		spu_mfcstat(MFC_TAG_UPDATE_ALL);

		/* Perform the conversion */
		convert_buffer_to_upper(conversion_buffer, current_transfer_size);

		/* Put the data back into system storage */
		mfc_putb(conversion_buffer, current_ea_pointer, 
		    current_transfer_size, 0, 0, 0);

		/* Advance to the next segment of data */
		remaining_data -= current_transfer_size;
		current_ea_pointer += current_transfer_size;
	}
	spu_mfcstat(MFC_TAG_UPDATE_ALL); /* Wait for Completion */
}

这个程序还需要来自以前文章的以下文件：convert_buffer_c.c（来自第 5 部分）和 ppu_dma_main.c（来自第 3 部分），本文后面还会提供另一个版本。像以前文章那样进行编译和运行（这些构建命令适用于本文中的所有示例）：

spu-gcc convert_buffer_c.c convert_driver_c.c -o spe_convert
embedspu -m64 convert_to_upper_handle spe_convert spe_convert_csf.o
gcc -m64 spe_convert_csf.o ppu_dma_main.c -lspe -o dma_convert
./dma_convert

为了让这个程序进行双缓冲，需要对代码略加重构。首先，应将所有与缓冲区相关的数据集中在一起。每个缓冲区都要依赖于这些数据：

缓冲区本身的地址
填充缓冲区所用数据的有效地址
处理的数据的大小

因此，创建以下结构来保存所有与缓冲区相关的信息：

struct {
	uint64 effective_address __attribute__((aligned(16)));
	uint32 size __attribute__((aligned(16)));
	char data[MAX_TRANSFER_SIZE] __attribute__((aligned(16)));
} buffer;

然后，只需为这两个缓冲区声明一个全局数组：

buffer buffers[2];

现在，将转换过程分成两个函数调用：

开始数据缓冲区装载
等待、处理并将数据存储回缓冲区

进行这样的划分是因为它们是需要重新安排的独立单元。必须在程序的开头开始数据装载，所以这个功能需要放在单独的函数中。下面是 MFC 代码的双缓冲版本（这个文件也是 convert_driver_c.c）：

清单 2. 双缓冲的 MFC 传输

                
#include <spu_intrinsics.h>
#include <spu_mfcio.h>

/* Constants */
#define MAX_TRANSFER_SIZE 16384

/* Data Structures */
typedef unsigned long long uint64;
typedef unsigned int uint32;
typedef struct {
	uint32 length __attribute__((aligned(16)));
	uint64 data __attribute__((aligned(16)));
} conversion_structure;

typedef struct {
	uint32 size __attribute__((aligned(16)));
	uint64 effective_address __attribute__((aligned(16)));
	char data[MAX_TRANSFER_SIZE] __attribute__((aligned(16)));
} buffer;

/* Global Variables */
buffer buffers[2];

/* Utility Functions */
inline uint32 MIN(uint32 a, uint32 b) {
	return a < b ? a : b;
}

inline void wait_for_completion(uint32 mask) {
	mfc_write_tag_mask(mask);
	spu_mfcstat(MFC_TAG_UPDATE_ALL);
}

inline void load_conversion_info(uint64 cinfo_ea, uint64 *data_ea, uint32 *data_size) {
	conversion_structure cinfo;
	mfc_get(&cinfo, cinfo_ea, sizeof(cinfo), 0, 0, 0);
	wait_for_completion(1<<0);
	*data_size = cinfo.length;
	*data_ea = cinfo.data;
}

/* Processing Functions */
inline void initiate_transfer(uint32 buf_idx, uint64 *current_ea_pointer,
uint32 *remaining_data) {
	/* Setup buffer information */
	buffers[buf_idx].size = MIN(*remaining_data, MAX_TRANSFER_SIZE);
	buffers[buf_idx].effective_address = *current_ea_pointer;
	/* Initiate transfer using the buffer index as the DMA tag */
	mfc_getb(buffers[buf_idx].data, buffers[buf_idx].effective_address,
		buffers[buf_idx].size, buf_idx, 0, 0);
	/* Move the data pointers */
	*remaining_data -= buffers[buf_idx].size;
	*current_ea_pointer += buffers[buf_idx].size;
}

inline void process_and_put_back(uint32 buf_idx) {
	wait_for_completion(1<<buf_idx);
	/* Perform conversion */
	convert_buffer_to_upper(buffers[buf_idx].data, buffers[buf_idx].size);
	/* Initiate the DMA transfer back using the buffer index as the DMA tag */
	mfc_putb(buffers[buf_idx].data, buffers[buf_idx].effective_address,
		buffers[buf_idx].size, buf_idx, 0, 0);
}

/* Main Code */
int main(uint64 spe_id, uint64 conversion_info_ea) {
	uint32 remaining_data;
	uint64 current_ea_pointer;

	load_conversion_info(conversion_info_ea, &current_ea_pointer, &remaining_data);

	/* Start filling buffers to prepare for loop (loop assumes both buffers have
	 * data coming in) */
	initiate_transfer(0, &current_ea_pointer, &remaining_data);
	initiate_transfer(1, &current_ea_pointer, &remaining_data);

	do {
		/* Process buffer 0 */
		process_and_put_back(0);
		initiate_transfer(0, &current_ea_pointer, &remaining_data);

		/* Process buffer 1 */
		process_and_put_back(1);
		initiate_transfer(1, &current_ea_pointer, &remaining_data);
	} while(buffers[0].size != 0);

	wait_for_completion(1<<0|1<<1);
}

注意，因为这段代码只处理缓冲区，所以除了大写转换特有的函数调用之外几乎没有特殊代码。在其他上下文中，非常容易重用它。

回页首

多重缓冲

前一节中使用的思想称为 “软件管道连接（software pipelining）”。也就是将处理划分为多个阶段，在执行时这些阶段可以重叠以使吞吐量最大化。在这个示例中，管道实际上只有两个阶段 —— 装载/存储和处理。但是，当应用这个思想处理其他问题时，可以建立任意数量的 “管道阶段”。基本思想是给每个管道分配一个要处理的缓冲区，然后按一次处理一个阶段的方式处理每个缓冲区。当一个软件管道使用的缓冲区多于两个时，它就称为多重缓冲（multibuffering）。对于 SPU，两阶段的管道适合大多数应用程序，因为数据移动由 MFC（而不是处理器）处理，因此管道阶段可以并发操作。处理和数据传输的并发本质使两阶段的管道连接很适合 SPU 编程。

除了增加管道阶段之外，还可以通过其他方法使用更多的缓冲区。主要方法是在 MFC 上启动许多数据传输，然后让 MFC 决定处理的次序。例如，假设一个内存区当前在交换空间中，而另一个在内存中。通过在 MFC 上启动许多传输，MFC 可以决定最佳的传输次序。另外，这有助于缓解总线冲突问题 —— 当总线繁忙时，程序可以处理额外的缓冲区，而不是等待总线空闲下来。当总线空闲下来时，它可以重新填充额外的缓冲区。在这个程序中，以这种方式执行缓冲区处理并不会显著影响执行时间，而对于某些数据集，可能有消极影响。但是，这仍然是一种可能有意义的缓冲区处理技术。下面的示例演示这种技术，并说明如何使用 MFC_TAG_UPDATE_ANY。

新的过程如下：

对所有缓冲区的 DMA GET 进行排队。如果缓冲区要传输多于零字节的数据，就将它标为 “filling”。每个缓冲区获得一个惟一的 DMA 标记 ID。
如果没有缓冲区标为 “filling”，就等待所有 DMA PUT 操作完成并退出。
等待一个标为 “filling” 的缓冲区变成为 “filled”。
处理这个缓冲区。
对 DMA PUT 进行排队，从而将缓冲区传输回主内存。
对 DMA GETB 进行排队，从而在将现有数据存储回去之后，用更多数据重新填充缓冲区。
如果前一步中的 DMA 传输至少传输 1 字节（换句话说，还有数据要传输），那么将缓冲区标为 “filling”。
返回到步骤 2。

在这个算法中，处理缓冲区的次序是很不确定的。关键的难点在于尽可能减少分支的数量。可能产生分支的地方是，判断是否应该将一个缓冲区标为 “filling”，以及对缓冲区进行轮询来寻找可用的缓冲区。通过仔细选择 SPU intrinsic 和良好的数据结构设计，可以轻松地避免分支。

等待缓冲区可用实际上非常容易。给出您感兴趣的缓冲区的掩码，可以调用 spu_mfcstat(MFC_STAT_UPDATE_ANY)，它会返回没有未完成操作（换句话说，所有操作都完成了）的所有缓冲区的掩码，并等待到至少有一个缓冲区可用。可以把它看作 C 库函数 select 的特殊版本，但是针对的是 DMA 传输。目前，它将返回所有可用的缓冲区，但是我们只需要一个。因此，必须将掩码转换为一个索引（可以使用这个索引指定要处理的缓冲区），而且进行转换时不用任何分支。SPU 指令 clz（count leading zeroes，计算前导零；在 C 语言 intrinsic 中称为 spu_cntlz）很适合进行这样的转换。计算前导零，然后从 31 中减去前导零数量，就可以将产生的掩码转换为索引。可能的汇编语言指令序列如下：

	#assume the mask is in $10
	#Count the leading zeroes
	clz $11, $10
	#Subtract that from 31
	sfi $12, $11, 31
	#$12 now has the index of the buffer we want to use.

在 C 中，这段代码可以写成：

	/* buffers_completed holds the mask */
	spu_extract(
		spu_sub(
			(int32)31,
			spu_cntlz(
				spu_promote(
					(uint32)buffers_completed, 0
				)
			)
		),
		0
	);

当然，这只会检索到第一个可用的缓冲区，可能有更多的可用缓冲区。但是，那些缓冲区也会在后续的循环迭代中返回。

现在，需要决定如何存储当前为 “filling” 的缓冲区，并能够设置这些标志；同样，不使用分支。存储这一信息的最好方法是存储为标记掩码形式，这样就可以将它直接用作 spu_mfcstat 的掩码。但是，在不用分支的情况下，有条件地设置这些位比较困难。汇编语言版本如下：

	#$10 holds our buffer mask
	#$11 holds the size of the last transfer
	#$12 holds the index of the current buffer

	#Convert the current buffer index to a bit for a bit mask (stored in $14)
	il $13, 1
	shl $14, $13, $12

	#Turn the bit off in the original mask
	xor $10, $10, $14

	#is the last transfer greater than zero? (answer stored in $15)
	cgti $15, $11, 0

	#Turn the bit on or off based the previous result (answer stored in $14)
	and $14, $14, $15

	#Turn the bit on based on our existing results
	or $10, $10, $14

通过适当的调度，可以将这段代码缩减到 10 个周期。但是，这种操作可以由编译器负责处理。实际上，只需编写下面这样的代码，编译器就会进行适当的优化：

	/* clear the bit */
	*buffers_with_data &= ~(1<<buf_idx);
	/* Set the bit conditionally */
	*buffers_with_data |= (buffers[buf_idx].size > 0 ? (1<<buf_idx) : 0);

在这个程序中，因为问题在于可并行性很一般，所以实际上可以使用 SPU 的本地存储能够支持的任意数量的缓冲区。因为在这个程序中每个缓冲区（在理论上）可以有两个活动的 DMA 传输（一个存储，一个装载），而 MFC 只能有 16 个未完成的 DMA 操作，所以程序最多有 8 个缓冲区。现在，如果超过这个限制，并不会影响程序的逻辑操作。当添加第 17 个 DMA 操作时，仅仅会停止 SPU，直到一个操作完成；此时，允许程序继续执行排队的下一个操作。

下面是新版本的代码（文件同样是 convert_driver_c.c）：

清单 3. 多重缓冲 MFC 传输

                
#include <spu_intrinsics.h>
#include <spu_mfcio.h>
typedef unsigned long long uint64;
typedef unsigned int uint32;
typedef int int32;

/* Constants */
#define MAX_TRANSFER_SIZE 16384
#define NUM_BUFFERS 8 /* The MFC supports only 16 queued transfers,
                       * and we have up to two active per buffer */

/* Data Structures */
typedef struct {
	uint32 length __attribute__((aligned(16)));
	uint64 data __attribute__((aligned(16)));
} conversion_structure;

typedef struct {
	uint32 size __attribute__((aligned(16)));
	uint64 effective_address __attribute__((aligned(16)));
	char data[MAX_TRANSFER_SIZE] __attribute__((aligned(16)));
} buffer;

buffer buffers[NUM_BUFFERS];

/* Utility functions */
inline uint32 MIN(uint32 a, uint32 b) {
	return a < b ? a : b;
}

/* Processes the buffer, queues a DMA transfer to put the data back, and clears out
 * the "waiting for data" bit in buffers_with_data */
inline void process_and_put_back(uint32 buf_idx, uint32 *buffers_with_data) {
	convert_buffer_to_upper(buffers[buf_idx].data, buffers[buf_idx].size);
	mfc_putb(buffers[buf_idx].data, buffers[buf_idx].effective_address,
		buffers[buf_idx].size, buf_idx, 0, 0);
	*buffers_with_data &= ~(1<<buf_idx); /* Clear out bit for this buffer */
}

/* Queues up a DMA GET transfer, ad, if there is any data to transfer, sets
 * the appropriate bit in buffers_with_data to indicate that we are waiting
 * for data in this buffer */
inline void initiate_transfer(uint32 buf_idx, uint32 *buffers_with_data,
uint64 *current_ea_pointer, uint32 *remaining_data) {
	/* Setup buffer */
	buffers[buf_idx].size = MIN(*remaining_data, MAX_TRANSFER_SIZE);
	buffers[buf_idx].effective_address = *current_ea_pointer;

	/* Move Data Pointers */
	*remaining_data -= buffers[buf_idx].size;
	*current_ea_pointer += buffers[buf_idx].size;

	/* Initiate transfer (does nothing if there is no data) */
	mfc_get(buffers[buf_idx].data, buffers[buf_idx].effective_address,
		buffers[buf_idx].size, buf_idx, 0, 0);

	/* Set the "Buffer Waiting for Data" bit only if there is data to read */
	*buffers_with_data |= (buffers[buf_idx].size > 0 ? (1<<buf_idx) : 0);
}

/* Waits for all of the given buffers to complete */
inline void wait_for_completion(uint32 mask) {
	mfc_write_tag_mask(mask);
	spu_mfcstat(MFC_TAG_UPDATE_ALL);
}

/* Loads information about the whole conversion process */
inline void load_conversion_info(uint64 conversion_info_ea, uint64 *current_ea_pointer,
uint32 *remaining_data) {
	conversion_structure conversion_info;

	mfc_get(&conversion_info, conversion_info_ea, sizeof(conversion_info), 0, 0, 0);
	wait_for_completion(1<<0);

	*remaining_data = conversion_info.length;
	*current_ea_pointer = conversion_info.data;
}

/* Returns the index of the first buffer with data available*/
inline uint32 get_next_buffer(uint32 buffers_with_data) {
	uint32 buffers_completed; /* This will contain a mask of buffers whose
	                           * transfers have completed */

	/* These are the buffers to look for */
	mfc_write_tag_mask(buffers_with_data);

	/* Wait for at least one buffer to come available */
	buffers_completed = spu_mfcstat(MFC_TAG_UPDATE_ANY);

	/* Use "count leading zeros" to determine the buffer index from
	 * the buffers_completed mask */
	return spu_extract(
		spu_sub(
			(int32)31,
			spu_cntlz(
				spu_promote((uint32)buffers_completed, 0)
			)
		),
		0
	);
}

/* Steps are numbered according to the description in this section */
int main(uint64 spe_id, uint64 conversion_info_ea) {
	uint32 remaining_data;
	uint64 current_ea_pointer;
	uint32 buffers_with_data = 0; 
	 /* This is the bit mask for each buffer waiting on data,
	                               * used for spu_mfcstat in the main loop */
	uint32 all_buffers = 0; /* This is used to wait on all remaining transfers at
	                         * the end of the program*/
	uint32 current_buffer_idx;

	load_conversion_info(conversion_info_ea, &current_ea_pointer, &remaining_data);

/* Step 1: Get all buffers loading (because NUM_BUFFERS is a constant, 
	 *        the compiler should unroll the loop all the way) */
	for(current_buffer_idx = 0; current_buffer_idx < NUM_BUFFERS; 
	   current_buffer_idx++) {
		initiate_transfer(current_buffer_idx, &buffers_with_data,
			&current_ea_pointer, &remaining_data);
		all_buffers |= 1<<current_buffer_idx;
	}

	/* Step 2: Continue while there are still buffers pending */
	while(buffers_with_data != 0) {
		/* Step 3: Get the next buffer that gets filled */
		current_buffer_idx = get_next_buffer(buffers_with_data);
		/* Steps 4 and 5: Process the buffer and queue up a DMA transfer 
		back to main memory */
		process_and_put_back(current_buffer_idx, &buffers_with_data);
		/* Steps 6 and 7: Queue up a buffer reload, and mark the buffer
		* as "filling"  (by setting the appropriate bit in remaining_data) */
		initiate_transfer(current_buffer_idx, &buffers_with_data,
			&current_ea_pointer, &remaining_data);
	}

	/* Wait for all PUTs to complete */
	wait_for_completion(all_buffers);
}

这里的许多代码是为了确保主循环和对 convert_buffer_to_upper 的函数调用是仅有的分支。其他代码要么是内联函数（显然可以由编译器进行内联），要么是可以由编译器轻松消除分支的代码。许多可以缩写为三元操作 ? : 的分支，如果没有副作用，也没有调用非内联函数，那么编译器（GCC 或 XLC）就可以消除这些分支。

以前看到的 PPE 示例代码（ppu_dma_main.c）只使用一个缓冲区，所以它没有采用前面讨论的优化措施。为了了解这些程序处理比较大的数据集的效果，下面给出 ppu_dma_main.c 程序的另一个版本，它使用大数据集并对 SPU 进行计时：

清单 4. 测试大数据集的驱动程序

                
#include <stdio.h>
#include <libspe.h>
#include <errno.h>
#include <string.h>
#include <sys/time.h>
#include <malloc.h>

/* Size of Buffer -- MUST be a multiple of 16 */
#define BUF_SIZE (16 * 200000)

/* embedspu actually defines this in the generated object file, 
we only need an extern reference here */
extern spe_program_handle_t convert_to_upper_handle;

/* This is the parameter structure that our SPE code expects */
/* Note the alignment on all of the data that will be passed to the SPE is 16-bytes */
typedef struct {
	int length __attribute__((aligned(16)));
	unsigned long long data __attribute__((aligned(16)));
} conversion_structure;

int main() {
	int status = 0;
	int i;
	struct timeval initial_time, final_time;

	/* Create the string on an aligned boundary */
	char *str = memalign(16, BUF_SIZE);

	/* Fill the string with data */
	for(i = 0; i < BUF_SIZE - 1 ; i++) {
		str[i] = 'a' + i % 26;
	}

	/* Null-terminate string */
	str[BUF_SIZE - 1] = '\0';

	/* Create conversion structure on an aligned boundary */
	conversion_structure conversion_info __attribute__((aligned(16)));

	/* Set the data elements in the parameter structure */
	conversion_info.length = BUF_SIZE; /* add one for null byte */
	conversion_info.data = (unsigned long long)str;

	/* Check starting time */
	gettimeofday(&initial_time, NULL);

	/* Create the thread and check for errors */
	speid_t spe_id = spe_create_thread(0, &convert_to_upper_handle, 
	&conversion_info, NULL, -1, 0);
	if(spe_id == 0) {
		fprintf(stderr, "Unable to create SPE thread: errno=%d\n", errno);
		return 1;
	}

	/* Wait for SPE thread completion */
	spe_wait(spe_id, &status, 0);

	/* Check final time */
	gettimeofday(&final_time, NULL);

	/* Print SPU execution time */
	fprintf(stderr, "%llu microseconds\n",
		((long long)final_time.tv_sec * 1000000 + final_time.tv_usec) -
		((long long) initial_time.tv_sec * 1000000 + initial_time.tv_usec));

	/* Print out result - uncomment if you really want to see it*/
	//printf("The converted string is: %s\n", str);

	return 0;
}

回页首

结束语

本文讨论了在 SPE 上进行缓冲区管理的两种技术 —— 双缓冲和多重缓冲。讲解了如何扩展现有的代码来同时启用多个缓冲区，并让 MFC 决定填充缓冲区的次序，同时确保每个步骤不会引入任何不必要的分支。

参考资料

您可以参阅本文在 developerWorks 全球网站上的英文原文。
请参考在 Cell BE 处理器上编写高性能的应用程序系列文章的其他部分。
MIT 开办了一个关于 PS3 编程的课程。IBM 的 Michael Perrone 的精彩演讲（PDF 格式）详细介绍了 EIB（Element Interconnect Bus，连接 SPE 和主内存以及相互连接的总线）的工作方式及其限制，以及其他问题。
在同一门课程中，IBM 的 Rodric Rabbah 针对并行编程设计模式做了两次演讲，涉及了我们讨论的一些软件管道连接概念（第 1 部分和第 2 部分，都是 PDF 格式）。
在 IBM Cell/B.E. workshop 提供的幻灯片（PDF 格式）中，有另一个 SPU 双缓冲示例（以及关于 DMA 传输的大量信息）。
可以在 PPU & SPU C/C++ Language Extension Specification 中了解所有 C/C++ intrinsic。
要想及时了解关于 Cell/B.E. 的所有情况，请订阅 IBM microNews。