Thrift线程和状态机分析

class ThreadManager::Task: public Runnable
{
public:
void run()
{
// runnable_实际为TNonblockingServer::TConnection::Task
runnable_->run();
}
private:
// 这里的Runnable实际为TNonblockingServer::TConnection::Task
// 在TNonblockingServer::TConnection::transition()中被push进来
boost::shared_ptr<Runnable> runnable_;
};

TNonblockingServer::TConnection::Task
{
public:
void run()
{
// 回调
processor_->process(input_, output_, connectionContext_);
// 回调完后通知，
// 从工作线程重回到IO线程
connection_->notifyIOThread(); // ioThread_->notify(this);
// 这个将触发TNonblockingIOThread::notifyHandler()
}
};
TNonblockingIOThread::notifyHandler()
{
// 从管道中取出connection的指针地址
TNonblockingServer::TConnection* connection = NULL;
int nBytes = recv(fd, cast_sockopt(&connection), kSize, 0);
connection->transition(); // 进入状态转换函数
}

class TNonblockingServer: public TServer
{
public:
void serve() // 用户可以直接调用server()，但直接调用run()是更好的做法
{
// 创建socket监听
// 创建TNonblockingIOThread
// 通过Thread启动TNonblockingIOThread
}
};
class TServer: public concurrency::Runnable
{
public:
virtual void serve() = 0;
virtual void run() // 用户也可以直接调用run()
{
serve();
}
};

