关于线程的问题
Posted
技术标签:
【中文标题】关于线程的问题【英文标题】：Questions About Threads 【发布时间】：2011-05-01 19:26:24 【问题描述】：我是线程编程的新手，我有一个概念问题。我正在为我的班级做矩阵乘法。但是，我不使用线程，然后使用线程计算答案矩阵的每个单元格的标量积，然后再次将第一个矩阵拆分为比例，以便每个线程都有相等的部分要计算。我的问题是标量产品实现完成得非常快，这是我所期望的，但是第三个实现并没有比非线程实现更快地计算出答案。例如，如果它使用 2 个线程，它将在大约一半的时间内计算它，因为它可以同时在矩阵的两半上工作，但根本不是这种情况。我觉得第三个实现有问题，我不认为它是并行运行的，代码如下。谁能让我直截了当？并非所有代码都与问题相关，但我将其包括在内，以防问题不是本地问题。 谢谢，
主程序：
#include <iostream>
#include <cstdio>
#include <cstdlib>
#include <cmath>
#include<fstream>
#include<string>
#include<sstream>
#include <matrix.h>
#include <timer.h>
#include <random_generator2.h>
const float averager=2.0; //used to find the average of the time taken to multiply the matrices.
//Precondition: The matrix has been manipulated in some way and is ready to output the statistics
//Outputs the size of the matrix along with the user elapsed time.
//Postconidition: The stats are outputted to the file that is specified with the number of threads used
//file name example: "Nonparrallel2.dat"
void output(string file, int numThreads , long double time, int n);
//argv[1] = the size of the matrix
//argv[2] = the number of threads to be used.
//argv[3] =
int main(int argc, char* argv[])
random_generator rg;
timer t, nonparallel, scalar, variant;
int n, total = 0, numThreads = 0;
long double totalNonP = 0, totalScalar = 0, totalVar = 0;
n = 100;
/*
* check arguments
*/
n = atoi(argv[1]);
n = (n < 1) ? 1 : n;
numThreads = atoi(argv[2]);
/*
* allocated and generate random strings
*/
int** C;
int** A;
int** B;
cout << "**NOW STARTING ANALYSIS FOR " << n << " X " << n << " MATRICES WITH " << numThreads << "!**"<< endl;
for (int timesThrough = 0; timesThrough < averager; timesThrough++)
cout << "Creating the matrices." << endl;
t.start();
C = create_matrix(n);
A = create_random_matrix(n, rg);
B = create_random_matrix(n, rg);
t.stop();
cout << "Timer (generate): " << t << endl;
//---------------------------------------------------------Ends non parallel-----------------------------
/*
* run algorithms
*/
cout << "Running non-parallel matrix multiplication: " << endl;
nonparallel.start();
multiply(C, A, B, n);
nonparallel.stop();
//-----------------------------------------Ends non parallel----------------------------------------------
//cout << "The correct matrix" <<endl;
//output_matrix(C, n);
cout << "Timer (multiplication): " << nonparallel << endl;
totalNonP += nonparallel.user();
//D is the transpose of B so that the p_scalarproduct function does not have to be rewritten
int** D = create_matrix(n);
for (int i = 0; i < n; i++)
for(int j = 0; j < n; j++)
D[i][j] = B[j][i];
//---------------------------------------------------Start Threaded Scalar Poduct--------------------------
cout << "Running scalar product in parallel" << endl;
scalar.start();
//Does the scalar product in parallel to multiply the two matrices.
for (int i = 0; i < n; i++)
for (int j = 0; j < n; j++)
C[i][j] = 0;
C[i][j] = p_scalarproduct(A[i],D[j],n,numThreads);
//ends the for loop with j
scalar.stop();
cout << "Timer (scalar product in parallel): " << scalar << endl;
totalScalar += scalar.user();
//---------------------------------------------------Ends Threaded Scalar Poduct------------------------
//---------------------------------------------------Starts Threaded Variant For Loop---------------
cout << "Running the variation on the for loop." << endl;
boost :: thread** thrds;
//create threads and bind to p_variantforloop_t
thrds = new boost::thread*[numThreads];
variant.start();
for (int i = 1; i <= numThreads; i++)
thrds[i-1] = new boost::thread(boost::bind(&p_variantforloop_t,
C, A, B, ((i)*n - n)/numThreads ,(i * n)/numThreads, numThreads, n));
cout << "before join" <<endl;
// join threads
for (int i = 0; i < numThreads; i++)
thrds[i]->join();
variant.stop();
// cleanup
for (int i = 0; i < numThreads; i++)
delete thrds[i];
delete[] thrds;
cout << "Timer (variation of for loop): " << variant <<endl;
totalVar += variant.user();
//---------------------------------------------------Ends Threaded Variant For Loop------------------------
// output_matrix(A, n);
// output_matrix(B, n);
// output_matrix(E,n);
/*
* free allocated storage
*/
cout << "Deleting Storage" <<endl;
delete_matrix(A, n);
delete_matrix(B, n);
delete_matrix(C, n);
delete_matrix(D, n);
//avoids dangling pointers
A = NULL;
B = NULL;
C = NULL;
D = NULL;
//ends the timesThrough for loop
//output the results to .dat files
output("Nonparallel", numThreads, (totalNonP / averager) , n);
output("Scalar", numThreads, (totalScalar / averager), n);
output("Variant", numThreads, (totalVar / averager), n);
cout << "Nonparallel = " << (totalNonP / averager) << endl;
cout << "Scalar = " << (totalScalar / averager) << endl;
cout << "Variant = " << (totalVar / averager) << endl;
return 0;
void output(string file, int numThreads , long double time, int n)
ofstream dataFile;
stringstream ss;
ss << numThreads;
file += ss.str();
file += ".dat";
dataFile.open(file.c_str(), ios::app);
if(dataFile.fail())
cout << "The output file didn't open." << endl;
exit(1);
//ends the if statement.
dataFile << n << " " << time << endl;
dataFile.close();
//ends optimalOutput function
矩阵文件：
#include <matrix.h>
#include <stdlib.h>
using namespace std;
int** create_matrix(int n)
int** matrix;
if (n < 1)
return 0;
matrix = new int*[n];
for (int i = 0; i < n; i++)
matrix[i] = new int[n];
return matrix;
int** create_random_matrix(int n, random_generator& rg)
int** matrix;
if (n < 1)
return 0;
matrix = new int*[n];
for (int i = 0; i < n; i++)
matrix[i] = new int[n];
for (int j = 0; j < n; j++)
//rg >> matrix[i][j];
matrix[i][j] = rand() % 100;
return matrix;
void delete_matrix(int** matrix, int n)
for (int i = 0; i < n; i++)
delete[] matrix[i];
delete[] matrix;
//avoids dangling pointers.
matrix = NULL;
/*
* non-parallel matrix multiplication
*/
void multiply(int** C, int** A, int** B, int n)
if ((C == A) || (C == B))
cout << "ERROR: C equals A or B!" << endl;
return;
for (int i = 0; i < n; i++)
for (int j = 0; j < n; j++)
C[i][j] = 0;
for (int k = 0; k < n; k++)
C[i][j] += A[i][k] * B[k][j];
void p_scalarproduct_t(int* c, int* a, int* b,
int s, int e, boost::mutex* lock)
int tmp;
tmp = 0;
for (int k = s; k < e; k++)
tmp += a[k] * b[k];
//cout << "a[k]= "<<a[k]<<"b[k]= "<< b[k] <<" "<<k<<endl;
lock->lock();
*c = *c + tmp;
lock->unlock();
int p_scalarproduct(int* a, int* b, int n, int m)
int c;
boost::mutex lock;
boost::thread** thrds;
c = 0;
/* create threads and bind to p_merge_sort_t */
thrds = new boost::thread*[m];
for (int i = 0; i < m; i++)
thrds[i] = new boost::thread(boost::bind(&p_scalarproduct_t,
&c, a, b, i*n/m, (i+1)*n/m, &lock));
/* join threads */
for (int i = 0; i < m; i++)
thrds[i]->join();
/* cleanup */
for (int i = 0; i < m; i++)
delete thrds[i];
delete[] thrds;
return c;
void output_matrix(int** matrix, int n)
cout << "[";
for (int i = 0; i < n; i++)
cout << "[ ";
for (int j = 0; j < n; j++)
cout << matrix[i][j] << " ";
cout << "]" << endl;
cout << "]" << endl;
void p_variantforloop_t(int** C, int** A, int** B, int s, int e, int numThreads, int n)
//cout << "s= " <<s<<endl<< "e= " << e << endl;
for(int i = s; i < e; i++)
for(int j = 0; j < n; j++)
C[i][j] = 0;
//cout << "i " << i << " j " << j << endl;
for (int k = 0; k < n; k++)
C[i][j] += A[i][k] * B[k][j];
//ends the function
【问题讨论】：
如果您只隔离并运行第三种变体，您是否看到所有 CPU 内核同时处于活动状态（并且希望是固定的）？ ...从代码看来，它应该是并行工作的，但这至少可以让您快速检查是否有问题。 @jason 我不确定如何使用 Linux 进行检查，您对如何执行此操作有任何建议吗？我在学校使用计算机，所以我没有 root 权限。 什么发行版？在 Ubuntu 上，我只需转到 System->Administration->System Monitor 查看 CPU 负载... Fedora/RH/SuSE/etc。有同样的东西，虽然菜单可以在不同的地方。此外，如果您只有一个命令行可用，您可以输入top 以获取基于进程的 CPU 监视器（还提供许多其他统计信息）。
是的，这就是我所做的，但它只有一个 CPU 图，这对我来说根本没有意义，因为其他线程实现实际上是并行计算的......是的，我正在运行 Ubuntu跨度>
顺便说一句，top 是交互式的，所以如果你按下“h”键，你会得到一堆有用的命令来告诉你如何操作你在主显示屏上看到的东西。例如，“”将在进程之间滚动，按“u”将提示您为用户过滤进程，按“1”将为您提供 CPU 的 SMP 视图。最后按 'd' 你可以改变更新速度（1.0 是一秒，但它可以小于）。
【参考方案1】：
我猜你遇到了False Sharing。尝试在p_variantforloop_t中使用局部变量：
void p_variantforloop_t(int** C, int** A, int** B, int s, int e, int numThreads, int n)
for(int i = s; i < e; i++)
for(int j = 0; j < n; j++)
int accu = 0;
for (int k = 0; k < n; k++)
accu += A[i][k] * B[k][j];
C[i][j] = accu;
【讨论】：
好的，这对我来说很有意义，但是当我将它添加到我的程序中时，它根本不会反映在我的输出中，即使我添加了那篇文章中的内容 如果你有一个完整的临时矩阵作为你的函数中的累加器，并且只在p_variantforloop_t函数的最后将结果从临时矩阵复制到C而不是复制单元格- for 循环中间的按单元格？
@Jason：我也是这么想的，但是 OP 沿着最大步幅维度（第一个索引，对吗？）划分工作，而写入沿着最小步幅维度进行，所以我不是所以确定它实际上是错误共享，除非范围非常小。
另一种可能的解释是程序受到 CPU 和 RAM 之间带宽的限制......在这种情况下，将任务划分为多个内核并没有真正的帮助，因为现在他只有两个等待 RAM 读/写完成的内核，而不是一个。
那么为什么只有一个实现受到影响，而另一个并行实现不受影响呢？【参考方案2】：
根据您在 cmets 中的回复，理论上，由于您只有一个线程（即 CPU）可用，所有线程版本应该与单线程版本相同或更长，因为线程管理开销.您根本不应该看到任何加速，因为解决矩阵的一部分所花费的时间片是从另一个并行任务中窃取的时间片。使用单个 CPU，您只是分时共享 CPU 的资源——在给定的单个时间片内没有真正的并行工作。我怀疑你的第二个实现运行得更快的原因是因为你在你的内部循环中做更少的指针取消引用和内存访问。例如，在来自multiply 和p_variantforloop_t 的主要操作C[i][j] += A[i][k] * B[k][j]; 中，您正在查看汇编级别的许多操作，其中许多与内存相关。在“汇编伪代码”中它看起来像下面这样：
1) 将堆栈上A引用的地址的指针值移动到寄存器R1中
2) 将寄存器R1 中的地址增加变量i、j 或k 引用的堆栈外的值
3）将指针地址值从R1指向的地址移动到R1中
4) 将R1 中的地址增加变量i、j 或k 引用的堆栈外的值
5) 将R1指向的地址中的值移动到R1中（所以R1现在持有A[i][k]的值）
6) 对堆栈上B 引用的地址执行步骤 1-5 到寄存器 R2 中（所以 R2 现在保存 B[k][j] 的值）
7) 对堆栈上C 引用的地址执行步骤1-4 到寄存器R3
8) 将R3指向的地址中的值移动到R4中（即R4将实际值保存在C[i][j]）
9) 寄存器R1和R2相乘，存入寄存器R5
10) 添加寄存器R4和R5并存储在R4
11）将最终值从R4移回R3指向的内存地址（现在C[i][j]有最终结果）
这是假设我们有 5 个通用寄存器可供使用，并且编译器正确优化了您的 C 代码以利用它们。我将循环索引变量i、j 和k 留在了堆栈中，因此访问它们将比在寄存器中花费更多的时间……这实际上取决于您的编译器必须有多少个寄存器在您的平台上玩。此外，如果您在没有任何优化的情况下进行编译，您可能会在堆栈外进行更多的内存访问，其中一些临时值存储在堆栈中而不是寄存器中，然后从堆栈中重新访问，这需要更长的时间而不是在寄存器之间移动值。无论哪种方式，上面的代码都很难优化。它可以工作，但如果您使用的是 32 位 x86 平台，那么您将不会有那么多通用寄存器可供使用（但您至少应该有 6 个）。 x86_64 有更多寄存器可供使用，但仍然有所有内存访问需要处理。
另一方面，像 tmp += a[k] * b[k] 和 p_scalarproduct_t 这样的操作在紧密的内部循环中会移动得更快......这是汇编伪代码中的上述操作：
循环会有一个小的初始化步骤
1) 将tmp 设为寄存器R1 而不是堆栈变量，并将其值初始化为0
2）将栈上a引用的地址值移动到R2
3) 将堆栈中的s 的值添加到R2 并将结果地址保存在R2
4）将堆栈上b引用的地址值移动到R3
5) 将堆栈中的s 的值添加到R3 并将结果地址保存在R3
6）在R6中设置一个计数器，初始化为e - s
在一次性初始化之后，我们将开始真正的内循环
7) 将R2指向的地址中的值移动到R4中
8) 将R3指向的地址中的值移动到R5中
9) 将R4 和R5 相乘并将结果存储在R5
10) 将R5 添加到R1 并将结果存储在R1
11）增加R2和R3
12) 递减R6 中的计数器，直到它达到零，我们终止循环
我不能保证这正是您的编译器设置此循环的方式，但您通常可以看到，在您的标量示例中，内部循环所需的步骤更少，更重要的是内存访问更少。因此，仅使用寄存器的操作可以完成更多操作，而不是包含内存位置并需要内存提取的操作，这比仅使用寄存器的操作慢得多。所以总的来说它会移动得更快，这与线程无关。
最后，我注意到标量积只有两个嵌套循环，所以它的复杂度是 O(N^2)，而对于其他两种方法，你有三个嵌套循环，复杂度为 O(N^3) .这也会产生影响。
【讨论】：
以上是关于关于线程的问题的主要内容，如果未能解决你的问题，请参考以下文章