Linux：多线程（二.理解pthread_t、线程互斥与同步、基于阻塞队列的生产消费模型）

#include <iostream>
#include <thread> // C++里的库
#include <unistd.h>
#include <sys/types.h>

using namespace std;

void *task(void *argv)
{
    int count = 5;
    while (true)
    {
        cout << "I'm a new thread ,pid : " << getpid() << ". count:" << count << "count's address:" << &count << endl;
        sleep(1);
        count--;
    }
    return nullptr;
}

int main()
{
    pthread_t tid1;
    pthread_t tid2;
    pthread_create(&tid1, nullptr, task, nullptr);
    pthread_create(&tid2, nullptr, task, nullptr); // 这里我们两个线程执行一个函数，里面有临时变量，看二者地址如何

    pthread_join(tid1, nullptr);
    pthread_join(tid2, nullptr);

    return 0;
}

#ifndef __THREAD_HPP__
#define __THREAD_HPP__

#include <iostream>
#include <string>
#include <unistd.h>
#include <functional>
#include <pthread.h>

namespace ThreadModule
{
    template <typename T>
    using func_t = std::function<void(T &)>;
    // typedef std::function<void(const T&)> func_t;

    template <typename T>
    class Thread
    {
    public:
        void Excute()
        {
            _func(_data);
        }

    public:
        Thread(func_t<T> func, T &data, const std::string &name = "none-name")
            : _func(func), _data(data), _threadname(name), _stop(true)
        {
        }
        static void *threadroutine(void *args) // 类成员函数，形参是有this指针的！！
        {
            Thread<T> *self = static_cast<Thread<T> *>(args);
            self->Excute();
            return nullptr;
        }
        bool Start()
        {
            int n = pthread_create(&_tid, nullptr, threadroutine, this);
            if (!n)
            {
                _stop = false;
                return true;
            }
            else
            {
                return false;
            }
        }
        void Detach()
        {
            if (!_stop)
            {
                pthread_detach(_tid);
            }
        }
        void Join()
        {
            if (!_stop)
            {
                pthread_join(_tid, nullptr);
            }
        }
        std::string name()
        {
            return _threadname;
        }
        void Stop()
        {
            _stop = true;
        }
        ~Thread() {}

    private:
        pthread_t _tid;
        std::string _threadname;
        T &_data; // 为了让所有的线程访问同一个全局变量
        func_t<T> _func;
        bool _stop;
    };
} // namespace ThreadModule

#endif

#include "Thread.hpp"
using namespace MyThread;

class ThreadData
{
public:
    ThreadData(int &tickets, const std::string &name)
        : _tickets(tickets), _name(name), _total(0)
    {
    }
    ~ThreadData()
    {
    }

public:
    int &_tickets;     // 所有的线程，最后都会引用同一个全局的g_tickets
    std::string _name; // 进程的名字
    int _total;        // 这个进程抢了多少票
};

int g_tickets = 10000; // 共享资源，没有保护的, 临界资源
const int num = 4;     // 线程数量

void route(ThreadData *td)
{
    while (true)
    {
        if (td->_tickets > 0)
        {
            usleep(1000);
            printf("%s running, get tickets: %d\n", td->_name.c_str(), td->_tickets);
            td->_total++;
            td->_tickets--;
        }
        else
        {
            break;
        }
    }
}

int main()
{
    std::vector<Thread<ThreadData *>> threads; // 所有的线程存在一个数组里
    std::vector<ThreadData *> datas;           // 所有的数据也是
    // 1. 创建一批线程
    for (int i = 0; i < num; i++)
    {
        std::string name = "thread-00" + std::to_string(i + 1);
        ThreadData *td = new ThreadData(g_tickets, name);
        threads.emplace_back(route, td, name);
        datas.emplace_back(td); // 创建完后，都插入
    }

    for (auto &e : threads)
    {
        e.Start();
    }

    for (auto &e : threads)
    {
        e.Join();
    }

    return 0;
}

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER

int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr);

int pthread_mutex_destroy(pthread_mutex_t *mutex);

int pthread_mutex_lock(pthread_mutex_t *mutex);

int pthread_mutex_unlock(pthread_mutex_t *mutex);

int g_tickets = 1000; // 共享资源，没有保护的,  临界资源
const int num = 4;    // 线程数量

pthread_mutex_t gmutex = PTHREAD_MUTEX_INITIALIZER;

