在 PlayStation 3（Cell 架构）上进行并行编程示例：解谜

• 下载解谜代码 - 38.8 KB
• 下载测试 - 563 KB

引言

本文旨在为程序员提供全新 PlayStation 3 主机编程模型的概述。为此，我将文章分为两部分：第一部分介绍硬件架构、编程模型，以及资源、API 的查找和使用方法。第二部分展示一个示例应用程序：我们将尝试尽可能多地利用提供的编程功能来解决一个谜题。

第一部分：Cell

历史背景

随着对处理能力需求的增长，硬件设计者发现满足需求越来越困难。

第一个障碍是所谓的“频率墙”，它基本上意味着通过提高处理器频率只能取得边际效益，因为更深的流水线会带来边际收益递减。一个相关的障碍是散热（“功耗墙”），随着功耗的增加，这个问题变得更加棘手。

另一个障碍是“内存墙”。内存的工作速度远低于 CPU；多层内存（硬盘、RAM、L3 缓存、L2 缓存、L1 缓存）尽力隐藏这种延迟，但当今的系统通常受内存速度的限制。提高性能的可能解决方案是：提高效率和增加并发性。

2000 年由索尼、IBM 和东芝组成的联盟旨在实现这两者。他们创造的名为 Cell Broadband Engine 的处理架构，现已用于需要强大处理能力（如 3D 游戏）的消费类产品（PlayStation 3）和服务器。

架构概述

该架构是一种异构多核架构：我们有一个用于控制任务的 Power Processing Element (PPE) 和 8 个用于数据密集型处理的 Synergetic Processing Elements (SPE)。

从 Flynn 分类法的角度来看，Cell 架构是 MIMD（多指令处理多数据）。也就是说，我们可以将 Cell 看作是单个芯片上的 9 个独立处理器，其中一个为主处理器，为其他处理器提供工作，而其他处理器则执行实际工作。

Cell PPE 是一个 64 位 RISC PowerPC 处理器，工作频率为 3.2GHz，支持 2 路硬件多线程，并具有独立的指令和数据 L1 缓存以及统一的 L2 缓存。

每个芯片上有 8 个 Cell SPE，提供整数和浮点运算以及 256KB 的本地存储。设计者有趣的选择是用这个本地存储器（普通内存，不像缓存那样是关联式的）取代片上缓存。每个 SPE 提供 4 个浮点单元和整数单元，使得在任何时刻都可以进行四次数学运算。如您所见，性能不仅来自于增加核心数量的并发性，也来自于每个核心内部的并发性。为了控制这些单元，我们使用 SIMD（单指令多数据）流；这意味着我们可以同时处理一个包含四个 32 位元素的向量，但我们只能对向量中的所有元素应用一个（相同的）操作。或者，我们可以为相同的操作使用半字数据类型。

编程工具和 API

