一种并行计算系统中的磁盘缓存方法及装置

摘要

本发明提供了一种并行计算系统中的磁盘缓存方法及装置。其中所述方法包括：预先分配记录数据和消息数据在所述处理数据内存区中各自所占的比例；在数据加载过程中若记录数据在处理数据内存区中的容量将要超过预先分配的比例时，则以Hash桶为单位将部分记录数据缓存到磁盘空间；在计算任务对记录数据的遍历访问过程中，若将要求访问的Hash桶位于磁盘空间，并且处理数据内存区中剩余的记录数据空间不足以载入所述将要求访问的Hash桶，则将处理数据内存区中已访问过的Hash桶逐个缓存到磁盘空间，直至释放的空间能够载入所述将要求访问的Hash桶。本发明能在基于BSP模型的并行迭代计算系统中实现数据向磁盘的自动化缓存。

说明书

技术领域

本发明涉及一种整体同步并行(Bulk Synchronous Parallel Computing  Model，简称BSP)计算模型技术领域，具体涉及一种基于BSP计算模型的并 行计算系统中的磁盘缓存方法及装置。

背景技术

BSP计算模型，又名大同步模型或BSP模型，由哈佛大学Viliant和牛津 大学Bill McColl（1990）提出。一个BSP模型包括由网络互联的多个处理单 元。每一个处理单元都有一个快速的本地内存，并且可以启动多个线程的计算 任务。一个BSP计算过程包括一系列的全局超级步的迭代过程。如图1所示， 一个超级步包括以下三个顺序的阶段：

（1）并发本地计算：在每一个处理单元上都进行若干计算任务。每一个 计算进程只使用存储于本地内存中的数据。各单元、各计算进程之间执行的本 地计算是相互独立的，是异步发生的。

（2）通信：在这一阶段，进程之间可交换数据。

（3）路障同步：当一个进程完成上述本地计算和通信后，到达这个点（路 障），等待所有其他进程都完成它们的通信阶段，称为大同步。

图1展示了超级步的具体过程，这些进程间的排列是没有顺序的，可以任 意方式映射到处理单元上。现有的基于BSP模型的并行迭代处理系统，主要 包括：Google的Pregel系统、Apache开源的HAMA系统以及Giraph系统， 以下进行简单介绍。

（1）Google的Pregel系统

Pregel系统是为了解决云环境下大规模图并行迭代计算而设计的一个可 扩展的、容错的、基于BSP模型的并行计算平台。

Pregel的计算过程是由一系列的迭代步骤组成，即超级步。在每一个超级 步中，系统框架会并行的调用用户自定义函数对存储于本地的每个顶点处理， 并将结果以消息形式发送给其他目标顶点。该函数描述了每一个顶点在一个超 级步中的行为。同时本地顶点可以接收和处理在上一个超级步中发送给该顶点 的消息，并且向其他的顶点发送消息，而在本超级步中发送出去的消息会被接 收的顶点在下一个超级步中进行处理。顶点还可以更新自身及其出边的状态。 消息则可以沿着出边发送给能够到达的顶点，也可以发送给任意它已知标识符 的顶点。

Pregel的输入是一个有向图，其中每一个顶点都由一个全局ID进行唯一 的标识。每一个顶点带有一个值以及一个出边的列表，每条边也都带有一个值 以及它的到达顶点的ID。

Pregel将一个图分成若干分片，每一分片包含一系列的顶点和它们自己的 出边。将某一个顶点划分到哪一个分片中仅仅取决于顶点的ID，这就使得给 定一个顶点就可以知道它所在的分片是哪一个，即使该分片在其他的机器上。 系统默认的分片函数是哈希函数，同时允许用户自定义的分配方式。

目前的Pregel系统，整个计算过程中的所有数据都是驻留在内存中，因此， 它是一个完全基于内存运行的BSP并行计算系统。

（2）HAMA系统

HAMA系统是一个基于Hadoop的针对大型矩阵和图运算的分布式框架。 HAMA目标是成为对不同科学应用的有力工具，通过简单的API接口为开发 者和研究者提供基础工具。HAMA目前已经被归为Hadoop的一个子项目。

HAMA采用了分层的架构模式，其中包含了三个组件：HAMA Core为矩 阵和图运算提供了原语，HAMA Shell提供了与用户交互的控制台，还有HAMA  API。

HAMA作为一个开源项目，目前最新的发布版本是0.4.0（Feb,5,2012）， 在基于BSP模型的图处理框架中仍然是一个完全基于内存运行的版本，和前 面的Pregel系统一样，在整个计算过程中的所有数据均驻留内存存储，当数据 量超过内存容量时，将抛出内存不足的异常而终止作业。

（3）Giraph系统

Giraph系统作为通用大规模数据处理平台，基于MapReduce模型开发的 Hadoop已广泛使用，Giraph正是建立在Hadoop基础之上的面向图处理的算法 库。从计算模型上看，Giraph可以视为MapReduce模型和BSP模型的结合体。

在Giraph中，一个图处理作业就是一个典型的只有Map任务的Hadoop 作业。在Map任务中，参考Pregel设计思想，嵌入了BSP模型以支持图处理 的特殊需求。Giraph利用了现有Hadoop框架，针对图应用的特点提供特有的 计算模式：

（1）作业启动、初始化和任务的初始化、调度分配、运行框架，沿用Hadoop 机制；

（2）通过Map任务中的内置循环控制实现图的迭代处理，整个图处理过 程只需要启动一次MapRedcue作业，载入一次原始数据，消息和数据驻留内 存，避免了传统Hadoop多次启动作业、任务分配、数据反复重载和Shuffle 过程的开销。