void route(ThreadData *td)
{
    while (true)
    {
        pthread_mutex_lock(&gmutex); // 要在这里进行加锁，让一次只有一个线程（竞争力强的那个）能进里面
        if (td->_tickets > 0)        // 每个线程内部，访问临界资源的代码，就叫做临界区
        {
            usleep(1000);
            printf("%s running, get tickets: %d\n", td->_name.c_str(), td->_tickets);
            td->_tickets--;
            pthread_mutex_unlock(&gmutex);
            td->_total++;
        }
        else
        {
            pthread_mutex_unlock(&gmutex);
            break;
        }
    }
}

#include "Thread.hpp"
using namespace MyThread;

class ThreadData
{
public:
    ThreadData(int &tickets, const std::string &name, pthread_mutex_t &mutex)
        : _tickets(tickets), _name(name), _total(0), _mutex(mutex)
    {
    }
    ~ThreadData()
    {
    }

public:
    int &_tickets;           // 所有的线程，最后都会引用同一个全局的g_tickets
    std::string _name;       // 进程的名字
    int _total;              // 这个进程抢了多少票
    pthread_mutex_t &_mutex; // 传一个动态锁过来，因为是引用，所以都是同一个锁
};

int g_tickets = 1000; // 共享资源，没有保护的,  临界资源
const int num = 4;    // 线程数量

// pthread_mutex_t gmutex = PTHREAD_MUTEX_INITIALIZER;

void route(ThreadData *td)
{
    while (true)
    {
        // pthread_mutex_lock(&gmutex); // 要在这里进行加锁，让一次只有一个线程（竞争力强的那个）能进里面

        pthread_mutex_lock(&td->_mutex);
        if (td->_tickets > 0) // 每个线程内部，访问临界资源的代码，就叫做临界区
        {
            usleep(1000);
            printf("%s running, get tickets: %d\n", td->_name.c_str(), td->_tickets);
            td->_tickets--;
            pthread_mutex_unlock(&td->_mutex);
            td->_total++;
        }
        else
        {
            pthread_mutex_unlock(&td->_mutex);
            break;
        }
    }
}

int main()
{
    pthread_mutex_t mutex;
    pthread_mutex_init(&mutex, nullptr); // 进行初始化

    std::vector<Thread<ThreadData *>> threads; // 所有的线程存在一个数组里
    std::vector<ThreadData *> datas;           // 所有的数据也是
    // 1. 创建一批线程
    for (int i = 0; i < num; i++)
    {
        std::string name = "thread-00" + std::to_string(i + 1);
        ThreadData *td = new ThreadData(g_tickets, name, mutex);
        threads.emplace_back(route, td, name);
        datas.emplace_back(td); // 创建完后，都插入
    }
    for (auto &e : threads)
    {
        e.Start();
    }
    for (auto &e : threads)
    {
        e.Join();
    }
    pthread_mutex_destroy(&mutex);
    return 0;
}

lock:
    movb $0, al        ; 将值0加载到al寄存器中
    xchgb al, mutex    ; 将al寄存器的值和mutex的值进行交换
    cmpb $0, al        ; 比较al寄存器的值和0
    jne wait           ; 如果al寄存器的值不等于0，则跳转到等待（wait）标签
    ret                ; 返回，表示加锁成功,会去执行下面的代码

wait:
    suspend            ; 挂起线程等待
    jmp lock           ; 跳转到lock标签，重新尝试加锁

unlock:
    movb $1, mutex     ; 将值1写入mutex，表示解锁
    wakeup             ; 唤醒等待mutex的线程
    ret                ; 返回，表示解锁成功

pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr);

int pthread_cond_destroy(pthread_cond_t *cond);

int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);

int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);

#include <iostream>
#include <string>
#include <cstring>
#include <vector>
#include <pthread.h>
#include <unistd.h>

void *Master(void *args)
{
    std::string name = static_cast<char *>(args);
    while (true)
    {
        std::cout << name << std::endl;
        sleep(1);
    }
}

void StartMaster(std::vector<pthread_t> *tids)
{
    pthread_t tid;
    int n = pthread_create(&tid, nullptr, Master, (void *)"Master Thread");
    if (n == 0)
    {
        std::cout << "create master success" << std::endl;
    }
    tids->emplace_back(tid);
}

void *Slaver(void *args)
{
    std::string name = static_cast<char *>(args);
    while (true)
    {
        std::cout << name << std::endl;
        sleep(1);
    }
}

