【发布时间】：2016-12-16 15:58:37
【问题描述】：
我正在用 C 语言实现一个实时信号处理算法，并且我正在尝试使用多线程并行化一段代码。
单线程实现的代码是
void calcTheta(float *theta, float **s, float ***q, float ***g,
int *Ki, int m, int numObv, int numTask) {
int i, j, k;
for (i = 0; i < m; i++) {
theta[i] = 0;
for (j = 0; j < numObv; j++) {
for (k = 0; k < numTask; k++) {
theta[i] += (Ki[k] * (pow(fabs(q[i][j][k]), 2) / g[i][j][k]) - s[i][k]) /
(s[i][k] * (s[i][k] - (pow(fabs(q[i][j][k]), 2) / g[i][j][k])));
}//k
}//j
theta[i] = (numTask * numObv) / theta[i];
}//i
}
多线程实现使用线程假脱机的想法，我创建了几个线程并不断向它们发出信号以使用特定的数据数组进行处理。代码如下：
#define NUM_THREADS_THETA 2
#define TRUE 1
#define FALSE 0
#define READY 1
#define DONE 0
struct threadThetaData {
float *theta;
float **s;
float ***q;
float ***g;
int *Ki;
int numObv;
int numTask;
int threadId;
};
struct threadThetaData dataArrayTheta[NUM_THREADS_THETA];
int termThread[NUM_THREADS_THETA];
int statusThread[NUM_THREADS_THETA];
int iVal[NUM_THREADS_THETA];
pthread_mutex_t mutexThreadProc[NUM_THREADS_THETA];
pthread_mutex_t mutexMainProc[NUM_THREADS_THETA];
pthread_cond_t condThreadProc[NUM_THREADS_THETA];
pthread_cond_t condMainProc[NUM_THREADS_THETA];
void *doProcTheta(void *threadArg) {
struct threadThetaData *myData = (struct threadThetaData *)threadArg;
float *theta = myData->theta;
float **s = myData->s;
float ***q = myData->q;
float ***g = myData->g;
int *Ki = myData->Ki;
int numObv = myData->numObv;
int numTask = myData->numTask;
int threadId = myData->threadId;
int j, k;
while(1) {
//printf("thread %d waiting for signal from master..\n", threadId);
pthread_mutex_lock(&mutexThreadProc[threadId]);
while (statusThread[threadId] != READY)
pthread_cond_wait(&condThreadProc[threadId], &mutexThreadProc[threadId]);
pthread_mutex_unlock(&mutexThreadProc[threadId]);
//printf("thread %d got signal from master..\n", threadId);
if (termThread[threadId] == TRUE)
pthread_exit(NULL);
theta[iVal[threadId]] = 0;
for (j = 0; j < numObv; j++) {
for (k = 0; k < numTask; k++) {
theta[iVal[threadId]] += (Ki[k]*(pow(fabs(q[iVal[threadId]][j][k]),2)/g[iVal[threadId]][j][k]) - s[iVal[threadId]][k])/(s[iVal[threadId]][k]*(s[iVal[threadId]][k] - (pow(fabs(q[iVal[threadId]][j][k]),2)/g[iVal[threadId]][j][k])));
}//k
}//j
theta[iVal[threadId]] = (numTask*numObv)/theta[iVal[threadId]];
pthread_mutex_lock(&mutexMainProc[threadId]);
statusThread[threadId] = DONE;
pthread_cond_signal(&condMainProc[threadId]);
pthread_mutex_unlock(&mutexMainProc[threadId]);
//printf("thread %d signaled to master..\n", threadId);
}
}
void calcTheta(float *theta,float **s,float ***q,float ***g,int *Ki,int m, int numObv, int numTask)
{
int i,j;
pthread_t thetaThreads[NUM_THREADS_THETA];
int numThreadBlks = m/NUM_THREADS_THETA;
int numThreadRem = m%NUM_THREADS_THETA;
int mCount = 0;
for(i=0;i<NUM_THREADS_THETA;i++)
{
pthread_mutex_init(&mutexThreadProc[i], NULL);
pthread_mutex_init(&mutexMainProc[i], NULL);
pthread_cond_init (&condThreadProc[i], NULL);
pthread_cond_init (&condMainProc[i], NULL);
dataArrayTheta[i].theta = theta;
dataArrayTheta[i].s = s;
dataArrayTheta[i].q = q;
dataArrayTheta[i].g = g;
dataArrayTheta[i].Ki = Ki;
dataArrayTheta[i].numObv = numObv;
dataArrayTheta[i].numTask = numTask;
dataArrayTheta[i].threadId = i;
termThread[i] = FALSE;
statusThread[i] = DONE;
pthread_create(&thetaThreads[i],NULL,doProcTheta,(void *)&dataArrayTheta[i]);
}
for(i=0;i<numThreadBlks;i++)
{
for(j=0;j<NUM_THREADS_THETA;j++)
{
pthread_mutex_lock(&mutexThreadProc[j]);
statusThread[j] = READY;
iVal[j] = mCount;
mCount++;
pthread_cond_signal(&condThreadProc[j]);
pthread_mutex_unlock(&mutexThreadProc[j]);
//printf("Signaled thread %d from master ... Waiting on signal ..\n",j);
}
for(j=0;j<NUM_THREADS_THETA;j++)
{
pthread_mutex_lock(&mutexMainProc[j]);
while (statusThread[j] != DONE)
pthread_cond_wait(&condMainProc[j], &mutexMainProc[j]);
pthread_mutex_unlock(&mutexMainProc[j]);
//printf("Got signal from thread %d to master \n",j);
}
}
for(j=0;j<numThreadRem;j++)
{
pthread_mutex_lock(&mutexThreadProc[j]);
statusThread[j] = READY;
iVal[j] = mCount;
mCount++;
pthread_cond_signal(&condThreadProc[j]);
pthread_mutex_unlock(&mutexThreadProc[j]);
}
for(j=0;j<numThreadRem;j++)
{
pthread_mutex_lock(&mutexMainProc[j]);
while (statusThread[j] != DONE)
pthread_cond_wait(&condMainProc[j], &mutexMainProc[j]);
pthread_mutex_unlock(&mutexMainProc[j]);
}
for(j=0;j<NUM_THREADS_THETA;j++)
{
pthread_mutex_lock(&mutexThreadProc[j]);
statusThread[j] = READY;
termThread[j] = TRUE;
pthread_cond_signal(&condThreadProc[j]);
pthread_mutex_unlock(&mutexThreadProc[j]);
pthread_join(thetaThreads[j],NULL);
pthread_mutex_destroy(&mutexThreadProc[j]);
pthread_cond_destroy(&condThreadProc[j]);
pthread_mutex_destroy(&mutexMainProc[j]);
pthread_cond_destroy(&condMainProc[j]);
}
}
数组尺寸：
float theta[m];
float s[m][numTask];
float q[m][numObv][numTask];
float g[m][numObv][numTask];
int Ki[numTask];
对于特定的数据集，
m=661
numObv=96
numTask=1024
运行时是：
Single threaded : 4.5 seconds
Multithreaded with 2 threads : 6.9 seconds
我希望多线程代码的运行时能够给我一些性能改进，而不是单线程代码。任何指向我在这里缺少的东西的指针将不胜感激。
【问题讨论】：
-
锁之间的工作不足导致处理锁所花费的时间比并行工作节省的时间更多。
-
@anshu：可能会，也可能不会……只有基准测试会告诉你。但是请注意，您在锁内做的越多，您获得并行性的机会就越少。
-
您正在将特定作业分配给特定线程。这不聪明。您希望任何一个线程恰好运行来完成任何需要完成的工作。使用工作队列。
-
看一下顺序代码，由于你是在 theta[i] 中总结的，你似乎不需要任何锁。只需启动尽可能多的线程来处理每个 theta[i]。当达到 m 时，线程退出。
-
这与“实时”完全无关。请不要随便使用随机词，因为它们可能“看起来很酷”。
标签： c multithreading performance