Giraph同样作为一个开源项目，目前最新的发布版本为0.1（Feb,6,2012）， 与Pregel、HAMA一样，在运行计算任务的整个过程中也是数据全部驻留内存， 当数据量超过内存容量时，抛出内存不足的异常而终止作业。

从以上所述可以看出，现有的基于BSP模型的并行迭代处理系统，主要 包括：Google的Pregel系统、Apache开源的HAMA系统以及Giraph系统， 运行本地计算过程中将待计算数据（图数据）和消息数据均是完全驻留在内存 中，不具有缓存到外存的能力，当在内存资源有限的集群系统中运行数据量超 过内存上限的作业时，均无法处理。

发明内容

有鉴于此，本发明实施例的目的是提供一种基于BSP计算模型的并行计 算系统中的磁盘缓存方法及装置，在基于BSP模型的并行迭代计算系统中实 现数据向磁盘的自动化缓存。

为解决上述技术问题，本发明实施例提供方案如下：

一种基于整体同步并行BSP计算模型的并行计算系统中的磁盘缓存方法， 应用于将计算任务的处理数据内存区中的数据缓存到磁盘空间，预先分配记录 数据和消息数据在所述处理数据内存区中各自所占的比例；

该方法包括：

在计算任务的数据加载阶段，根据各个记录数据的记录ID，为待处理的 记录数据建立Hash索引表，将载入的记录数据存放到处理数据内存区中对应 的Hash桶中，得到记录数据的Hash表RHT，并在数据加载过程中若记录数 据在处理数据内存区中的容量将要超过预先分配的比例时，则以Hash桶为单 位将部分记录数据缓存到磁盘空间；

在计算任务对记录数据的遍历访问过程中，按照预设顺序遍历访问RHT 中的每个Hash桶，其中，若将要求访问的Hash桶位于磁盘空间，并且处理数 据内存区中剩余的记录数据空间不足以载入所述将要求访问的Hash桶，则按 照长度由大到小的顺序，将处理数据内存区中已访问过的Hash桶逐个缓存到 磁盘空间，直至释放的空间能够载入所述将要求访问的Hash桶。

本发明实施例还提供了一种基于整体同步并行BSP计算模型的并行计算 系统中的磁盘缓存装置，应用于将计算任务的处理数据内存区中的数据缓存到 磁盘空间，该装置包括：

第一缓存单元，用于在计算任务的数据加载阶段，根据各个记录数据的记 录ID，为待处理的记录数据建立Hash索引表，将载入的记录数据存放到处理 数据内存区中对应的Hash桶中，得到记录数据的Hash表RHT，并在数据加 载过程中若记录数据在处理数据内存区中的容量将要超过预先分配的比例时， 则以Hash桶为单位将部分记录数据缓存到磁盘空间；

遍历单元，用于在计算任务对记录数据的遍历访问过程中，按照预设顺序 遍历访问RHT中的每个Hash桶，其中，若将要求访问的Hash桶位于磁盘空 间，并且处理数据内存区中剩余的记录数据空间不足以载入所述将要求访问的 Hash桶，则按照长度由大到小的顺序，将处理数据内存区中已访问过的Hash 桶逐个缓存到磁盘空间，直至释放的空间能够载入所述将要求访问的Hash桶。

从以上所述可以看出，本发明实施例提供的基于BSP计算模型的并行计 算系统中的磁盘缓存方法及装置，使得此类基于BSP模型的并行迭代处理系 统打破了完全基于内存运行的限制，在处理数据量超过内存限制时，能够自动 将溢出的部分缓存到磁盘文件，可以避免其他系统出现的抛出异常甚至作业失 败的问题，保证了系统的可运行性和扩展性。另外，本实施例提出的数据磁盘 缓存方法由于是基于内存优先的缓存策略，因此，在处理数据量少于内存限制 时可以实现完全基于内存运行的效果，所以性能也能达到其他内存系统的水 平；在处理数据量超过内存限制时，能以较小的扇入扇出代价，快速地从磁盘 读入内存进行计算。

附图说明

图1为BSP模型中一个超级步包括的三个顺序阶段的示意图；

图2为本发明实施例的并行计算系统中的磁盘缓存方法的流程示意图；

图3为本发明实施例中本地计算任务槽的示意图；

图4为本发明实施例中每个Task对应任务槽的可用内存划分示意图；

图5为本发明实施例中记录数据和消息数据的Hash索引及Hash桶组织形 式示意图；

图6为本发明实施例中为记录数据建立Hash的索引的过程示意图；

图7为本发明实施例中消息数据区生产消费模型的示意图；

图8为本发明实施例的并行计算系统中的磁盘缓存装置的结构示意图。

具体实施方式

为使本发明的目的、技术方案和优点更加清楚，下面将结合附图及具体实 施例对本发明进行详细描述。

本发明实施例所要解决的问题就是：如何在基于BSP模型的并行迭代计 算系统中支持数据向磁盘的自动化缓存，即：当内存能够容纳计算数据和消息 数据时，仍在内存中运行；当处理的数据量和消息量超出内存容量时，能够自 动将溢出的部分缓存到磁盘文件。更进一步的，本发明实施例还可以保证后续 计算过程需要这些数据时，能以较小的扇入扇出代价，快速地从磁盘读入内存 进行计算，从而提供系统对大规模数据的可运行性和扩展性。

如图2所示，本发明一个实施例提供了一种基于整体同步并行BSP计算 模型的并行计算系统中的磁盘缓存方法，应用于将计算任务的处理数据内存区 中的数据缓存到磁盘空间。这里，处理数据内存区是指计算任务的可用内存中 可供任务存储记录数据（图数据）和收发消息数据的数据空间。本实施例所述 方法预先分配记录数据和消息数据在所述处理数据内存区中各自所占的比例， 假设Θ表示计算任务的处理数据内存区的空间大小，β表示预先分配的记录数 据在处理数据内存区所占的比例，则消息数据所占比例为1-β，则处理数据内 存区中用于存储记录数据的记录数据空间的大小为βΘ，用于存储消息数据的 消息数据空间的大小为(1-β)Θ。如图2所示，该方法具体包括：