要为 Cell 编程，我们必须使用 IBM Cell Broadband Engine 资源中心 (http://www.ibm.com/developerworks/power/cell/) 中的 Cell SDK。在那里您可以找到大量关于硬件和所用编程模型的有用信息。

获取功能齐全的 Cell 编程环境的最快方法是下载一个 Cell 虚拟机，例如，从这个页面。

实际编程在 Linux 中使用此 SDK 进行，也可以选择 Eclipse IDE，它有一些插件支持与 Cell 编译器集成。Cell SDK 包含一个模拟器，允许您在没有实际 Cell 硬件的情况下进行 Cell 编程。

编程模型

从程序员的角度来看，我们使用的语言是 C++，带有特殊的扩展。让我们来分析一下 Cell 的编程模型。

SPE 和 PPE 可用的库不同（考虑到硬件资源不同，这是正常的）；这意味着某些函数仅在 SPE 代码中工作，另一些在 PPE 代码中工作；您需要查阅手册。通常，包含这些函数的头文件会提供线索（例如，spu_mfcio.h 用于 SPU 内存 IO 访问）。

SPE 线程

一个通用的 Cell 程序会在 PPE 代码中包含类似这样的代码；此代码用于将可执行镜像加载到 SPE 并启动它们，提供它们期望的参数。

void *spe_thread( void *voidarg ) 
{

  thread_args_t *arg = (thread_args_t *)voidarg;

  unsigned int runflags = 0;
  unsigned int entry = SPE_DEFAULT_ENTRY;

  spe_context_run( arg->spe_context, &entry, 
      runflags, arg->argp, arg->envp, NULL );
  pthread_exit( NULL );
}

void StartSpu(int i, int* originalPieceAddr, int** data)
{
    sprintf(buffer,"%d %d", piece_height, (int)originalPieceAddr);
    printf("Started SPU with %d %d %d and buffer %s", 
            originalPieceAddr[0], 
            originalPieceAddr[1],
            originalPieceAddr[2], buffer);
    spe_contexts[i] = spe_context_create( SPE_EVENTS_ENABLE, NULL ); 

    events[i].spe = spe_contexts[i];
    events[i].events = SPE_EVENT_OUT_INTR_MBOX;
    events[i].data.u32 = i; 
    spe_event_handler_register(handler, &events[i] );
    spe_program_load( spe_contexts[i], &spu );
    thread_args[i].spe_context = spe_contexts[i];
    thread_args[i].argp = buffer;
    thread_args[i].envp = buffer;

    pthread_create( &threads[i], NULL, &spe_thread, &thread_args[i] );

}

邮箱

自然，我们希望支持 SPE 和 PPE 之间的通信。这通常通过一个称为邮箱的功能来实现。邮箱用于发送短消息（32 位），从 SPE 发送到 PPE 或从 PPE 发送到 SPE，通常用于两者之间的同步。我们可以通过两种方式使用邮箱：

1. 阻塞方式。接收方在收到消息之前一直等待，在此期间不做任何事情。
2. 轮询方式。接收方定期检查是否有消息，然后执行一些工作，并重复此过程直到收到消息。

显然，性能更好的选择是第二种，因为它不会导致“忙等待”（PPU 等待消息，在此期间无法执行任何操作）。要初始化邮箱，需要少量代码，如下所示：

events[i].events = SPE_EVENT_OUT_INTR_MBOX;
// setting the type of events we use

spe_event_handler_register(handler, &events[i] );
//register the handler for the specified events

在邮箱中，PPE 将接收来自任何 SPE 的消息，所以我们想知道发送消息的确切 SPE。我们通过为每个 SPE 关联一个数字来实现这一点，这个数字可以与接收到的数据一起获得。

events[i].data.u32 = i;

在 PPE 中等待消息的操作如下：

spe_event_wait(handler, events_generated, NUM_THREADS, -1);
//printf("Got event! from spe no %d:", events_generated[0].data.u32);
spe_out_intr_mbox_read (events_generated[0].spe, 
   (unsigned int*) &data, 1, SPE_MBOX_ANY_BLOCKING);

从 PPE 发送消息的操作如下：

spe_in_mbox_write( events_generated[0].spe, value_p, 1, SPE_MBOX_ANY_BLOCKING);

DMA 访问

正如我之前所说，邮箱用于 PPE 和 SPE 之间的同步；这样 PPE 就能知道每个 SPE 在特定时间正在做什么。但是，也需要将数据发送到 SPE。SPE 擅长处理密集数据，因此一次发送 32 位数据是不切实际的。要将数据从 PPE 发送到 SPE，我们使用 DMA 访问。在普通的 x86 C++ 编程中，您不必显式使用 DMA，但像这样使用它提供了更好的控制、灵活性，从而提高了性能。

int tag = 30;  
int tag_mask = 1<<tag;
mfc_get((volatile void*)original_piece, (uint64_t)originalPieceAddrAsInt, 
         piece_height*piece_height*sizeof(int), tag, 0, 0);    
mfc_write_tag_mask(tag_mask);
mfc_read_tag_status_any();

实际传输是通过 mfc_get 函数完成的。此函数启动从 PPE 内存中的 originalPieceAddrAsInt 地址开始的 piece_height*piece_height*sizeof(int) 字节的传输到 SPE 正在运行的代码；数据从 original_piece 开始写入。标记是 DMA 传输的可选功能，它允许选择性地等待 DMA 传输完成。通常，当我们想等待 DMA 完成时，我们会使用 read_status 函数，如下所示：

mfc_read_tag_status_all();

当我们想等待某个特定的 DMA 时，我们在启动它时为其分配一个标签，并在等待它时应用一个标签掩码：

mfc_write_tag_mask(tag_mask);
mfc_read_tag_status_any();

标签掩码是一个整数，其中对应于所需传输的标签位置的位为 1。这对应于位移：

int tag_mask = 1<<tag;

SIMD 处理

Cell SPU 库提供了映射到 Cell 的内联汇编指令的函数；其中最有趣的是 SIMD 指令；它们允许一次对多个变量进行操作，特别是最多四个 32 位变量（整数或浮点数）。这些操作需要使用一种称为向量的新数据类型，该类型允许对这些变量进行分组。这些向量的主要限制是它们必须排列在内存中的 128 位边界上。这可以通过使用 aligned 属性对静态分配的数据来实现：

int temp[MAX_LEN] __attribute__ ((aligned(128)));

对于动态数据，则使用 memalign 函数：

piece_pos = (int*)memalign(128, piece_height*piece_height*sizeof(int));

向量类型包括：vector int、vector float、vector char 等，操作包括加法、减法、除法等。我们可以将两个向量 A 和 B 中的四个元素相加，并将结果存储在另一个向量 C 中：C[0]=A[0]+B[0] ...C[3]=A[3]+B[3]，使用 spu_add 函数，该函数以 A 和 B 作为参数并返回 C。更复杂的 SIMD 函数允许其他操作，例如同时比较四个 32 位数字：

// vi - vector int
cmp = spu_sub( *((vi*)puzzle_piece), *((vi*)original_piece));
zero = spu_orx(cmp);
if (*( (int*)(&zero) ) != 0) return 0; 

第二部分：解谜

本练习的目标是解决一个谜题。我们有两个初始图像：原始图像和一个修改后的图像，该图像可能是将原始图像分割成碎片，然后将碎片打乱并旋转（0、90、180、270 度）以生成新图像。已知谜题块的大小（以像素为单位），我们想要从两个图像中重构出原始图像。输出将是重构后的图像（以便我们能看到结果是否正确）以及一个文本文件，显示每个原始碎片在谜题图像中的位置。我们希望最大限度地利用 Cell 处理器的效率。此外，程序必须正确处理包含多个相同碎片的图像。

我的解决方案包括使用：

• 通过 SPE 进行并发
• DMA 访问
• 邮箱
• 双缓冲

我决定使用一种简单的图像格式：基于文本的 PPM (http://netpbm.sourceforge.net/doc/ppm.html) 格式；这种格式基本上使用 ASCII 文本存储的 RGB 像素格式，并用简单的分隔符（如空格）分隔。这使得理解图像格式本身的任务更容易，并允许我们专注于要执行的处理。虽然它不是最高效的格式。我们希望能够处理正常大小的图像（至少 1280*1024 分辨率），其碎片大小从 2*2 到 64*64，并且有 65500 种颜色（这样每个像素都适合一个整数）。

SPE 和 PPE 之间的基本通信协议如下：

1. PPE 启动 SPE 线程。
2. PPE 会记住当前发送到每个 SPE 的 original_piece 以及发送到每个 SPE 的当前 puzzle_piece。每个 SPE 将接收一个 original_piece，然后接收 puzzle_piece 直到它识别出一个拼图块与其原始块相同。对 original_piece 和 puzzle_piece 的请求将使用邮箱，实际数据将通过 DMA 发送。要请求另一个 puzzle_piece，SPU 发回 0。要宣布 puzzle_piece 和 original_piece 匹配，SPE 将返回 1、2、3 或 4，具体取决于识别出的旋转。当没有更多要比较的拼图块时，PPE 通过拒绝请求（发送 0）来响应 puzzle_piece 请求。如果还有更多的 puzzle_piece，它将发送指向拼图块所在内存区域的指针。

SPE 的主循环如下所示：

while ( NULL != (original = RequestNewOriginalPiece(length)) )
{
    FindPuzzlePieceForOriginalPiece(piece_height);
}

出现的问题是，如果我们想一次使用一个 DMA 传输来传输一个块，那么这个块必须位于内存中的连续区域。如果我们按照文件中的原始方式读取图像，那么拼图块将分散开；因此，我们需要通过更改每个像素的坐标，将每个拼图块读入一个数组。

for(j=0; j<height; j++)
    for(i=0; i< width; i++)
    {
        p_no_y = j / piece_height;
        piece_offset = (j - p_no_y*piece_height) * piece_height;

        p_no_x = i / piece_height;
        piece_offset += i - p_no_x*piece_height;  // pixel offset in piece

        piece_no = p_no_y* (width/piece_height) +p_no_x;
        // get number of piece from piece coordinates

        fscanf(f_original, "%d", &r);
        fscanf(f_original, "%d", &g);
        fscanf(f_original, "%d", &b);
        original[piece_no][piece_offset] = (r<<16) + (g<<8) + b;
        // cram pixel data in a single int
    }

SPE 的双缓冲是另一个有趣的区域：

turn++;
if (turn%2 == 1) 
{
    // wait for tranfer1
    // !!!value =  transfer1.addr;
    value =  (float*)transfer1.addr;
    // !!!printf("Waiting for transfer 1 at addr %d.\n", value);
    //printf("Waiting for transfer 1 at addr %d.\n", (int)value);
    if ( transfer1.addr != 0)
    {
        //printf("Transfer1 not 0.\n");
        //printf("Checking tranfer status for tag = %d",transfer1.tag);
        mfc_write_tag_mask(1<<transfer1.tag);
        mfc_read_tag_status_any();
        // use tranfer1
        //printf("Tranfer 1 data (%d %d %d) length %d\n", 
        //    transfer1.buffer[0],transfer1.buffer[1], transfer1.buffer[2], length);
        memcpy(input, transfer1.buffer, length*4); 
        // start transfer 2:
        // -request a new backup DMA tranfer address
        //printf("Sending request for transfer2.\n");
        spu_write_out_intr_mbox(match_result);
        addr = spu_read_in_mbox();
        //printf("Got addr=%d", addr);
        transfer2.addr = addr;
        // -initialize actual tranfer
        transfer2.tag = transfer1.tag + 1;
        if (transfer2.tag>30) 
        { 
            transfer2.tag -= 30;
            //printf("Decremented tag.\n");
        } 
        tag_mask = 1<<transfer2.tag;
        if (addr != 0)
        { 
                mfc_get((volatile void*)transfer2.buffer, 
                  (uint64_t)addr, length * 4, transfer2.tag, 0, 0);
        }
        else
        {
            //printf("Don't wait\n");
    }
}

我们在测试两个块是否相等时利用了 SIMD 功能：

int test_equality(int* puzzle_piece, int* original_piece, int piece_height)
{
    vi zero,cmp;
    int i = 0; 

    for(i=0; i<piece_height*piece_height/4; i++)
    {
        cmp = spu_sub( *((vi*)puzzle_piece), *((vi*)original_piece));
        zero = spu_orx(cmp);
        if (*( (int*)(&zero) ) != 0) return 0;
        puzzle_piece += 4;
        original_piece += 4;
    }
    return 1;
}

一个需要妥善处理的有趣情况是：假设图像包含多个相同的碎片。那么，原始图像中两个相同的区域必须用谜题图像中的两个碎片填充，**而不是**用同一个谜题碎片填充！如果程序是顺序执行的，我们可以通过不发送一个已经安装在图像中的谜题碎片来检查它是否适合另一个位置来简单地停止这个过程。但是，考虑到我们有 8 个并发指令流（因此同时处理 8 个碎片），这无法做到。处理这种情况的最简单方法是（当 SPE 返回正响应时）检查谜题碎片的位​​置是否未被使用。 （注意：在发送给 SPU 之前检查是不够的！）另一种更有效的方法是始终同时处理不同的碎片，但这需要更复杂的代码。

参考文献

• SDK 和编程标准在 IBM Cell Resource Center 的文档页面中进行了介绍。
• Cell Broadband Engine Programmer's Guide（已更新）
• Cell Broadband Engine Programming Tutorial（已更新）
• Cell Broadband Engine Programming Handbook
• Cell Broadband Engine Architecture 的 SIMD 数学库规范
• Cell Broadband Engine Architecture 的 C/C++ 语言扩展
• 布加勒斯特理工大学计算机科学系举行的“计算机系统结构”精彩课程。

许可证

代码根据专有许可证授权。除获得我的明确许可外，禁止使用此代码。