CppConcurrencyInAction/threadsafe_queue.cpp at main · yangyueren/CppConcurrencyInAction · GitHub

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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
//
// Created by yryang on 2021/10/19.
//

#include <iostream>
#include <vector>
#include <thread>
#include <chrono>

#include "threadsafe_queue.h"
#include "util.h"

namespace threadsafe_queue{

    template <class T>
    void push_test(T &queue, std::vector<int> v){
        for (auto &&vv: v){
            queue.push(vv);
        }


        v.clear();
        while (!queue.empty()){
            try {
                int t;
                bool flag = queue.try_pop(t);
                if (flag){
                    v.push_back(t);
                }

            } catch (...) {
                std::cout << "wrong empty\n";   // queue.empty() queue.pop() 有并发冲突
            }

        }
        for (auto &&vv: v){
            queue.push(vv);
        }
    }

    template <class T>
    void test(){
        T queue;
        int N = 1000;
        int thread_num = 10;
        std::vector<std::thread> threads;
        for (int i = 0; i < thread_num; ++i) {
            std::vector<int> v;
            for (int j = 0; j < N/thread_num; ++j) {
                v.push_back(j + (N/thread_num)*i);
            }
            threads.push_back(std::thread(push_test<T>, std::ref(queue), v));
        }
        for (int i = 0; i < thread_num; ++i) {
            threads[i].join();
        }
        std::vector<int> res;
        while (!queue.empty()){
            int t;
            queue.try_pop(t);
            res.push_back(t);
        }
        std::sort(res.begin(), res.end());
        for (int i = 0; i < N; ++i) {
            ASSERT(res[i] == i, "thread safe queue: wrong " );
        }
        std::cout << "test success\n";

    }

    template <class T>
    void test_wait_and_pop(){
        T queue;
        T queue_ans;
        int N = 100;
        int thread_num = 10;

        auto consumer = [](T &queue, T &queue_ans){
            while (true){
                int t;
                queue.wait_and_pop(t);
                std::cout << "thread_id :" << std::this_thread::get_id() <<" consume data " << t << std::endl;
                queue_ans.push(t);
            }

        };

        std::cout << "creating threads..." << std::endl;
        std::vector<std::thread> threads;
        threads.reserve(thread_num);
        for (int i = 0; i < thread_num; ++i) {
            threads.push_back(std::thread(consumer, std::ref(queue), std::ref(queue_ans)));
        }


        std::cout << "produce data...\n";
        for (int i = 0; i < N; ++i) {
            std::cout << "produce data " << i << std::endl;
            queue.push(i);
            std::this_thread::sleep_for(std::chrono::milliseconds (20));
        }
        std::cout << "producer done\n";

        std::vector<int> res;
        while (!queue_ans.empty()){
            int t;
            queue_ans.try_pop(t);
            std::cout << t << std::endl;
            res.push_back(t);
        }
        std::sort(res.begin(), res.end());
        for (int i = 0; i < N; ++i) {
            ASSERT(res[i] == i, "thread safe queue: wrong " );
        }
        std::cout << "test_wait_and_pop success\n";


    }
}

int main(){
//    threadsafe_queue::test<threadsafe_queue::threadsafe_queue<int> >();
    threadsafe_queue::test_wait_and_pop<threadsafe_queue::queue<int> >();
    return 0;
}