步骤21，在计算任务的数据加载阶段，根据各个记录数据的记录ID，为 待处理的记录数据建立Hash索引表，将载入的记录数据存放到处理数据内存 区中对应的Hash桶中，得到记录数据的Hash表RHT，并在数据加载过程中 若记录数据在处理数据内存区中的容量将要超过预先分配的比例时，则以 Hash桶为单位将部分记录数据缓存到磁盘空间；

本步骤中，计算当前到来的记录数据的记录ID所对应的Hash值，建立 Hash索引表，确定处理数据内存区中与所述HASH值对应的HASH桶，并将 当前到来的记录数据存放到所确定的HASH桶中。若当前到来的记录数据存 放到处理数据内存区将导致记录数据的容量超出βΘ，则按照HASH桶的长度 由大到小的顺序，将当前处理数据内存区中的HASH桶逐个缓存到磁盘空间， 直至当前到来的记录数据缓存到处理数据内存区时不会导致处理数据内存区 中的记录数据的容量超出βΘ。

步骤22，在计算任务对记录数据的遍历访问过程中，按照预设顺序遍历 访问RHT中的每个Hash桶，其中，若将要求访问的Hash桶位于磁盘空间， 并且处理数据内存区中剩余的记录数据空间不足以载入所述将要求访问的 Hash桶，则按照长度由大到小的顺序，将处理数据内存区中已访问过的Hash 桶逐个缓存到磁盘空间，直至释放的空间能够载入所述将要求访问的Hash桶。

上述步骤22中，在遍历访问RHT中的每个Hash桶的过程中，通常需要 针对当前访问的HASH桶遍历其中的记录数据并执行对应的计算，进而根据 计算结果更新RHT中的HASH桶。

以上说明了本实施例对于记录数据的载入及遍历过程。上述过程中通过预 先设置记录数据在处理数据内存区的比例，在记录数据量超过预设比例限制 时，能够自动将溢出的部分缓存到磁盘文件，可以避免其他系统出现的抛出异 常甚至作业失败的问题，保证了系统的可运行性和扩展性。

更进一步的，本实施例所述方法还可以针对消息数据执行相应的缓存处 理，此时该方法还包括以下步骤：

步骤23，在每个超级步的执行过程中，根据各个消息数据的目的记录ID， 为新加入的消息数据建立Hash索引表，将消息数据存放到处理数据内存区中 对应的Hash桶中，并在消息数据在处理数据内存区中的容量超过预先分配的 比例时，以Hash桶为单位将处理数据内存区中的部分消息数据缓存到磁盘空 间。

本步骤23中，新加入的消息数据包括上一轮迭代过程中已接收的消息数 据、当前迭代过程中接收的消息数据或当前迭代过程中新产生的待发送的消息 数据。本步骤中，将部分消息数据缓存到磁盘空间具体可以包括：

在每个超级步的执行过程中，在向处理数据内存区中加入当前迭代过程中 接收的消息数据或新产生的待发送的消息数据时，若新加入的消息数据将导致 消息数据在处理数据内存区中的总量超出(1-β)Θ，将处理数据内存区中的当 前迭代过程中接收的消息数据队列和上一轮迭代过程中已接收的消息数据队 列中优先级最高且最长的hash桶逐个缓存到磁盘空间，直至处理数据内存区 的空间满足要求，其中，当前迭代过程中接收的消息数据队列的优先级高于上 一轮迭代过程中已接收的消息数据队列。更进一步的，若处理数据内存区中的 待发送的消息数据队列占用的空间比例超出预设门限，则阻塞局部计算线程以 避免产生新消息，直至所述空间比例低于所述预设门限。

以上步骤中，可以根据预先设置的Hash函数对记录ID或目的记录ID进 行计算，得到对应的Hash值，建立Hash索引表，根据预设的Hash桶的映射 策略，将记录数据或消息数据存放到对应的Hash桶中，形成Hash桶文件。其 中，具体采用哪种Hash函数，以及采用何种Hash桶的映射策略，本实施例不 作限定。

可以看出，本实施例中，当记录数据或消息数据的容量超出预设门限（如 βΘ或(1-β)Θ）时，将把处理数据内存区中的部分记录数据或消息数据从内存 空间缓存到磁盘空间，以使得新载入的记录数据或新加入的消息数据在存放到 处理数据内存区时不会超出预设门限，从而实现了基于BSP模型的并行迭代 计算系统中数据向磁盘的自动化缓存。

为帮助更好地理解本发明实施例的上述步骤，以下通过更为详尽的描述对 本实施例的具体实施作进一步的说明。

1．内存划分模型

本实施例基于BSP模型的并行迭代处理系统中完成计算任务的单位是本 地计算任务进程（即Task），每台计算节点上允许启动若干个Task，这个上限 数量可以由系统管理员根据节点的计算资源（如内存、CPU和磁盘等）进行 配置，即指定所谓的任务槽数量（maxTaskSlot）。当用户内存大小确定后，每 个本地计算Task即是一个应用程序，配置了任务槽数量maxTaskSlot后，也就 确定了每个Task所占用的内存数量Ω。例如，Ω可以等于1GB、2GB或者4GB 等。假设Ω=2GB，则一台拥有16GB用户内存的计算节点的本地计算任务槽情 况如图3所示。

