线程池问题求助， 谢谢！

lucky多多求

正在实现一个线程池，现有一个问题点不太明确，我的主线程里初始化创建N个work_thread和一个manage_thread，想法是等所有的创建work_thread准备就绪后，主线程才继续向下运行,manage_thread暂不用考虑就绪与否，我的实现大概是
tp_init（）
{
  创建N个work_thread；

  创建manage_thread；

  等待所有work_thread就绪的通知；   
}

我的想法：1. tp_init（）里用pthread_cond_wait等待条件变量， work_thread的工作函数里检查已经就绪的线程的个数，到达N个后 pthread_cond_signal通知主线程继续运行， 但是pthread_cond_wait pthread_cond_signal的时序是一个问题，如果pthread_cond_signal 在 pthread_cond_wait 之前，信号会丢失，主线程永久阻塞。
          2. 用一个全局变量counter和锁实现，每个work_thread就绪后counter++；主线程轮询或sleep()检查，但这个问题是各个线程会频繁加解锁，效率是不是会很低？？？
          3. 使用一个信号实现，主线程阻塞信号调用 sig_wait()；所有线程就绪后pthread_kill（）；主线程继续运行。

          请帮忙看看这三种是不是都能实现，每一种的优缺点？ 第一种的时序问题怎么搞定？

linux_c_py_php

1, 条件变量毫无问题, 你觉得会死锁说明你不会用.(1:N是可以不广播的, N:M必须广播, 但如果你不会设计, 那么让生产者和消费者无条件的广播)
2, 只是一个启动, 谈什么效率, 如果线程池启动是比较快的, 轮一下无所谓.
3, 你自己都说不明白, 怎么实现? (我帮你说了, 用实时信号, 主线程sigwait数一数信号个数就可以了.)

我建议做法:
1, 条件变量:
  1) 如果线程完成初始化之前, 管理员给进程发送信号希望进程退出, 似乎不太好处理, 这是它的缺点.
  2) 哪个线程完成了初始化, 你还需要一个额外的容器来保存, 在主线程每次被唤醒后要消耗整个容器内的元素进行线程监控.

2, select + pipe: 主线程select N秒超时监听一个pipe[0], 每一个线程初始化完成后, 将pthread_self()写给pipe, 主线程读取pipe即可知哪个线程准备好了. 而且在select超时期间, 你可以处理信号函数事件, 比较灵活.

lucky多多求

我没有说条件变量会死锁，我是说时序问题，假设这种情况，所有工作线程都已经就绪了，会发送信号，在发送信号之前，主线程必须已经调用了pthread_signal_wait，如果发送信号在调用wait之前，就起不到效果，主线程阻塞，
我现在的实现是第二种方法，因为只是启动的时候考虑，直接一个全局的计数和锁来实现，
至于第三，你说的是主线程数信号个数，我之前想法是work_thread那里计数个数，然后发信号，原理差不多

主要是第一个的时序问题，怎么就保证 wait一定得再signal之后
回复 2# linux_c_py_php


   

linux_c_py_php

看样你是完全不知道条件变量是怎么用的, 还强词夺理, 那你就自己琢磨去.

liuspring6

工作线程就绪用
__sync_add_and_fetch   
增加条件变量

主线程用
__sync_val_compare_and_swap
判断条件变量和目标数是否相同

lucky多多求

请问我怎么强词夺理了，不是在和你讨论嘛？ 如果用条件变量这种方式的话，在所有工作线程都就绪后调用pthread_cond_signal通知主线程，我的意思是在这个通知之前是不是必须保证主线程已经在调用pthread_cond_wait等待了？ 如果就绪后pthread_cond_signal了，而主线程还没有调用pthread_cond_wait进入等待态，就没法唤醒了啊。谢谢
回复 4# linux_c_py_php


   

linux_c_py_php

给你个简单例子, 主要示范条件变量怎么用.

[root@vps616 cpp]# cat main.cpp
#include <pthread.h>
#include <unistd.h>

#include <iostream>

class ThreadPoolManager
{
    public:
        typedef void* (*ThreadMain)(void*);

        enum ThreadErrorNumber
        {
            ALREADY_START,
            CREATING_FAIL,
            MEMORY_LACK,
        };

        ThreadPoolManager(int thread_number, ThreadMain thread_main)
        {
            m_thread_info   = NULL;
            m_thread_main   = thread_main;
            m_thread_number = thread_number;
            m_thread_online = 0;
            pthread_mutex_init(&m_mutex, NULL);
            pthread_cond_init(&m_cond, NULL);
        }

