muduo库阅读（5）——线程

/*
* 线程类
*/
namespace muduo
{

class Thread : boost::noncopyable
{
public:
// 线程回调函数
typedef boost::function<void ()> ThreadFunc;

explicit Thread(const ThreadFunc&, const string& name = string());
#ifdef __GXX_EXPERIMENTAL_CXX0X__
explicit Thread(ThreadFunc&&, const string& name = string());
#endif
~Thread();

// 启动线程
void start();

// 等待线程结束
int join(); // return pthread_join()

// 判断线程是否已经启动
bool started() const { return started_; }
// pthread_t pthreadId() const { return pthreadId_; }

// 获取进程id
pid_t tid() const { return *tid_; }

// 获取线程的名字
const string& name() const { return name_; }

// 创建了多少个线程
static int numCreated() { return numCreated_.get(); }

private:
void setDefaultName();

bool       started_;					// 是否已经开始执行
bool       joined_;						// 等待结束
pthread_t  pthreadId_;					// 线程id
boost::shared_ptr<pid_t> tid_;			// 线程id
ThreadFunc func_;						// 线程函数（执行线程任务的地方）
string     name_;						// 线程的名字

static AtomicInt32 numCreated_;			// 已经创建的线程数量，是个静态对象
};

}

namespace muduo
{
/*
* 当前线程
*/
namespace CurrentThread
{
// 线程id
__thread int t_cachedTid = 0;

// 线程id（字符串）
__thread char t_tidString[32];
__thread int t_tidStringLength = 6;

// 默认的线程名
__thread const char* t_threadName = "unknown";

// 判断两个类型是否为同一类型
const bool sameType = boost::is_same<int, pid_t>::value;
BOOST_STATIC_ASSERT(sameType);
}

namespace detail
{

// 获取线程id
pid_t gettid()
{
// 直接使用系统调用
return static_cast<pid_t>(::syscall(SYS_gettid));
}

// 当一个进程调用了fork之后，子进程需要调用这个函数
void afterFork()
{
muduo::CurrentThread::t_cachedTid = 0;

// 当前线程是主线程
muduo::CurrentThread::t_threadName = "main";

// 获取线程id
CurrentThread::tid();
// no need to call pthread_atfork(NULL, NULL, &afterFork);
}

/*
* 线程初始化类，在主线程中使用，在创建任何线程之前使用
*/
class ThreadNameInitializer
{
public:
ThreadNameInitializer()
{
muduo::CurrentThread::t_threadName = "main";
CurrentThread::tid();
pthread_atfork(NULL, NULL, &afterFork);
}
};

// 线程初始化对象
ThreadNameInitializer init;

/*
* 线程数据结构
*/
struct ThreadData
{
typedef muduo::Thread::ThreadFunc ThreadFunc;

// 回调函数
ThreadFunc func_;

// 线程名字
string name_;

// 线程id
// weak_ptr不具有普通指针的行为，没有重载operator*和->,它的最大作用在于协助shared_ptr工作，像旁观者那样观测资源的使用情况.
boost::weak_ptr<pid_t> wkTid_;

ThreadData(const ThreadFunc& func,
const string& name,
const boost::shared_ptr<pid_t>& tid)
: func_(func),
name_(name),
wkTid_(tid)
{ }

// 执行线程函数
void runInThread()
{
// 获取线程id
pid_t tid = muduo::CurrentThread::tid();

boost::shared_ptr<pid_t> ptid = wkTid_.lock();
if (ptid)
{
*ptid = tid;
ptid.reset();
}

muduo::CurrentThread::t_threadName = name_.empty() ? "muduoThread" : name_.c_str();

// 这个系统调用指令是为进程制定而设计的，明确的选择取决于option
// 这里就是设置线程的名字
::prctl(PR_SET_NAME, muduo::CurrentThread::t_threadName);

try
{
// 执行
func_();
// 执行完毕之后，将线程名字设置为finished
muduo::CurrentThread::t_threadName = "finished";
}
catch (const Exception& ex)
{
muduo::CurrentThread::t_threadName = "crashed";
fprintf(stderr, "exception caught in Thread %s\n", name_.c_str());
fprintf(stderr, "reason: %s\n", ex.what());
fprintf(stderr, "stack trace: %s\n", ex.stackTrace());
abort();
}
catch (const std::exception& ex)
{
muduo::CurrentThread::t_threadName = "crashed";
fprintf(stderr, "exception caught in Thread %s\n", name_.c_str());
fprintf(stderr, "reason: %s\n", ex.what());
abort();
}
catch (...)
{
muduo::CurrentThread::t_threadName = "crashed";
fprintf(stderr, "unknown exception caught in Thread %s\n", name_.c_str());
throw; // rethrow
}
}
};

/*
* 这个函数其实并没有多线程的作用
* 它是直接调用任务的执行函数，是阻塞的
* 它的主要作用是被线程类的start函数调用
*/
void* startThread(void* obj)
{
// 获取线程数据
ThreadData* data = static_cast<ThreadData*>(obj);

// 执行线程函数
data->runInThread();
delete data;
return NULL;
}

}
}