如图4所示，每个Task对应任务槽的可用内存Ω中包含三部分的内容： 程序和栈及其他空间41、用于存放处理数据的空间43、用于存放辅助数据的 空间42。程序和栈及其他空间41指的是一个应用程序中必须包含的存放程序 代码、栈、常量及其他控制信息等内容的内存空间；用于存放处理数据的空间 43指的就是可供任务存储图数据和收发消息数据的数据空间；用于存放辅助 数据的空间42指的就是可供任务存储图和消息数据的一些控制信息和索引结 构数据的空间。这三者中能够被本实施例提出的磁盘缓存技术中用到的真正数 据内存区（设为Θ）就是其中的处理数据空间43了。因此，本实施例引进一 个参数θ，表示可用的数据空间占Task的内存空间Ω的比例，即每个任务的处 理数据内存区（数据空间）大小为Θ=θ□Ω。

Task的处理数据内存区Θ中所要管理暂存的数据包括：记录数据（Records  Data，即待处理的原始数据）和消息数据（Messages Data，即BSP模型中计 算过程中的通信消息）。记录数据中是一条条长短不一的由记录ID和各个维度 列表组成的类似邻接表形式的多维数据记录。根据BSP模型的迭代特点可知， 消息数据（Messages Data）又具体包括三类消息队列：当前迭代过程中接收的 消息队列（Incoming Queue）、上轮迭代过程中已接收到的消息队列（Incomed  Queue）和待发送消息队列（Outgoing Queue）。

考虑到把处理数据内存区Θ看成一个整体，管理这么多种类型的数据在内 存和磁盘之间的置换关系将会既混乱又复杂，所以，为了简化以后的数据调度 模型，本实施例再引入一个参数β，它表示处理数据内存区Θ中，记录数据 （Records Data）所占的比例系数，则消息数据（Messages Data）所占的比例 即为1-β。这样，本实施例就把要管理的内存区域划分成了两个部分。一个 Task中的内存划分模型如图4所示。

本实施例将处理数据内存区划分成两个部分，这样就无需再考虑记录数据 与消息数据之间的关系，而用一个比例系数β来调节两者之间的比例。那么对 于这两者各自的内存管理和调度策略就可遵循一条相同的原则：优先将数据存 放在内存，当数据量超出内存容量时将多出的部分缓存到磁盘上即可。

2．数据索引模型

根据BSP模型可知，每个超步的本地并行计算过程中，需要遍历每条记 录数据，并对它接收到的所有消息，进行遍历、计算处理。如此，对每条记录 数据，需要确定哪些消息是该记录数据上一步迭代接收到的所有消息。如此， 若不对记录数据和消息数据进行索引，每个超步中的数据查询代价就是复杂度 O(n*m)。这里，n表示记录数据的数量，m表示每个记录数据接收到的消息数 据的平均数量。对于完全基于内存的系统来说，这样的查询代价已经很高，而 对于带有磁盘缓存的系统来说，不断的扫描遍历磁盘上的缓存数据更是难以容 忍的开销。因此，本实施例对记录数据和消息数据分别建立索引。

实际上，对于记录数据和消息数据的索引方式可以采用很多种：二叉树、 B+树、Hash索引、顺序表等等。如果记录ID的值类型是连续编号的整数时， 采用树的结构索引顶点可能会获得较高的效率，但是考虑到方法的通用性，例 如当记录的ID是URL字符串时，基于数值比较的树形索引就不再适用，因此 本实施例优选地采用了对于多种情况下都能适用的Hash索引。

考虑BSP模型的特性，每个超步中记录数据和消息数据要做的相当于一 次连接（join）操作，本实施例考虑将两者都采用记录ID值作为索引的键值， 从而可以将两者连接起来。本实施例在具体实现时：

首先，在数据的加载阶段为待处理数据记录按照其ID值建立Hash索引表， 将记录数据组织成一系列的Hash桶，当记录数据需要缓存时，一部分存储在 内存，另一部分以Hash桶文件形式缓存在磁盘上。

其次，在每个超步的执行中，将接收到的消息数据（包括incoming、 incomed、outgoing消息数据）同样按照其目的记录ID的Hash值，划分到不 同的Hash桶中，当消息数据的容量超过内存时，消息数据将以Hash桶为单位 缓存到磁盘上。

图5展示了经过上述步骤处理后，记录数据和消息数据的Hash索引及 Hash桶组织方式。

如此，每次超步的本地计算任务需要对某个数据记录对应的所有消息处理 时，就只需按照Hash桶为单位遍历数据记录：给定某个目的记录ID值，首先 可以根据ID值的Hash值以O(1)的代价找到存放所有以此ID为目的记录的消 息的Hash桶，然后，再顺序遍历该Hash桶中的不同记录ID，从而得到该目 的记录的消息，这个代价为O(Hd)，Hd为平均每个Hash桶中不同记录的数量。 当Hash桶的数量很多的时候，Hd趋向于1，则获取消息的代价就趋向于O(1)； 当Hash桶的数量很少的时候，Hd趋向于n，即只有一个Hash桶，退化到所 有消息存放到一起，则获取消息的代价就趋向于O(m)，即每次都需要遍历消 息列表。

建立了Hash索引后，记录数据和消息数据在内存和磁盘之间的读入、写 出的调度单位就转化成了Hash桶，类似于内存管理中的页面文件扇入扇出。 其调度方法决定缓存机制的有效性。

下面将详细阐述本实施例中基于内存划分模型和Hash索引模型的记录和 消息数据索引的建立过程及其在超步迭代计算中的Hash桶调度算法。

3．记录数据的磁盘缓存方法

记录数据的磁盘缓存部分需要解决两个问题：每个任务Task在载入数据 的过程中如何建立对记录的Hash索引并存储这些Hash桶；在每个超级步的迭 代计算中，如何根据Hash索引机制遍历这些记录。

3.1记录数据Hash索引的建立

