线程操作对象

线程操作对象

有关线程的操作，可以查看Linux系统编程：线程

今天介绍三个对象：CThread，CMutex，CCond。分别负责创建线程，线程互斥以及线程同步

CThread

CThread继承ThreadClass，ThreadClass是一个只包含纯虚函数run()接口基类，如下：

class ThreadClass
{
public:
    virtual void run() = 0;
};

CThread有两种使用方式：

1. 继承CThread，实现run()方法，根据C++的特性，派生类在重定义一个基类的函数时，基类的函数会被隐藏，所以在调用start()方法启动线程时，会执行派生类的run()方法
2. 继承ThreadClass，实现run()方法，将派生类作为参数创建CThread，调用CThread的start()方法启动线程

线程数据设置：

• pthread_key_create分配用于标识进程中线程特定数据的键。
• pthread_getspecific根据键获取特定的数据

这里用来保存线程名称，函数原型如下：

int pthread_key_create(pthread_key_t *key, void (*destructor)(void*));
void *pthread_getspecific(pthread_key_t key);
int pthread_setspecific(pthread_key_t key, const void *value);

具体实现如下：

#include "cthread.h"
#include <unistd.h>

pthread_key_t key;
bool is_key_create = false;

void *CThread::staticRun(void *ptr)
{
    CThread *thread = static_cast<CThread *>(ptr);
    thread->preparationRun();
    return thread;
}

void CThread::preparationRun()
{
    int ret, pthread_cancel;

    //do NOT invoke any thread-unsafe function in cancelable state
    //OR pthread_cancel to cancel thread may fall into deadlock.
    ret = pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &pthread_cancel);
    setCurrentThreadName((char*)getThreadName().c_str());

    mThreadStatus = C_THREAD_STATUS_RUNNING;

    run();

    mThreadStatus = C_THREAD_STATUS_EXIT;

    // printf("pthread_exit(%u)[%s].", m_thread_id, mThreadName.c_str());
    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &pthread_cancel);
    pthread_exit(nullptr);
}

void CThread::run()
{
    if (mThreadClass != nullptr)
    {
        mThreadClass->run();
    }
}

CThread::CThread()
{
    mThreadId = 0;
    mThreadStatus = C_THREAD_STATUS_NEW;
    pthread_attr_init(&mThreadAttr);
}

CThread::CThread(ThreadClass *thread_class)
{
    mThreadClass = thread_class;
    CThread();
}

CThread::~CThread()
{
    pthread_attr_destroy(&mThreadAttr);
}

bool CThread::start(std::string name)
{
    int ret = pthread_create(&mThreadId, &mThreadAttr, staticRun, this);

    if (ret != 0)
    {
        mThreadId = 0;
        return false;
    }

    mThreadName = name;
    return true;
}

int CThread::stop()
{
    int ret = -1;

    if (mThreadId > 0)
    {
        ret = pthread_cancel(mThreadId);
        // printf("pthread_cancel(%u)[%s], ret:%d.", m_thread_id, mThreadName.c_str(), ret);
    }

    return ret;
}

void CThread::join()
{
    int ret, count;
    for(count = 0; count < 20; count++)
    {
        if (mThreadId > 0)
        {
            ret = pthread_tryjoin_np(mThreadId, nullptr);
            // printf("pthread_tryjoin_np(%u)[%s], ret:%d.", mThreadId, mThreadName.c_str(), ret);
            if (ret == 0)
            {
                ret = pthread_join(mThreadId, nullptr);
                // printf("pthread_join(%u)[%s], ret:%d.", mThreadId, mThreadName.c_str(), ret);
                return;
            }
        }
        sleep(10);
    }
}

void CThread::join(unsigned long millis_time)
{
    if (mThreadId > 0)
    {
        if (millis_time == 0)
        {
            join();
        }
        else
        {
            // printf("join thread(%u)[%s], warit:%d.", m_thread_id, mThreadName.c_str(), millis_time);
            while (mThreadStatus != C_THREAD_STATUS_EXIT && millis_time > 0)
            {
                sleep(1);
                millis_time--;
            }
        }
    }
}

void CThread::sleep(unsigned long millis_time)
{
    usleep(millis_time * 1000);
}