void StartSlaver(std::vector<pthread_t> *tids, int num)
{
    for (int i = 0; i < num; i++)
    {
        pthread_t tid;
        char *name = new char[20];
        snprintf(name, 20, "salver-00%d", i + 1);
        int n = pthread_create(&tid, nullptr, Slaver, name);
        if (n == 0)
        {
            std::cout << "create success: " << name << std::endl;
            tids->emplace_back(tid);
        }
    }
}

void WaitThread(std::vector<pthread_t> tids)
{
    for (auto &tid : tids)
    {
        pthread_join(tid, nullptr);
    }
}

int main()
{
    std::vector<pthread_t> tids; // 这里放所有的线程的tid
    StartMaster(&tids);          // 启动主线程
    StartSlaver(&tids, 5);       // 启动新线程
    WaitThread(tids);            // 等待新线程
    return 0;
}

#include <iostream>
#include <string>
#include <cstring>
#include <vector>
#include <pthread.h>
#include <unistd.h>

pthread_mutex_t gmutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t gcond = PTHREAD_COND_INITIALIZER; // 创建一个锁和条件变量

void *Master(void *args) // 我们选择在主线程里面进行条件的唤醒
{
    std::string name = static_cast<char *>(args);
    while (true)
    {
        // std::cout << name << std::endl;
        sleep(1);
        pthread_cond_signal(&gcond); // 唤醒其中一个队列首部的线程
        // pthread_cond_broadcast(&gcond); // 唤醒队列中所有的线程
        std::cout << "master 唤醒一个线程..." << std::endl;
    }
}

void StartMaster(std::vector<pthread_t> *tids)
{
    pthread_t tid;
    int n = pthread_create(&tid, nullptr, Master, (void *)"Master Thread");
    if (n == 0)
    {
        std::cout << "create master success" << std::endl;
    }
    tids->emplace_back(tid);
}

void *Slaver(void *args)
{
    std::string name = static_cast<char *>(args);
    while (true)
    {
        // 1. 加锁
        pthread_mutex_lock(&gmutex);
        // 2. 一般条件变量是在加锁和解锁之间使用的
        pthread_cond_wait(&gcond, &gmutex); // gmutex:这个是，是用来被释放互斥锁的
        std::cout << name << std::endl;
        sleep(1);
        pthread_mutex_unlock(&gmutex);
        // 3.解锁
    }
}

void StartSlaver(std::vector<pthread_t> *tids, int num)
{
    for (int i = 0; i < num; i++)
    {
        pthread_t tid;
        char *name = new char[20];
        snprintf(name, 20, "salver-00%d", i + 1);
        int n = pthread_create(&tid, nullptr, Slaver, name);
        if (n == 0)
        {
            std::cout << "create success: " << name << std::endl;
            tids->emplace_back(tid);
        }
    }
}

void WaitThread(std::vector<pthread_t> tids)
{
    for (auto &tid : tids)
    {
        pthread_join(tid, nullptr);
    }
}

int main()
{
    std::vector<pthread_t> tids; // 这里放所有的线程的tid
    StartMaster(&tids);          // 启动主线程
    StartSlaver(&tids, 5);       // 启动新线程
    WaitThread(tids);            // 等待新线程
    return 0;
}

#ifndef __BLOCK_QUEUE_HPP__
#define __BLOCK_QUEUE_HPP__

#include <iostream>
#include <string>
#include <queue>
#include <pthread.h>