对于系统中的每个任务Task来说，待处理的记录数据来自于数据划分模 块从HDFS或者HBase等数据的存储介质中读取出来，并根据一定的策略一 条一条记录数据发送给自己的。因此，原始记录数据的加载阶段，每条记录数 据不会按照定义的Hash索引的顺序聚簇的到来，也就是在Task为本地所处理 的记录数据建立Hash索引的过程是一个动态的处理随机数据的过程，如图6 所示，而且，在这个过程中，按照随机顺序到达的第i个记录数据R_i被放到它 所属于的Hash桶H_j的时候，可能会出现当前内存中的数据总量（假设当前 时间为t）已经达到了记录数据部分的内存上限βΘ的情况，这时需要将某些 Hash桶缓存到磁盘空间，为接下来可能到来的记录数据释放足够的空间。

3.1.1MF-RHIC算法的动态建立部分

一条记录数据由两部分组成：记录头数据和各维度数据。其中，记录头数 据一般包括记录头的ID值（也是记录的整体ID值）和记录头的Value值；各 维度数据列表则包含了若干条由ID值和Value值组成的元组。由于每一条记 录的维度列表长度可能大不相同，有些记录可能只有几条维度记录，而有些记 录可能有数万条维度记录。例如，以社交网络的用户关注关系图数据为例， facebook、twitter等网站上用户之间的关注视为一条边，则有些明星或知名机 构的关注者可能有数以万计的关注者。因此，即使本实施例可以通过对记录的 ID值的Hash函数进行均衡化，尽量使每个Hash桶中的顶点数均匀，但是， 由于它们的边数可能相差悬殊的原因，不同Hash桶的长度还是有可能相差很 大。那么，当时，本实施例选取当前长度最长的Hash桶H_m缓存到磁盘 其中从而释放最多的空间给后续到达的记录数 据。这里，length(H_m)、length(H_j)分别表示Hash桶H_m和H_j的长度，h表示记录 数据的Hash桶的数量，N_h表示所有的Hash桶的数量。

考虑到当H_m被缓存到磁盘以后，有可能又到来了Hash值属于H_m的记录 数据R_k，本实施例进一步维护一张所有记录数据的Hash桶状态的元数据表， 该元数据表中记录有各个记录数据的HASH桶存放在处理数据内存区和/或磁 盘空间的信息。这样，若根据所述元数据表，确定当前到来的记录数据对应的 HASH桶仅存在于磁盘空间，则在处理数据内存区中重新建立当前到来的记录 数据对应的HASH桶，并将当前到来的记录数据存放于其中，并更新所述元 数据表中对应的HASH桶的元数据；若根据所述元数据表，确定当前需要缓 存到磁盘空间的HASH桶在磁盘空间中已经存在时，则将当前需要缓存到磁 盘空间的HASH桶中的数据合并至磁盘空间中已存在的该HASH桶中，并更 新所述元数据表中对应的HASH桶的元数据。

例如，在上述记录R_k到来后，本实施例在内存中重新建立Hash桶H_m，并 更新元数据表中H_m桶对应于H_m的元数据Meta_m的信息，记录它有一部分在内 存中，另外一部分已经被缓存到磁盘，还可以记录各部分的长度等信息。当这 个Hash桶H_m再次需要从内存空间被缓存到磁盘的时候，内存中的新记录直接 与磁盘上的Hash桶文件合并，追加到文件的末尾即可。

因此，本实施例根据以上描述的思想提出了基于内存优先（MF）的记录 数据Hash索引建立算法（Memory First-Record Hash Index Creation，下文中简 称为MF-RHIC算法）。MF-RHIC算法分为两个部分：动态建立部分和静态整 理部分。其中动态建立部分的一种具体的算法实现可以参考表1所示。

表1MF-RHIC算法-动态建立部分

3.1.2MF-RHIC算法的静态整理部分

根据MF-RHIC算法的动态建立部分可见，Task的记录数据加载阶段完成 后，即图数据的Hash索引动态建立过程结束后，图Hash表中的各个Hash桶 的状态可能都是内存中一部分数据并且磁盘上一部分数据。这样的状态不利于 以后的超步迭代计算时对记录数据的整体遍历，因为遍历的基本单位是Hash 桶，因此，为了提高遍历的效率，本实施例还进一步提出了MF-RHIC算法的 静态整理部分，与前面的动态建立部分合并起来构成完整的MF-RHIC算法。

具体的，本实施例在计算任务的数据加载完毕后，按照处理数据内存区中 的HASH桶的长度由大到小的顺序，逐个判断处理数据内存区中的HASH桶 是否也存在于磁盘空间，若是，则进一步判断将该HASH桶在磁盘空间中的 数据存放到处理数据内存区时是否会导致记录数据的容量超出βΘ，如果不超 出，则将该HASH桶在磁盘空间中的数据合并到处理数据内存区中，如果超 出，则将该HASH桶在处理数据内存区中的数据合并到磁盘空间中。

可见，静态整理部分的主要思想就是，遍历一遍记录数据Hash表（RHT， Record Hash Table），在内存优先的基本原则基础上，尽量将Hash桶完整的都 放在内存或者都放在磁盘，一种具体的算法实现可以参考表2所示。

表2MF-RHIC算法-静态整理部分

3.2记录数据的遍历计算

BSP并行迭代处理系统的处理过程分为一个个相同操作的超级步，在每个 超级步中本地计算任务Task需要遍历访问本地的记录数据。基于前文中设计 了记录数据的Hash索引形式，本实施例在每个超级步中，本地计算任务Task 需要遍历访问本地的记录数据时，可以按照预设顺序遍历访问记录数据的 Hash表RHT中的每个Hash桶，针对当前访问的HASH桶遍历其中的记录数 据并执行对应的计算，以及根据计算结果更新RHT中的HASH桶。并且，若 所述RHT超出βΘ，则在遍历所述RHT的过程中，若将要求访问的HASH桶 位于磁盘空间，则按照HASH桶的长度由大到小的顺序，将内存空间中已访 问过的HASH桶逐个缓存到磁盘空间，直至释放的内存空间能够载入所述将 要求访问的HASH桶。