        ~ThreadPoolManager()
        {
            //User should join thread explicitly
            if (m_thread_info)
            {
                delete []m_thread_info;
            }
            pthread_mutex_destroy(&m_mutex);
            pthread_cond_destroy(&m_cond);
        }

        int StartThreadPool(ThreadErrorNumber &errnum)
        {
            if (m_thread_info)
            {
                errnum = ALREADY_START;
                return -1;
            }
            
            m_thread_info = new (std::nothrow) ThreadInfo[m_thread_number];
            
            if (!m_thread_info)
            {
                errnum = MEMORY_LACK;
                return -1;
            }

            int i;

            for (i = 0; i < m_thread_number; ++ i)
            {
                if (m_thread_info.StartThread(this, m_thread_main) == -1)
                {
                    break;
                }
            }
            
            if (i != m_thread_number)
            {
                //better to kill the existing thread here
                std::cerr << "[Pthread Create Error]" << std::endl;
                errnum = CREATING_FAIL;
                return -1;
            }

            pthread_mutex_lock(&m_mutex);

            while (m_thread_online < m_thread_number)
            {
                pthread_cond_wait(&m_cond, &m_mutex);
            }

            pthread_mutex_unlock(&m_mutex);

            return 0;
        }

        void JoinThreadPool()
        {
            if (!m_thread_info)
            {
                return;
            }

            int i;

            for (i = 0; i < m_thread_number; ++ i)
            {
                std::cout << m_thread_info.m_tid << " -> Joined" << std::endl;
                m_thread_info.JoinThread();
            }
        }
        
        void TellThreadStart(pthread_t tid)
        {
            pthread_mutex_lock(&m_mutex);
            ++ m_thread_online;
            std::cerr << tid << " -> online" << std::endl;
            pthread_cond_signal(&m_cond);
            pthread_mutex_unlock(&m_mutex);
        }
   
        class ThreadInfo
        {
            public:
                ThreadInfo()
                {
                    m_tid = 0;
                    m_is_working = false;
                }
                int StartThread(ThreadPoolManager *manager, ThreadMain thread_main)
                {
                    m_manager = manager;
                    m_is_working = true;

                    if (pthread_create(&m_tid, NULL, thread_main, this) == -1)
                    {
                        m_is_working = false;
                        return -1;
                    }

                    return 0;
                }
                void TellThreadStart()
                {
                    m_manager->TellThreadStart(m_tid);
                }
                void JoinThread()
                {
                    if (m_is_working)
                    {
                        pthread_join(m_tid, NULL);
                        m_is_working = false;
                    }
                }
                ThreadPoolManager  *m_manager;
                pthread_t           m_tid;
                bool                m_is_working;
        };
      
    private:
        pthread_cond_t  m_cond;
        pthread_mutex_t m_mutex;
        ThreadMain      m_thread_main;   
        ThreadInfo     *m_thread_info;
        int             m_thread_number;
        int             m_thread_online;
};

void* thread_main(void *arg)
{
    ThreadPoolManager::ThreadInfo *thread_info = (ThreadPoolManager::ThreadInfo*)arg;

    std::cerr << pthread_self() << " -> Start" << std::endl;
    sleep(2);
    thread_info->TellThreadStart();

    return NULL;
}

int main(int argc, char* const argv[])
{
    ThreadPoolManager thread_manager(10, thread_main);
    ThreadPoolManager::ThreadErrorNumber errnum;

    thread_manager.StartThreadPool(errnum);
    thread_manager.JoinThreadPool();

    return 0;
}

[root@vps616 cpp]# ./main
139920094529280 -> Start
139920084039424 -> Start
139920073549568 -> Start
139920063059712 -> Start
139920105019136 -> Start
139920052569856 -> Start
139920042080000 -> Start
139920031590144 -> Start
139920115508992 -> Start
139920125998848 -> Start
139920084039424 -> online
139920094529280 -> online
139920073549568 -> online
139920063059712 -> online
139920105019136 -> online
139920052569856 -> online
139920042080000 -> online
139920031590144 -> online
139920115508992 -> online
139920125998848 -> online
139920125998848 -> Joined
139920115508992 -> Joined
139920105019136 -> Joined
139920094529280 -> Joined
139920084039424 -> Joined
139920073549568 -> Joined
139920063059712 -> Joined
139920052569856 -> Joined
139920042080000 -> Joined