并行编程实验一
实验内容
实验内容一：以数据划分的方式并行计算PI值
实验内容二：CPU多核编程——线程池开发
要求基于生产者—消费者模式进行框架开发，具体工作需求可以简化，但需要有线程管理和同步。
计算Pi
思路简述
依据莱布尼兹公式，通过多线程计算较多的次数，逼近$\pi$ 。用多线程的方式进行数据划分、即每个线程分担处理部分数据，从而进行加速。
同时由于多线程访问全局的结果可能会有冲突，因此使用互斥量和信号量组织线程有序将局部结果加到全局结果中。
代码实现
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <math.h>
const int BLOCK_SIZE = 100000;
double sum = 0;
int num_threads;
// 定义互斥量和条件变量
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
void *calculate_block(void *thread_id)
{
long id = (long)thread_id;
int start = id * BLOCK_SIZE;
int end = start + BLOCK_SIZE;
double block_sum = 0;
for (int i = start; i < end; i++) {
double term = pow(-1, i) / (2 * i + 1);
block_sum += term;
}
pthread_mutex_lock(&mutex);
sum += block_sum;
pthread_mutex_unlock(&mutex);
pthread_cond_signal(&cond);
}
int main()
{
num_threads = 8;
pthread_t threads[num_threads];
for (long i = 0; i < num_threads; i++)
pthread_create(&threads[i], NULL, calculate_block, (void *)i);
for(int i = 0; i < num_threads; i ++)
pthread_join(threads[i], NULL);
printf("%lf", sum * 4);
}线程池设计
线程池实现
使用一个任务队列作为生产者和消费者之间的缓冲区，任务队列中的每一个元素包含要执行的函数和函数参数，对应代码如下:
typedef struct task_t {
void (*func)(void *);
void *arg;
} task_t;线程池包含一个任务队列、若干线程、互斥量与信号量以及其它关键属性，定义如下：
typedef struct thread_pool_t {
pthread_t threads[MAX_THREADS];
int num_threads;
task_t queue[MAX_QUEUE];
int front, rear, size;
pthread_mutex_t mutex;
pthread_cond_t cond;
bool shutdown;
} thread_pool_t;在生产者方面，thread_pool_enqueue函数用于将任务添加到任务队列中。当生产者生产了一个任务后，它首先会通过线程池的互斥锁pool->mutex来保护任务队列，防止多个线程同时修改任务队列，然后使用条件变量pool->cond来通知消费者有新的任务到达。
bool thread_pool_enqueue(thread_pool_t *pool, void (*func)(void *), void *arg){
pthread_mutex_lock(&pool->mutex);
if (pool->size == MAX_QUEUE) {
pthread_mutex_unlock(&pool->mutex);
return false;
}
task_t task = { .func = func, .arg = arg };
pool->queue[pool->rear] = task;
pool->rear = (pool->rear + 1) % MAX_QUEUE;
pool->size++;
pthread_cond_signal(&pool->cond);
pthread_mutex_unlock(&pool->mutex);
return true;
}在消费者方面，thread_pool_worker函数用于从任务队列中取出任务并执行。当消费者从任务队列中取出一个任务时，它会使用互斥锁pool->mutex来保护任务队列。如果任务队列为空，消费者将会被条件变量pool->mutex阻塞，等待生产者添加新的任务到任务队列中。
void *thread_pool_worker(void *arg){
thread_pool_t *pool = (thread_pool_t *)arg;
while (true) {
pthread_mutex_lock(&pool->mutex);
while (pool->size == 0 && !pool->shutdown)
pthread_cond_wait(&pool->cond, &pool->mutex);
if (pool->size == 0 && pool->shutdown) {
pthread_mutex_unlock(&pool->mutex);
pthread_exit(NULL);
}
task_t task = pool->queue[pool->front];
pool->front = (pool->front + 1) % MAX_QUEUE;
pool->size--;
pthread_mutex_unlock(&pool->mutex);
task.func(task.arg);
}
}线程池的启动包括初始化任务队列、信号量、启动消费者线程；线程池的关闭最重要的是等待所有线程运行结束；两者实现如下：
void thread_pool_init(thread_pool_t *pool, int num_threads) {
pool->num_threads = num_threads;
pool->front = pool->rear = pool->size = 0;
pool->shutdown = false;
pthread_mutex_init(&pool->mutex, NULL);
pthread_cond_init(&pool->cond, NULL);
for (int i = 0; i < num_threads; i++)
pthread_create(&pool->threads[i], NULL, thread_pool_worker, pool);
}
void thread_pool_shutdown(thread_pool_t *pool) {
pthread_mutex_lock(&pool->mutex);
pool->shutdown = true;
pthread_cond_broadcast(&pool->cond);
pthread_mutex_unlock(&pool->mutex);
for (int i = 0; i < pool->num_threads; i++)
pthread_join(pool->threads[i], NULL);
pthread_mutex_destroy(&pool->mutex);
pthread_cond_destroy(&pool->cond);
}设计一个简单的任务，输出任务id与线程id，并放到线程池运行：
void my_task(void *arg){
int *num = (int *)arg;
printf("Task %d by thread %lu\n", *num, pthread_self());
free(num);
}
int main() {
thread_pool_t pool;
thread_pool_init(&pool, 8);
for (int i = 0; i < 20; i++) {
int *num = malloc(sizeof(int));
*num = i;
thread_pool_enqueue(&pool, my_task, num);
}
printf("main thread %lu\n", pthread_self());
thread_pool_shutdown(&pool);
return 0;
}运行结果
在ubuntu 20中编译gcc tp.c -o tp -lpthread，运行结果描述如下：
Task 0 by thread 140422668744478
Task 4 by thread 140422668744456
Task 5 by thread 140422668744434
...线程池
使用线程池完成矩阵相加$a+b=c$，相应的任务函数如下：
void add_matrix(void *arg) {
int *num = (int *)arg;
int st = *num, ed = st + 5;
for (int i = st; i < ed; i++)
c[i] = a[i] + b[i];
printf("Task [add_matrix] %d by thread %lu, range = %d ~ %d \n", *num /5, pthread_self(), st, ed);
free(num);
}