也就是说，在遍历记录数据时的单位就是一个Hash桶，即，每个超步中， 任务进程按照某个顺序依次访问RHT中的每个Hash桶H_j，为每个H_j遍历其 中的记录，遍历的过程不仅要读取记录，在对每条记录R_i调用一次用户compute 函数之后，可能会产生对R_i中的value值等信息的更新，因此，遍历计算后的 记录数据需要写回RHT，即遍历过的Hash桶H_j还需要写回到内存或者磁盘。

当任务可用的记录数据内存区不足以放下整个RHT的情况时，即S_R＞βΘ （其中S_R表示RHT对应的长度），遍历RHT的过程中就需要进行Hash桶的 内存与磁盘之间的置换，即将已经访问过的Hash桶缓存到磁盘，直至释放空 间足以从磁盘载入尚未访问过的Hash桶。那么，与前面建立Hash索引时的内 存优先的写磁盘策略一致，当发现当前将要访问的Hash桶H_t不在内存，则需 要将当前内存中长度最长的Hash桶H_m缓存到磁盘上，如果释放的内存空间仍 然不够H_t载入内存则循环这个过程，直到内存空闲区的长度超过H_t的长度为 止。

基于MF的置换策略同时也要求了遍历RHT的顺序，应该是一种内存长 度最长优先（Memory Longest First，缩写MLF）的遍历顺序，这也是本实施 例提出的带有磁盘缓存的记录数据的遍历算法的基本思想。基于以上的思想和 策略，本实施例提出了MLF记录数据遍历算法，其一种具体实现可以参考表 3所示。

表3MLF记录数据遍历算法

4．消息数据的磁盘缓存方法

根据BSP模型的迭代性质，每个任务需要管理的消息队列一共有三个， 即待发送消息队列、本轮迭代过程中的接收消息队列和上轮迭代过程中接收到 的消息队列。那么，内存模型中的消息数据（Messages Data）部分，即消息数 据内存区（容量(1-β)Θ）需要存放的内容就是这三个消息队列。本实施例提 出的记录数据和消息数据的磁盘缓存技术的基本思想是基于内存优先（MF） 策略的，因此，消息数据的磁盘缓存技术的基本思想也是优先把所有的消息队 列都存储在内存中，如果出现当前消息数据的总量大于消息数据内存区的 容量的情况时（），才会将一部分消息数据缓存到磁盘上。

下面将说明消息数据的磁盘缓存部分在内存占满时如何缓存不同消息队 列以及具体的缓存调度算法。

4.1消息队列的缓存优先级

对于基于BSP模型的并行迭代系统来说，消息数据区需要存放三个队列， 即本轮迭代过程中的待发送消息队列（Outgoing Queue）、本轮迭代过程中接 收的接收消息队列（Incoming Queue）和上轮迭代过程中接收到的消息队列 （Incomed Queue）。在第n个超步的执行过程中，局部计算模块不断的从接收 到的消息队列Incomed Queue（即第n-1个超步的接收消息队列）中取走消息， 经过调用用户定义的执行逻辑compute函数，产生待发送的消息再放入到待发 送消息队列Outgoing Queue中；同时，消息通信模块的Sender线程会不断的 从待发送消息队列Outgoing Queue中取走消息并通过网络发送出去；消息通 信模块的Receiver线程会不断的通过网络接收到发送给本地任务的消息，并把 它们放入到接收消息队列Incoming Queue中。这样就对消息数据区构成了两 对的生产者和消费者模型，如错误！未找到引用源。所示。

虽然这里的Incoming queue的长度是一直增加的，而Incomed queue的长 度是一直减少的，把三个消息队列所占用的总空间看成一个整体，针对整个消 息数据区而言，可以把它们看做两个生产者和两个消费者的模型。那么，为了 确定当消息数据区的内存容量不足时写磁盘的策略，需要分析这三个消息队列 的优先级。

首先，对于Incoming queue来说，根据BSP模型的定义可知，本超步接 收的消息必须要到下一个超步的计算中才会使用到，因此，本实施例将 Incoming queue被溢出到磁盘上的优先级设置为最高。

其次，对于Outgoing queue来说，假如待发送的消息被缓存至磁盘，那么 在本超步中必须再从磁盘读出来并通过网络发送出去，这是由于BSP模型的 限制，必须所有的任务的通信阶段都结束，即消息都被发送到目的地，才能退 出路障同步进入下一个超步的计算中，因此可见，对于待发送的消息来说，中 途将它们溢出到磁盘只会比它们不溢出磁盘增加了一次额外的磁盘I/O的开 销。而且，能够造成Outgoing queue长度溢出的原因只有它的消费者即Sender 的发送速率低于它的生产者即局部计算的计算速率，而根据BSP模型的同步 机制可知，即使通过增加对发送消息的缓存能力而允许局部计算模块提前与通 信模块结束，整个超步仍然要等待通信模块将待发消息全部发送完毕才能结 束，因此，向磁盘缓存Outgoing queue只会增加任务开销，不会降低开销。所 以，本实施例中Outgoing queue被溢出的磁盘的优先级是最低的，而且通常它 是不应该被缓存到磁盘的，如果出现整个消息数据区被Outgoing queue占满的 情况，则本实施例将阻塞局部计算线程以等待通信模块将积压的消息都发送出 去之后再继续进行。