using namespace muduo;

// 缓存线程id
void CurrentThread::cacheTid()
{
if (t_cachedTid == 0)
{
t_cachedTid = detail::gettid();
t_tidStringLength = snprintf(t_tidString, sizeof t_tidString, "%5d ", t_cachedTid);
}
}

// 判断是否为主线程
bool CurrentThread::isMainThread()
{
// 主线程的线程id和进程id相同
return tid() == ::getpid();
}

// 睡眠多少纳秒（usec是纳秒数）
void CurrentThread::sleepUsec(int64_t usec)
{
// timespec的第一个成员是妙，第二个成员是纳秒
struct timespec ts = { 0, 0 };
ts.tv_sec = static_cast<time_t>(usec / Timestamp::kMicroSecondsPerSecond);
ts.tv_nsec = static_cast<long>(usec % Timestamp::kMicroSecondsPerSecond * 1000);

// nanosleep是一个高精度的睡眠函数
::nanosleep(&ts, NULL);
}

// 创建的线程的数量，用于计算一个进程当中线程的数量（由Thread库创建的线程）
AtomicInt32 Thread::numCreated_;

Thread::Thread(const ThreadFunc& func, const string& n)
: started_(false),
joined_(false),
pthreadId_(0),
tid_(new pid_t(0)),
func_(func),
name_(n)
{
// 设置默认名字，并且使线程数量加一
setDefaultName();
}

#ifdef __GXX_EXPERIMENTAL_CXX0X__
Thread::Thread(ThreadFunc&& func, const string& n)
: started_(false),
joined_(false),
pthreadId_(0),
tid_(new pid_t(0)),
func_(std::move(func)),
name_(n)
{
setDefaultName();
}

#endif

Thread::~Thread()
{

if (started_ && !joined_)
{
// 线程分离，调用这个函数之后，线程就不受到控制了
pthread_detach(pthreadId_);
}
}

/*
* 设置线程的默认名字
*/
void Thread::setDefaultName()
{
// 线程数量加一
int num = numCreated_.incrementAndGet();
if (name_.empty())
{
char buf[32];
snprintf(buf, sizeof buf, "Thread%d", num);
name_ = buf;
}
}

/*
* 启动线程
*/
void Thread::start()
{
// 断言线程还没有执行
assert(!started_);
started_ = true;

// FIXME: move(func_)
// 创建线程，调用了startThread
detail::ThreadData* data = new detail::ThreadData(func_, name_, tid_);
if (pthread_create(&pthreadId_, NULL, &detail::startThread, data))
{
started_ = false;
delete data; // or no delete?
LOG_SYSFATAL << "Failed in pthread_create";
}
}

/*
* 等待线程结束
*/
int Thread::join()
{
assert(started_);
assert(!joined_);
joined_ = true;
return pthread_join(pthreadId_, NULL);
}