C++: 多线程同步机制

C++生产者消费者

  • 基于锁和条件变量的同步机制
  • 基于pipe的同步机制

基于变量的同步机制，在面对select/poll等场景时，无法做到loop线程的唤醒。因此，面对读写事件时，需要使用pipe来进行同步。

  1. 基于锁和条件变量的同步机制
#include <iostream>
#include <mutex>
#include <thread>
#include <memory>
#include <vector>
#include <string>

class context{

public:
    virtual int run()=0;
};

class TaskManager{
private:
    std::mutex mtx;
    std::condition_variable cond;

    std::thread m_handle;

    std::thread::id thread_id;
    bool stopped;
    std::string thread_name;

private:
    virtual void entry();
    std::vector<std::shared_ptr<context>> task;

    void _process(std::vector<std::shared_ptr<context>>&);

public:
    void create(const std::string& name){
        thread_name = name;
        m_handle = std::thread(&TaskManager::entry,this);
        thread_id = m_handle.get_id();
    }

    void join(){
        if (m_handle.joinable())
            m_handle.join();
    }
    void detach(){
        m_handle.detach();
    }
    int add_task(std::shared_ptr<context>);
    void stop(){
        std::lock_guard<std::mutex> lock{mtx};
        stopped = true;
        cond.notify_all();
    }
};

void TaskManager::entry(){
    std::unique_lock<std::mutex> lock{mtx};
    while (!stopped || !task.empty()){
        cond.wait(lock,[this](){return (!task.empty() || stopped);});
        std::vector<std::shared_ptr<context>> tmp;
        tmp.swap(task);
        lock.unlock();
        _process(tmp);
        lock.lock();
    }
}

void TaskManager::_process(std::vector<std::shared_ptr<context>>& tsk){
    for (const auto& tk:tsk){
        tk->run();
    }
}

int TaskManager::add_task(std::shared_ptr<context> ts){
    if (stopped)
        return -1;
    {
        std::lock_guard<std::mutex> lock{mtx};
        task.push_back(ts);
        cond.notify_all();
    }
    return 0;
}


class test: public context{
private:
    std::string name;
public:
    test(const std::string& name):name(name){}
    int run() override {
        std::cout<<"Task Run: "<<name<<std::endl;
        return 0;
    }
 };

int main(int argc,char* argv[]){
    std::shared_ptr<TaskManager> tm = std::make_shared<TaskManager>();
    tm->create("TaskManager");
    tm->add_task(std::make_shared<test>("TaskA"));
    tm->add_task(std::make_shared<test>("TaskB"));
    tm->add_task(std::make_shared<test>(std::string("TaskC")));
    tm->stop();
    tm->join();
}
  1. 基于pipe的同步机制
    基于锁和条件变量的同步机制，无法适用于读写事件的同步，如select/poll/epoll。主线程在select中等待时，则需要其它唤醒方式。
  • 可以使用pipe进行同步
  • 将pipe的读端加入select监听
  • 需要唤醒时，对pipe写端进行写入，此时唤醒等待线程

实现如下，注册监听一组fd，利用管道实现对监听fd的添加和删除：

#include <iostream>
#include <mutex>
#include <thread>
#include <memory>
#include <map>
#include <string>
#include <functional>
#include <array>
#include <string_view>
#include <algorithm>

#include <fcntl.h>
#include <unistd.h>
#include <sys/select.h>

int pipe_ocloexc(int pipefd[2]){
     /*
     ** pipe init
     */
    if (pipe2(pipefd, O_CLOEXEC) < 0){
        return -1;
    }   
    if (::fcntl(pipefd[0], F_SETFL, O_NONBLOCK) < 0){
         ::close(pipefd[1]);
         ::close(pipefd[0]);
         return -1;
    }
    return 0;
}

struct io_handler{
    io_handler(const std::string user):user(user){
        int r = pipe_ocloexc(pipefd);
        if (r < 0){
            exit(-1);
        }
    }
    std::string user;
    int pipefd[2];
    std::function<int(void*)> callback;
};

int read(int fd){
    std::array<char, 256> buf;
    int ret = ::read(fd, (void *)buf.data(), buf.size());
    if (ret < 0) { 
        return -1;
    }
    std::cout<<"Read Data: "<<std::string_view(buf.data(),ret)<<std::endl;
    return 0;
}

int process(void* arg){
    io_handler* ptr = (io_handler*)arg;
    //std::cout<<"IO: "<<ptr->user<<std::endl;
    int ret = read(ptr->pipefd[0]);
    if (ret < 0 )
        return -1;
    return 0;
}

class TaskManager{
private:
    std::mutex mtx;
    int thread_pipe[2];

    std::thread m_handle;

    std::thread::id thread_id;
    std::string thread_name;