基于以上分析，本实施例将Incomed queue被溢出到磁盘的优先级设置为 介于前两者之间。然而，Incomed queue与前两者仍有不同，本地计算在执行 计算的过程中，需要遍历图数据的Hash桶，并为每一个当前访问的记录数据 获得上一个超步发送给它的所有接收到的消息，即根据该记录的ID值到对应 的Incomed queue的Hash桶中查询目的顶点ID相同的所有消息数据。因此， 本实施例要保证在当前消息数据内存区中可以存放至少一个Incomed queue的 Hash桶，也就是说，当前正在访问的Incomed queue的Hash桶不能被缓存到 磁盘。

综上所述，三个消息队列被溢出到磁盘的优先级顺序是：P(Incoming)> P(Incomed)>P(Outgoing)。同时，Incomed queue至少在内存中驻留一个当前 正在访问的Hash桶，Outgoing queue则不会被溢出到磁盘。本实施例正是根 据这样的优先级关系，提出了基于优先级的消息队列的磁盘缓存策略及算法。

4.2基于消息队列优先级的磁盘缓存方法

根据前面的生产消费模型可知，只有当Receiver向消息数据区中加入接收 到的消息时和当本地计算模块向消息数据区加入新产生的待发消息时，这两种 情况时，才有可能达到消息数据区溢出的条件，因为只有它们两者 是内存消息区的生产者。因此，当这两者每新放入一条消息时，需要检测当前 的值，如果达到阈值，则根据它们的优先级关系选取一个消息队列，再按 照由大到小的顺序从该消息队列中选取一个最长的Hash桶，缓存到磁盘文件 中，直至新放入的消息数据在放入时不会溢出。

更进一步的，考虑到一种边界情况：当任务的消息数据内存区已经被当前 的Incomed queue和Outgoing queue几乎占满时，这时每到达少量的接收消息 就会使S_M达到阈值上限，如果仅仅根据优先级的话，一定是缓存Incoming  queue的最长Hash桶，然而很可能此时Incoming queue的最长Hash桶中也仅 仅只有几条消息，因此释放很少的空间，从而造成很快又启动磁盘缓存，以此 类推，形成一个恶性循环，释放越少的内存，就越多的启动缓存机制，从而导 致频繁的向磁盘中写入少量数据并打开和关闭Hash桶文件。

为克服上述不足，本实施例在当前迭代过程中接收的消息数据（Incoming  queue）队列中最长的hash桶的长度低于处理数据内存区中消息数据的平均 Hash桶的长度阈值时，如果在向处理数据内存区新加入的消息数据将导致消 息数据在处理数据内存区中的总量超出(1-β)Θ，则按照HASH桶长度的由大 到小的顺序，将上一轮迭代过程中已接收的消息数据队列中的hash桶逐个缓 存到磁盘空间，直至处理数据内存区的空间满足要求，其中，N_h表示hash桶 的数量。

也就是说，本实施例在基于优先级的基础上，还为单个Hash桶设置了一 个平均长度阈值，即如果Incoming queue的最长Hash桶长度低于消息数据区 的平均每个Hash桶的长度阈值时，就违反优先级，而先缓存Incomed queue 的最长Hash桶。作为一种优选实施方式，该阈值可以设置为(1-β)Θ/(3N_h)， 其中N_h表示Hash桶的数量。

根据以上的基本思想，本实施例提出了基于消息队列优先级的消息数据的 磁盘缓存算法。算法的启动时机就是前面所述的每向消息数据区中增加一条消 息。一种具体的算法实现可以参考表4所示。

表4基于消息队列优先级的消息数据的磁盘缓存算法

以上详细说明了本发明实施例的具体实现，从以上所述可以看出，本发明 实施例提供的磁盘缓存方法，包含了待处理记录数据和处理过程中的消息数据 的共同基于内存优先策略的数据缓存处理，使得此类基于BSP模型的并行迭 代处理系统打破了完全基于内存运行的限制，保证了系统的可运行性和扩展 性。另外，本实施例提出的数据磁盘缓存方法由于是基于内存优先的缓存策略， 因此，在处理数据量少于内存限制时可以实现完全基于内存运行的效果，所以 性能也能达到其他内存系统的水平；在处理数据量超过内存限制时，则可以避 免其他系统出现的抛出异常甚至作业失败的问题。

基于以上提供的磁盘缓存方法，本发明实施例还提供了一种基于BSP计 算模型的并行计算系统中的磁盘缓存装置，应用于将计算任务的处理数据内 存区中的数据缓存到磁盘空间。如图8所示，该装置包括：

第一缓存单元，用于在计算任务的数据加载阶段，根据各个记录数据的记 录ID，为待处理的记录数据建立Hash索引表，将载入的记录数据存放到处理 数据内存区中对应的Hash桶中，得到记录数据的Hash表RHT，并在数据加 载过程中若记录数据在处理数据内存区中的容量将要超过预先分配的比例时， 则以Hash桶为单位将部分记录数据缓存到磁盘空间；

遍历单元，用于在计算任务对记录数据的遍历访问过程中，按照预设顺序 遍历访问RHT中的每个Hash桶，其中，若将要求访问的Hash桶位于磁盘空 间，并且处理数据内存区中剩余的记录数据空间不足以载入所述将要求访问的 Hash桶，则按照长度由大到小的顺序，将处理数据内存区中已访问过的Hash 桶逐个缓存到磁盘空间，直至释放的空间能够载入所述将要求访问的Hash桶。

上述装置还可以实现对消息数据的缓存处理，此时该装置还包括：

第二缓存单元，用于在每个超级步的执行过程中，根据各个消息数据的目 的记录ID，为新加入的消息数据建立Hash索引表，将消息数据存放到处理数 据内存区中对应的Hash桶中，并在消息数据在处理数据内存区中的容量超过 预先分配的比例时，以Hash桶为单位将处理数据内存区中的部分消息数据缓 存到磁盘空间。

