使稀疏矩阵乘法快速

Question

代码使用 C++11 编写。每个进程都有两个矩阵数据（稀疏）。测试数据可以从enter link description here下载

测试数据包含 2 个文件：a0（稀疏矩阵 0）和 a1（稀疏矩阵 1）。文件中的每一行都是“i j v”，表示稀疏矩阵第i行，第j列的值为v。i，j，v都是整数。

使用 c++11 unordered_map 作为稀疏矩阵的数据结构。

unordered_map<int, unordered_map<int, double> > matrix1 ;
matrix1[i][j] = v ; //means at row i column j of matrix1 is value v;

以下代码花费了大约 2 分钟。编译命令为g++ -O2 -std=c++11 ./matmult.cpp。

g++ 版本是 4.8.1，Opensuse 13.1。我的电脑信息：Intel(R) Core(TM) i5-4200U CPU @ 1.60GHz，4G 内存。

#include <iostream>
#include <fstream>
#include <unordered_map>
#include <vector>
#include <thread>

using namespace std;

void load(string fn, unordered_map<int,unordered_map<int, double> > &m) {
  ifstream input ;
  input.open(fn);
  int i, j ; double v;
  while (input >> i >> j >> v)  {
    m[i][j] = v;
  }
}

unordered_map<int,unordered_map<int, double> > m1;
unordered_map<int,unordered_map<int, double> > m2;
//vector<vector<int> > keys(BLK_SIZE);

int main() {
  load("./a0",m1);
  load("./a1",m2);

  for (auto r1 : m1) {
    for (auto r2 : m2) {
      double sim = 0.0 ;
      for (auto c1 : r1.second) {
        auto f = r2.second.find(c1.first);
        if (f != r2.second.end()) {
           sim += (f->second) * (c1.second) ;
        }
      }
   }
  }
  return 0;
}

上面的代码太慢了。我怎样才能让它运行得更快？我使用多线程。 新代码如下，编译命令为g++ -O2 -std=c++11 -pthread ./test.cpp。大约花了1分钟。 我希望它更快。

我怎样才能更快地完成任务？谢谢！

#include <iostream>
#include <fstream>
#include <unordered_map>
#include <vector>
#include <thread>

#define BLK_SIZE 8

using namespace std;

void load(string fn, unordered_map<int,unordered_map<int, double> > &m) {
  ifstream input ;
  input.open(fn);
  int i, j ; double v;
  while (input >> i >> j >> v)  {
    m[i][j] = v;
  }
}

unordered_map<int,unordered_map<int, double> > m1;
unordered_map<int,unordered_map<int, double> > m2;
vector<vector<int> > keys(BLK_SIZE);

void thread_sim(int blk_id) {
  for (auto row1_id : keys[blk_id]) {
    auto r1 = m1[row1_id];
    for (auto r2p : m2) {
      double sim = 0.0;
      for (auto col1 : r1) {
        auto f = r2p.second.find(col1.first);
        if (f != r2p.second.end()) {
          sim += (f->second) * col1.second ;
        }
      }
    }
  }
}

int main() {

  load("./a0",m1);
  load("./a1",m2);

  int df = BLK_SIZE - (m1.size() % BLK_SIZE);
  int blk_rows = (m1.size() + df) / (BLK_SIZE - 1);
  int curr_thread_id  = 0;
  int index = 0;
  for (auto k : m1) {
    keys[curr_thread_id].push_back(k.first);
    index++;
    if (index==blk_rows) {
      index = 0;
      curr_thread_id++;
    }
  }
  cout << "ok" << endl;
  std::thread t[BLK_SIZE];
  for (int i = 0 ; i < BLK_SIZE ; ++i){
    t[i] = std::thread(thread_sim,i);
  }
  for (int i = 0; i< BLK_SIZE; ++i)
    t[i].join();

  return 0 ;
}

Answer 1

在处理稀疏矩阵时，大多数时候使用比嵌套映射更有效的表示。典型的选择是压缩稀疏行 (CSR) 或压缩稀疏列 (CSC)。详情请见https://en.wikipedia.org/wiki/Sparse_matrix。

Answer 2

您尚未指定希望示例运行的时间或希望运行的平台。这些是本示例中的重要设计约束。

我可以想到几个方面来提高它的效率：-

1. 改进数据的存储方式
2. 改进多线程
3. 改进算法

第一点是针对系统存储稀疏数组的方式和使数据能够被读取的接口。当速度不重要但可能有更具体的可用数据结构可以解决这个问题时，嵌套 unordered_maps 是一个不错的选择。最好的情况是您可能会找到一个比嵌套地图提供更好的数据存储方式的库，最坏的情况是您可能不得不自己想出一些东西。

第二点是指语言支持多线程的方式。多线程系统的原始规范旨在独立于平台，可能会遗漏某些系统可能具有的便利功能。确定您想要定位的系统并使用操作系统线程系统。您可以更好地控制线程的工作方式，可能会减少开销，但会失去跨平台支持。

第三点需要做一些工作。考虑到数据的性质，您将矩阵相乘的方式真的是最有效的方式。我不是这些方面的专家，但这是值得考虑的事情，但需要付出一些努力。

最后，您始终可以非常具体地了解您所运行的平台，然后进入汇编编程的世界。现代 CPU 是复杂的野兽。他们有时可以并行执行操作。例如，您可以进行 SIMD 操作或并行整数和浮点操作。这样做确实需要深入了解正在发生的事情，并且有一些有用的工具可以帮助您。英特尔确实有一个名为 VTune 的工具（现在可能是别的东西），它可以分析代码并突出潜在的瓶颈。最终，您将希望通过为 CPU 寻找其他事情或改进算法（或两者兼而有之）来消除 CPU 空闲等待某事发生（例如等待来自 RAM 的数据）的算法区域。

最终，为了提高整体速度，您需要知道是什么降低了速度。这通常意味着知道如何分析您的代码并理解结果。 Profiler 是用于此目的的通用工具，但也有特定于平台的工具可用。

我知道这不是您想要的，但是快速编写代码真的很难而且非常耗时。