我正在用 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
我希望多线程代码的运行时能够比单线程代码带来一些性能改进,而反之亦然。如果有人指出我在这里缺少的内容,我将不胜感激。
最佳答案
对于当前的问题,您的多线程实现似乎过于复杂。单线程代码显示每个 theta
元素的计算独立于所有其他 theta
元素。
因此,您不需要互斥体和条件,因为线程之间不需要数据交换/同步。只需让线程处理 theta
的不同范围即可。计算。
与 m=661
和 2 个线程,那么第一个线程应该计算 theta
在 0..330 范围内,第二个线程应该计算 theta
范围 331..660。启动两个线程并等待它们完成(也称为联接)。
您几乎可以使用单线程代码来实现多线程。您所需要的只是向函数添加一个起始索引。
关于c - 多线程导致 C 语言性能下降,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/38870492/