作为一种优选实施方式，图1中所述第一缓存单元包括：

计算存放单元，用于计算当前到来的记录数据的记录ID所对应的Hash 值，建立Hash索引表，确定处理数据内存区中与所述HASH值对应的HASH 桶，并将当前到来的记录数据存放到所确定的HASH桶中；

转移单元，用于在当前到来的记录数据存储到处理数据内存区将导致记录 数据的容量超出βΘ时，按照长度由大到小的顺序，将当前处理数据内存区中 的HASH桶逐个缓存到磁盘空间，直至当前到来的记录数据存储到处理数据 内存区时不会导致记录数据的容量超出βΘ，其中，Θ表示计算任务的处理数 据内存区的空间大小，β表示预先分配的记录数据在处理数据内存区所占的比 例。

优选地，上述装置还包括：

表项维护单元，用于维护一张所有记录数据的Hash桶状态的元数据表， 该元数据表中记录有各个记录数据的HASH桶存放在处理数据内存区和/或磁 盘空间的信息；

所述计算存放单元，进一步用于在根据所述元数据表，确定当前到来的记 录数据对应的HASH桶仅存在于磁盘空间时，在处理数据内存区中重新建立 当前到来的记录数据对应的HASH桶，并将当前到来的记录数据存放于其中；

所述转移单元，进一步用于在根据所述元数据表，确定当前需要缓存到磁 盘空间的HASH桶在磁盘空间中已经存在时，将当前需要缓存到磁盘空间的 HASH桶中的数据合并至磁盘空间中已存在的该HASH桶中；

静态整理单元，用于在计算任务的数据加载完毕后，按照处理数据内存区 中的HASH桶的长度由大到小的顺序，逐个判断处理数据内存区中的HASH 桶是否也存在于磁盘空间，若是，则进一步判断将该HASH桶在磁盘空间中 的数据存放到处理数据内存区时是否会导致记录数据的容量超出βΘ，如果不 超出，则将该HASH桶在磁盘空间中的数据合并到处理数据内存区中，如果 超出，则将该HASH桶在处理数据内存区中的数据合并到磁盘空间中；

本实施例中，所述遍历单元，还用于在遍历访问RHT中的每个Hash桶的 过程中，针对当前访问的HASH桶遍历其中的记录数据并执行对应的计算， 以及根据计算结果更新RHT中的HASH桶。

所述第二缓存单元，还用于在每个超级步的执行过程中，在向处理数据内 存区中加入当前迭代过程中接收的消息数据或新产生的待发送的消息数据时， 若新加入的消息数据将导致消息数据在处理数据内存区中的总量超出 (1-β)Θ，将处理数据内存区中的当前迭代过程中接收的消息数据队列和上一 轮迭代过程中已接收的消息数据队列中优先级最高且最长的hash桶逐个缓存 到磁盘空间，直至处理数据内存区的空间满足要求，其中，当前迭代过程中接 收的消息数据队列的优先级高于上一轮迭代过程中已接收的消息数据队列。

优选地，所述第二缓存单元，还用于在处理数据内存区中的待发送的消息 数据队列占用的空间比例超出预设门限时，阻塞局部计算线程以避免产生新消 息，直至所述空间比例低于所述预设门限。

优选地，所述第二缓存单元，还用于在当前迭代过程中接收的消息数据队 列中最长的hash桶的长度低于处理数据内存区中消息数据的平均Hash桶的长 度阈值时，如果向处理数据内存区新加入的消息数据将导致消息数据在处理数 据内存区中的总量超出(1-β)Θ，则按照HASH桶长度的由大到小的顺序，将 上一轮迭代过程中已接收的消息数据队列中的hash桶逐个缓存到磁盘空间， 直至处理数据内存区的空间满足要求，其中，其中，Θ表示计算任务的处理数 据内存区的空间大小，β表示预先分配的记录数据在处理数据内存区所占的比 例。

此说明书中所描述的许多功能部件都被称为模块、单元，以便更加特别地 强调其实现方式的独立性。

本发明实施例中，模块、单元可以用软件实现，以便由各种类型的处理器 执行。举例来说，一个标识的可执行代码模块可以包括计算机指令的一个或多 个物理或者逻辑块，举例来说，其可以被构建为对象、过程或函数。尽管如此， 所标识模块的可执行代码无需物理地位于一起，而是可以包括存储在不同位里 上的不同的指令，当这些指令逻辑上结合在一起时，其构成模块并且实现该模 块的规定目的。

实际上，可执行代码模块可以是单条指令或者是许多条指令，并且甚至可 以分布在多个不同的代码段上，分布在不同程序当中，以及跨越多个存储器设 备分布。同样地，操作数据可以在模块内被识别，并且可以依照任何适当的形 式实现并且被组织在任何适当类型的数据结构内。所述操作数据可以作为单个 数据集被收集，或者可以分布在不同位置上（包括在不同存储设备上），并且 至少部分地可以仅作为电子信号存在于系统或网络上。

在模块可以利用软件实现时，考虑到现有硬件工艺的水平，所以可以以软 件实现的模块，在不考虑成本的情况下，本领域技术人员都可以搭建对应的硬 件电路来实现对应的功能，所述硬件电路包括常规的超大规模集成（VLSI） 电路或者门阵列以及诸如逻辑芯片、晶体管之类的现有半导体或者是其它分立 的元件。模块还可以用可编程硬件设备，诸如现场可编程门阵列、可编程阵列 逻辑、可编程逻辑设备等实现。

以上所述仅是本发明的实施方式，应当指出，对于本技术领域的普通技术 人员来说，在不脱离本发明原理的前提下，还可以作出若干改进和润饰，这些 改进和润饰也应视为本发明的保护范围。