void CThread::setThreadName(const char *name)
{
    mThreadName = name;
}

void CThread::setThreadInfo(const char *name)
{
    if (name != nullptr)
    {
        setThreadName(name);
        setCurrentThreadName(name);
    }
}

void CThread::setCurrentThreadName(const char *name)
{
    int ret = -1;
    if (!is_key_create)
    {
        ret = pthread_key_create(&key, nullptr);
        // printf("pthread_key_create()[%s], ret:%d.", name, ret);
        if (ret == 0)
        {
            is_key_create = true;
        }
    }

    if (is_key_create)
    {
        ret = pthread_setspecific (key, (void *)name);
        // printf("pthread_setspecific()[%s], ret:%d.", name, ret);
        if(ret != 0)
        {
            // printf("ret = %d, system crash", ret);
        }
    }
}

std::string CThread::getCurrentThreadName()
{
    char *name = nullptr;
    if (is_key_create)
    {
        name = (char *)pthread_getspecific(key);
        if(name != nullptr)
        {
            return name;
        }
    }

    // printf("get current name fail.");
    return "Unknown";
}

int CThread::getCurrentThreadId()
{
    return pthread_self();
}

注：线程中可能会使用syslog或其他异步信号不安全的函数，所以当收到cancel信号可能会产生死锁，所以屏蔽cancel信号，设置PTHREAD_CANCEL_DISABLE标志位

CMutex

CMutex是一个简单的互斥锁实现类，提供lock和unloke方法，实现如下：

#include <string.h>
#include "cmutex.h"

CMutex::CMutex()
{
    memset(&mAttr, 0, sizeof(mAttr));
    pthread_mutexattr_init(&mAttr);
    pthread_mutex_init(&mMutex, nullptr);
}

//Recursive or not
CMutex::CMutex(bool recursive)
{
    memset(&mAttr, 0, sizeof(mAttr));
    pthread_mutexattr_init(&mAttr);
    if (recursive)
    {
        pthread_mutexattr_settype(&mAttr, PTHREAD_MUTEX_RECURSIVE);
        pthread_mutex_init(&mMutex, &mAttr);
    }
    else
    {
        pthread_mutex_init(&mMutex, nullptr);
    }
}

CMutex::~CMutex()
{
    pthread_mutex_destroy(&mMutex);
    pthread_mutexattr_destroy(&mAttr);
}

int CMutex::lock()
{
    return pthread_mutex_lock(&mMutex);
}

int CMutex::unlock()
{
    return pthread_mutex_unlock(&mMutex);
}

pthread_mutex_t *CMutex::getMutex()
{
    return &mMutex;
}

CCond

CCond是一个简单的线程同步实现类，主要提供了wait和signal方法，实现如下：

#include "ccond.h"
#include <ctime>
#include <cerrno>

CCond::CCond()
{
    pthread_condattr_init(&mCondAttr);
    /*
    CLOCK_MONOTONIC:
    The timing starts from the moment the system starts,
    independent of the system time being changed by the user
    */
    pthread_condattr_setclock(&mCondAttr, CLOCK_MONOTONIC);
    pthread_cond_init(&mCond, &mCondAttr);
}

CCond::~CCond()
{
    pthread_condattr_destroy(&mCondAttr);
    pthread_cond_destroy(&mCond);
}

void CCond::wait(CMutex *mutex)
{
    if (mutex != nullptr)
    {
        pthread_cond_wait(&mCond, mutex->getMutex());
    }
}

int CCond::wait(CMutex *mutex, unsigned int millis_time)
{
    int wait_value = ETIMEDOUT;
    timespec wait_time;

//    clock_gettime(CLOCK_REALTIME, &wait_time);
    clock_gettime(CLOCK_MONOTONIC, &wait_time);

    wait_time.tv_sec += millis_time / 1000;
    wait_time.tv_nsec += (millis_time % 1000) * 1000000;

    if (mutex != nullptr)
    {
        wait_value = pthread_cond_timedwait(&mCond, mutex->getMutex(), &wait_time);
        if (wait_value == ETIMEDOUT)
        {
            return 0;
        }
        else
        {
            return 1;
        }
    }
    return -1;
}

void CCond::signal()
{
    pthread_cond_signal(&mCond);
}