// 线程
// thrift支持原生posix线程和boost线程
void PthreadThread::start()
{
// PthreadThread是一个Posix线程类
pthread_create(&pthread_, &thread_attr, threadMain, (void*)selfRef);
}
static void* PthreadThread::threadMain(void* arg)
{
thread->runnable()->run();
}
// 以下为IO线程
/// Three states for sockets: recv frame size, recv data, and send mode
enum TSocketState
{
SOCKET_RECV_FRAMING,
SOCKET_RECV,
SOCKET_SEND
};
/**
* Five states for the nonblocking server:
*  1) initialize
*  2) read 4 byte frame size
*  3) read frame of data
*  4) send back data (if any)
*  5) force immediate connection close
*/
enum TAppState
{
APP_INIT,            // 初始化
APP_READ_FRAME_SIZE, // 接收包大小
APP_READ_REQUEST,    // 接收包数据
APP_WAIT_TASK,
APP_SEND_RESULT,     // 发送数据
APP_CLOSE_CONNECTION // 关闭连接
};
// 启动监听和IO线程
void TNonblockingServer::serve()
{
createAndListenOnSocket();
for (uint32_t id = 0; id < numIOThreads_; ++id)
{
// TNonblockingIOThread是一个Runnable
// 以委托方式被运行在PthreadThread中
thread = new TNonblockingIOThread(this, id, listenFd, useHighPriorityIOThreads_);
ioThreads_.push_back(thread);
}
for (uint32_t i = 1; i < ioThreads_.size(); ++i)
{
// PthreadThread thread;
thread->start();
}
ioThreads_[0]->run(); // 这将阻塞调用线程
for (uint32_t i = 0; i < ioThreads_.size(); ++i)
{
ioThreads_[i]->join();
}
}
void TNonblockingIOThread::run()
{
eventBase_ = event_base_new();
// IO线程在启动时会调用registerEvents()
// 在registerEvents()中完成两个回调函数的注册：listenHandler和notifyHandler
// listenHandler回调负责接受请求，并创建连接对象
registerEvents();
event_base_loop(eventBase_, 0); // libevent
}
void TNonblockingIOThread::registerEvents()
{
// listenHandler和socket关联
event_set(&serverEvent_, listenSocket_, EV_READ | EV_PERSIST,
TNonblockingIOThread::listenHandler, server_);
// notifyHandler和pipe关联
event_set(?ificationEvent_, getNotificationRecvFD(), EV_READ | EV_PERSIST,
TNonblockingIOThread::notifyHandler, this);
}
static void listenHandler(evutil_socket_t fd, short which, void* v)
{
((TNonblockingServer*)v)->handleEvent(fd, which);
}
void TNonblockingServer::handleEvent(int fd, short which)
{
accept();
createConnection();
}
TNonblockingServer::TConnection* TNonblockingServer::createConnection()
{
// 会将自己绑定到一个线程
// 采用轮询的方式选择线程
// int selectedThreadIdx = nextIOThread_;
// nextIOThread_ = (nextIOThread_ + 1) % ioThreads_.size();
// std::stack connectionStack_;
// 使用了内存池connectionStack_
// App状态：APP_INIT
// Socket状态：SOCKET_RECV_FRAMING
}
static void eventHandler(evutil_socket_t fd, short /* which */, void* v)
{
assert(fd == ((TConnection*)v)->getTSocket()->getSocketFD());
((TConnection*)v)->workSocket();
}
void TNonblockingServer::TConnection::setFlags(short eventFlags)
{
event_set(&event_, tSocket_->getSocketFD(), eventFlags_, TConnection::eventHandler, this);
}
void TNonblockingServer::TConnection::setRead()
{
setFlags(EV_READ | EV_PERSIST);
}
void TNonblockingServer::TConnection::setWrite()
{
setFlags(EV_WRITE | EV_PERSIST);
}
void TNonblockingServer::TConnection::setIdle()
{
setFlags(0);
}
void TNonblockingServer::TConnection::workSocket()
{
case SOCKET_RECV_FRAMING:
TSocket::read(); // 接收包大小
transition();
case SOCKET_RECV:
TSocket::read(); // 接收包数据
transition();
case SOCKET_SEND:
TSocket::write_partial(); // 发送数据（非阻塞的）
transition();
}
void TNonblockingIOThread::notifyHandler(evutil_socket_t fd, short which, void* v)
{
recv();
connection->transition();
}
// transition()为状态迁移函数
void TNonblockingServer::TConnection::transition()
{
case APP_INIT:
setRead();
case APP_WAIT_TASK:
setWrite();
case APP_READ_REQUEST:
setIdle();
}
TNonblockingServer::TConnection::Task
{
public:
void run()
{
// 回调
processor_->process(input_, output_, connectionContext_);
// 回调完后通知，
// 从工作线程重回到IO线程
// connection_的指针地址将通过管道传给工作线程
connection_->notifyIOThread(); // ioThread_->notify(this);
}
};
TNonblockingIOThread::notifyHandler()
{
// 从管道中取出connection的指针地址
TNonblockingServer::TConnection* connection = NULL;
int nBytes = recv(fd, cast_sockopt(&connection), kSize, 0);
connection->transition(); // 进入状态转换函数
}
// 以下为工作线程
class ThreadManager::Impl : public ThreadManager;
class SimpleThreadManager : public ThreadManager::Impl;
class ThreadManager::Worker: public Runnable;
class ThreadManager::Task : public Runnable;
void SimpleThreadManager::start()
{
// workerCount_为工作线程数
addWorker(workerCount_);
}
void ThreadManager::Impl::addWorker(size_t value)
{
for (size_t ix = 0; ix < value; ix++)
{
worker = new ThreadManager::Worker(this);
// thread为PthreadThread
// 调用了worker->run();
thread->start();
}
}
void ThreadManager::Worker::run()
{
ThreadManager::Task task;
task->run();
}
class ThreadManager::Task: public Runnable
{
public:
void run()
{
// runnable_实际为TNonblockingServer::TConnection::Task
runnable_->run();
}
private:
// 这里的Runnable实际为TNonblockingServer::TConnection::Task
// 在TNonblockingServer::TConnection::transition()中被push进来
boost::shared_ptr runnable_;
};
void ThreadManager::Impl::add(shared_ptr value)
{
// std::queue > tasks_;
task = new ThreadManager::Task(value, expiration);
tasks_.push(task);
}
// 两者关系
class TNonblockingServer: public TServer
{
public:
TNonblockingServer(const boost::shared_ptr& threadManager);
private:
// TNonblockingServer关联了ThreadManager
boost::shared_ptr threadManager_;
};
// 工作线程将回调TNonblockingServer::TConnection::Task
class TNonblockingServer::TConnection::Task: public Runnable
{
};
// task为TNonblockingServer::TConnection::Task
void TNonblockingServer::addTask(boost::shared_ptr task)
{
// 将任务交给工作线程
// threadManager_为SimpleThreadManager
threadManager_->add(task, 0LL, taskExpireTime_);
}
void TNonblockingServer::TConnection::transition()
{
case APP_READ_REQUEST:
if (server_->isThreadPoolProcessing())
{
boost::shared_ptr task =
new TNonblockingServer::TConnection::Task(
processor_, inputProtocol_, outputProtocol_, this);
// server_为TNonblockingServer
// 回调交给工作线程，IO线程不做这个工作
server_->addTask(task); // server_为TNonblockingServer
}
else
{
// 调用TNonblockingServer的构造函数时，
// 如果没有指定参数ThreadManager，则会走这条分支
// 这种情况下，isThreadPoolProcessing()返回false
processor_->process(inputProtocol_, outputProtocol_, connectionContext_);
}
}
void TNonblockingServer::TConnection::Task::run()
{
// 回调
processor_->process(input_, output_, connectionContext_);
}
内嵌关系：
1) TNonblockingServer内嵌了类TConnection，而TConnection又内嵌了类Task
2) ThreadManager内嵌了类Impl、类Worker和类Task（注意区分于TConnection内嵌的Task），而Impl又是ThreadManager的子类，而Task是对Runnable的实现
class TNonblockingServer: public TServer
{
public:
void serve() // 用户可以直接调用server()，但直接调用run()是更好的做法
{
// 创建socket监听
// 创建TNonblockingIOThread
// 通过Thread启动TNonblockingIOThread
}
};
class TServer: public concurrency::Runnable
{
public:
virtual void serve() = 0;
virtual void run() // 用户也可以直接调用run()
{
serve();
}
};