    bool stopped;

private:
    virtual void entry();
    using mio = std::map<int,std::shared_ptr<io_handler>>;
    mio io_callback;

    int signal_thread();
    int clear_signal(int);

public:
    int create(const std::string& name);

    void join();
    void detach();
    int register_fd(std::shared_ptr<io_handler>);
    void unregister_fd(std::shared_ptr<io_handler>);
    void stop();
    virtual ~TaskManager(){
        ::close(thread_pipe[1]);
        ::close(thread_pipe[0]);
    }
};

int TaskManager::create(const std::string& name){
    thread_name = name;

    if (pipe_ocloexc(thread_pipe)<0)
        return -1;

    m_handle = std::thread(&TaskManager::entry,this);
    thread_id = m_handle.get_id();
    return 0;
}

int TaskManager::clear_signal(int fd)
{
    std::array<char, 256> buf;
    int ret = ::read(fd, (void *)buf.data(), buf.size());
    if (ret < 0) { 
        return -1;
    }
    return 0;
}

void TaskManager::join(){
    if (m_handle.joinable())
        m_handle.join();
}

void TaskManager::detach(){
    m_handle.detach();
}

void TaskManager::stop(){
    if (stopped)
        return;
    stopped = true;
    signal_thread();
}

void TaskManager::entry(){
    fd_set fdread;
    fd_set fdwrite;
    fd_set fdexcep;
    while (!stopped || !io_callback.empty()){
        int maxfd = -1;
        FD_ZERO(&fdread);
        FD_ZERO(&fdwrite);
        FD_ZERO(&fdexcep);

        FD_SET(thread_pipe[0],&fdread);
        maxfd = thread_pipe[0]+1;

        std::unique_lock<std::mutex> lock{mtx};
        for (const auto& io:io_callback){
            FD_SET(io.first,&fdread);
            if (io.first >= maxfd){
                maxfd = io.first+1;
            }
        }
        lock.unlock();
        int ret = select(maxfd,&fdread,&fdwrite,&fdexcep,NULL);
        if ( ret < 0)
            return;
        lock.lock();
        for (const auto& io:io_callback){
            if (FD_ISSET(io.first,&fdread)){
                //clear_signal(io.first);
                (io.second)->callback(io.second.get());
            }
        }
        if (FD_ISSET(thread_pipe[0],&fdread)){
            clear_signal(thread_pipe[0]);
        }
        lock.unlock();
    }
}

int TaskManager::signal_thread(){
    uint32_t buf = 0;
    int ret = write(thread_pipe[1], (void *)&buf, sizeof(buf));
    if (ret < 0) {
        return -1;
    }
    return 0;
}

int TaskManager::register_fd(std::shared_ptr<io_handler> handler){
    if (stopped)
        return -1;
    {
        std::lock_guard<std::mutex> lock{mtx};
        io_callback.insert(std::pair<int,std::shared_ptr<io_handler>>(handler->pipefd[0],handler));
        signal_thread();
    }
    return 0;
}

void TaskManager::unregister_fd(std::shared_ptr<io_handler> io){
    {
        std::lock_guard<std::mutex> lock{mtx};
        io_callback.erase(io->pipefd[0]);
        signal_thread();
    }
}

int test(std::shared_ptr<io_handler> io){
    std::array<char,256> buf;
    std::copy(io->user.cbegin(),io->user.cend(),buf.data());
    int ret = write(io->pipefd[1], (void *)&buf, io->user.size());
    if (ret < 0) {
        return -1;
    }
    return 0;
}

int main(int argc,char* argv[]){
    std::shared_ptr<TaskManager> tm = std::make_shared<TaskManager>();
    tm->create("TaskManager");
    std::shared_ptr<io_handler> io1 = std::make_shared<io_handler>("io_1");
    io1->callback = process;
    std::shared_ptr<io_handler> io2 = std::make_shared<io_handler>("io_2");
    io2->callback = process;
    std::shared_ptr<io_handler> io3 = std::make_shared<io_handler>("io_3");
    io3->callback = process;
    tm->register_fd(io1);
    tm->register_fd(io2);
    tm->register_fd(io3);

    test(io1);
    test(io3);
    test(io2);

    sleep(3);
    
    tm->unregister_fd(io1);
    tm->unregister_fd(io2);
    tm->unregister_fd(io3);
    tm->stop();
    tm->join();
}
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 199,519评论 5赞 468
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 83,842评论 2赞 376
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 146,544评论 0赞 330
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 53,742评论 1赞 271
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 62,646评论 5赞 359
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,027评论 1赞 275
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,513评论 3赞 390
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,169评论 0赞 254
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,324评论 1赞 294
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,268评论 2赞 317
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,299评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 32,996评论 3赞 315
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,591评论 3赞 303
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,667评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,911评论 1赞 255
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 42,288评论 2赞 345
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 41,871评论 2赞 341