template <class T>
class BlockQueue
{
private:
    bool IsFull()
    {
        return _block_queue.size() == _cap;
    }
    bool IsEmpty()
    {
        return _block_queue.empty();
    }

public:
    BlockQueue(int cap) : _cap(cap)
    {
        _consum_wait_num = 0;
        _product_wait_num = 0;
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_product_cond, nullptr);
        pthread_cond_init(&_consum_cond, nullptr);
    }

    void Enqueue(T &in) // 生产者用的接口
    {
        pthread_mutex_lock(&_mutex);
        while (IsFull())
        {
            // 生产线程去等待,是在临界区中休眠的！你现在还持有锁呢！！！
            // 1. pthread_cond_wait调用是: a. 让调用进程等待 b. 自动释放曾经持有的_mutex锁
            _product_wait_num++;
            pthread_cond_wait(&_product_cond, &_mutex);
            _product_wait_num--;
        }
        // 进行生产
        _block_queue.push(in);
        // 通知消费者来消费
        if (_consum_wait_num > 0)
        {
            pthread_cond_signal(&_consum_cond);
        }
        pthread_mutex_unlock(&_mutex); // 其实解锁和唤醒条件顺序无所谓，先唤醒后那边等着，解锁后直接竞争
        // 如果先解锁，后唤醒：先解锁没任何效果，因为都在wait那里等，一唤醒就直接得到锁
    }

    void Pop(T *out) // 消费者用的接口
    {
        pthread_mutex_lock(&_mutex);
        while (IsEmpty())
        {
            // 消费线程去等待,是在临界区中休眠的！你现在还持有锁呢！！！
            // 1. pthread_cond_wait调用是: a. 让调用进程等待 b. 自动释放曾经持有的_mutex锁——
            _consum_wait_num++;
            pthread_cond_wait(&_consum_cond, &_mutex);
            _consum_wait_num--;
        }

        // 进行消费
        *out = _block_queue.front();
        _block_queue.pop();
        // 通知生产者来生产
        if (_product_wait_num > 0)
        {
            pthread_cond_signal(&_product_cond);
        }
        pthread_mutex_unlock(&_mutex);
    }
    ~BlockQueue()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_product_cond);
        pthread_cond_destroy(&_consum_cond);
    }

private:
    std::queue<T> _block_queue;   // 阻塞队列
    int _cap;                     // 总上限
    pthread_mutex_t _mutex;       // 保护_block_queue的锁
    pthread_cond_t _product_cond; // 专门给生产者提供的条件变量
    pthread_cond_t _consum_cond;  // 专门给消费者提供的条件变量
    int _product_wait_num;        // 等待的生产者数量
    int _consum_wait_num;         // 等待的消费者数量
};

#endif

#pragma once

#include <iostream>
#include <string>

class Task
{
public:
    Task() {}
    Task(int a, int b) : _a(a), _b(b), _result(0)
    {
    }
    void Excute()
    {
        _result = _a + _b;
    }
    std::string ResultToString()
    {
        return std::to_string(_a) + "+" + std::to_string(_b) + "=" + std::to_string(_result);
    }
    std::string DebugToString()//测试的时候使用
    {
        return std::to_string(_a) + "+" + std::to_string(_b) + "=?";
    }

private:
    int _a;
    int _b;
    int _result;
};

#include "BlockQueue.hpp"
#include "Thread.hpp"
#include "Task.hpp"
#include <string>
#include <vector>
#include <unistd.h>

using namespace ThreadModule;
int a = 10;

using blockqueue_t = BlockQueue<Task>;

void Consumer(blockqueue_t &bq)
{
    while (true)
    {
        // 1.从Blockqueue里面取出任务
        Task t;
        bq.Pop(&t);
        // 2.开始执行任务
        t.Excute();
        std::cout << "Consumer Consum result is : " << t.ResultToString() << std::endl;
        sleep(2);
    }
}

void Productor(blockqueue_t &bq)
{
    int cnt = 1;
    srand(time(nullptr));
    while (true)
    {
        int a = rand() % 10;
        int b = rand() % 5;
        Task t(a, b);
        bq.Enqueue(t);
        cnt++;
    }
}

void StartComm(std::vector<Thread<blockqueue_t>> *threads, int num, blockqueue_t &bq, func_t<blockqueue_t> func)
{
    for (int i = 0; i < num; i++)
    {
        std::string name = "thread-" + std::to_string(i + 1);
        threads->emplace_back(func, bq, name);
        threads->back().Start();
    }
}

void StartConsumer(std::vector<Thread<blockqueue_t>> *threads, int num, blockqueue_t &bq)
{
    StartComm(threads, num, bq, Consumer);
}

void StartProductor(std::vector<Thread<blockqueue_t>> *threads, int num, blockqueue_t &bq)
{
    StartComm(threads, num, bq, Productor);
}

void WaitAllThread(std::vector<Thread<blockqueue_t>> &threads)
{
    for (auto &thread : threads)
    {
        thread.Join();
    }
}

int main()
{
    blockqueue_t *bq = new blockqueue_t(5);
    std::vector<Thread<blockqueue_t>> threads;

    StartProductor(&threads, 1, *bq);
    StartConsumer(&threads, 1, *bq);

    WaitAllThread(threads);

    return 0;
}