StanFord CS149 Assignment2 简记
2024-02-04 14:20:10 # Labs #C++ #Lab #Parallel #CUDA

Github
失去了优化性能的梦想

Part_A

任务 1: TaskSystemParallelSpawn

首先不能对于每个任务创建一个线程, 那样不仅不符合线程数要求, 还效率低下

第一个任务不要求实现线程池, 所以类中只需要记录一个 num_threads

1
2
3
4
5
6
7
8
9
10
11
12
// tasksys.h
class TaskSystemParallelSpawn: public ITaskSystem {
    public:
        int num_threads; // new
        TaskSystemParallelSpawn(int num_threads);
        ~TaskSystemParallelSpawn();
        const char* name();
        void run(IRunnable* runnable, int num_total_tasks);
        TaskID runAsyncWithDeps(IRunnable* runnable, int num_total_tasks,
                                const std::vector<TaskID>& deps);
        void sync();
};   
1
2
3
4
5
// tasksys.cpp
TaskSystemParallelSpawn::TaskSystemParallelSpawn(int num_threads) : ITaskSystem(num_threads)
{
    this->num_threads = num_threads;
}

分配任务: 可以用一个共享变量表示当前已经处理完的任务数量, 用 mutex 锁住即可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
void TaskSystemParallelSpawn::run(IRunnable *runnable, int num_total_tasks)
{
    std::shared_ptr<std::mutex> mu = std::make_shared<std::mutex>();
    int count = 0;

    auto func = [&count, mu, num_total_tasks, runnable]()
    {
        while (true)
        {
            // 取出共享变量
            mu->lock();
            int tmp = count++;
            mu->unlock();

            if (tmp >= num_total_tasks) break;
            runnable->runTask(tmp, num_total_tasks);
        }
    };

    std::vector<std::thread> threadId(num_threads);
    // 子线程
    for (int i = 1; i < num_threads; ++i)
        threadId[i] = std::thread(func);

    // 主线程
    func();

    // 等待子线程
    for (int i = 1; i < num_threads; ++i)
        threadId[i].join();
    
}

任务 2

任务的状态表示

正常线程池使用任务队列, 这里可以直接记录 runnable

任务有三种状态: 已完成, 完成中, 未完成

  1. 已完成: 0 ~ num_finished
  2. 完成中: num_finished ~ num_toBeDone - 1
  3. 未完成: num_toBeDone ~ num_total_tasks

这里使用了两个锁分别锁住 num_finished 和 num_toBeDone

目的是为了提高性能, 只用一把锁可以保证正确, 但是性能不够

num_total_tasks 不用专门锁的原因是他只会在主线程被写入一次

还需要一个 killed 表示任务是否已经全部执行完毕

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class TaskSystemParallelThreadPoolSpinning: public ITaskSystem {
    private:
        int num_threads;
        std::vector<threadPtr> threads;

        IRunnable* runnable;

        int num_total_tasks;
        int num_finished;
        int num_toBeDone;
        
        bool killed;

        std::shared_ptr<std::mutex> mu;
        std::shared_ptr<std::mutex> mu_finished;
        std::shared_ptr<std::condition_variable> se;
};

线程工作流程

1
2
3
4
5
6
7
8
9
10
// 主线程
// 构造函数中
初始化
// run 函数中
持有两把锁, 修改各种变量

只持有锁2, 判断任务是否结束, 不结束就 wait
// 析构函数中
killed 设置为 true
关闭所有线程
  1. 子线程忙等的时候不要修改共享变量
  2. (锁1中)确定有任务可做的时候才修改 num_toBeDone
  3. (锁外)判断有无任务被取出, 取出就做
  4. (锁2中)判断是否完成了所有任务, 完成了就通知主线程
1
2
3
4
5
6
7
8
// 子线程
while (!killed)
{
    获取当前任务
    执行任务
    更新任务状态
    任务结束唤醒主线程
}

任务 3

与任务 2 类似, 不过优化不足, 不能通过测试

新增部分

子线程循环开始之后要判断当前是否有任务, 否则等待

主线程 run 函数初始值设定完毕之后要通知所有子线程

析构函数 killed 设置完之后要通知一下子线程从而关闭

Part_b

  1. 任务的表示: 包括上述三个变量, 新增一个 id
  2. 依赖关系的表示: 一个邻接表 G, 一个入度 in_degree
  3. 任务队列: 存储 TaskID, 另用一个 vector 存储 TaskInfo
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
private:
    struct TaskInfo
    {
        TaskID id;
        IRunnable *runnable;
        int num_toBeDone;
        int num_finished;
        int num_total_task;
        TaskInfo(TaskID id, IRunnable *runnable, int a, int b, int c)
        {
            this->id = id;
            this->runnable = runnable;
            num_toBeDone = a;
            num_finished = b;
            num_total_task = c;
        }
    };

    bool killed;

    std::mutex mu;
    std::condition_variable S, T;

    int num_all_undone_task;

    std::vector<std::thread> threads;
    std::vector<std::vector<int>> G;
    std::vector<int> in_degree;
    std::queue<TaskID> tasks;
    std::vector<TaskInfo> tasks_info;
    static constexpr int N = 1e5 + 10;

子线程函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
while (true)
{
    等待唤醒

    找到当前任务队列的队头, 取出 task
    接收子任务, 并判断子任务是否全部进入 
        正在执行和执行完毕这两个状态, 是就出队

    runTask

    标记子任务已完成, 判断 task 是否全部完成

    全部完成之后沿着图减少一下入度, 如果发现某个后继 task 入度为 0
        那么入队, 通知一下其他线程可以开始了

    如果未完成的 task 数量为 0, 通知主线程可以结束任务
}

析构函数

1
2
3
4
设置 `killed` 标志位
sync() 等待任务全部完成
通知子线程退出
等待所有子线程退出

run 和 runAsyncWithDeps 函数

1
2
3
4
5
void TaskSystemParallelThreadPoolSleeping::run(IRunnable *runnable, int num_total_tasks)
{
    runAsyncWithDeps(runnable, num_total_tasks, {});
    sync();
}
1
2
3
4
5
6
7
TaskID TaskSystemParallelThreadPoolSleeping::runAsyncWithDeps(IRunnable *runnable, int num_total_tasks,
                                                              const std::vector<TaskID> &deps)
{
    设置 TaskID
    加入图中, 更新入度
    如果入度为 0, 那么进入 task 队列, 通知子线程
}

sync 函数

等待任务全部完成即可

Prev
2024-02-04 14:20:10 # Labs #C++ #Lab #Parallel #CUDA
Next