核心服务器通过创建四种类型的线程来完成自己的工作，具体如下：
- 服务器自己拥有的主线程（Main Thread）。这个线程负责检查服务器是否需要关闭，记录状态信息，或者打印统计信息。
- 空闲任务线程（Idle Task Thread）。空闲任务线程管理一个周期性的任务队列。该任务队列有两种类型：超时任务和套接口任务。
- 事件线程（Event Thread）。事件线程负责侦听套接口事件，比如收到RTSP请求和RTP数据包，然后把事件传递给任务线程。
- 一个或者多个任务（Task）线程。任务线程从事件线程中接收RTSP和RTP请求，然后把请求传递到恰当的服务器模块进行处理，把数据包发送给客户端。缺省情况下，核心服务器为每一个处理器创建一个任务线程。
EventThread
负责侦听Socket的事件，全局只有一个事件线程。在RunServer.cpp中创建并启动。
1、网络事件的请求
通过void EventContext::RequestEvent(int theMask)去请求对应的网络事件。
2、网络事件的响应
EventThread::Entry()为网络事件处理过程，主要是等待网络事件，然后获取对应的EventContext，并通过ProcessEvent把事件投递出去。
3、内部调用EventThread
//创建
static void Initialize() { sEventThread = new EventThread(); }
//运行
static void StartThread() { sEventThread->Start(); }
void EventThread::Entry()
{
struct eventreq theCurrentEvent;
::memset(&theCurrentEvent, '\0', sizeof(theCurrentEvent));
while (true)
{
int theErrno = EINTR;
while (theErrno == EINTR)
{
//1、等待网络事件
int theReturnValue = select_waitevent(&theCurrentEvent, NULL);
//Sort of a hack. In the POSIX version of the server, waitevent can return
//an actual POSIX errorcode.
if (theReturnValue >= 0)
theErrno = theReturnValue;
else
theErrno = OSThread::GetErrno();
}
AssertV(theErrno == 0, theErrno);
//ok, there's data waiting on this socket. Send a wakeup.
if (theCurrentEvent.er_data != NULL)
{
//The cookie in this event is an ObjectID. Resolve that objectID into
//a pointer.
//StrPtrLen idStr((char*)&theCurrentEvent.er_data, sizeof(theCurrentEvent.er_data));
StrPtrLen idStr((char*)&theCurrentEvent.er_data, sizeof(PointerSizedInt));
OSRef* ref = fRefTable.Resolve(&idStr);
if (ref != NULL)
{
//2、获取EventContext，并通过ProcessEvent把事件投递出去
EventContext* theContext = (EventContext*)ref->GetObject();
theContext->ProcessEvent(theCurrentEvent.er_eventbits);
fRefTable.Release(ref);
}
}
}
IdleTaskThread
负责侦听IdleTask的定时任务，全局只有一个事件线程。在RunServer.cpp中创建并启动。
1、IdleTask与IdleTaskThread的关联
调用IdleTask的接口SetIdleTimer，会把IdleTask任务加入到IdleTaskThread线程。
2、内部调用IdleTaskThread
//RunServer调用
IdleTask::Initialize();
//启动
void IdleTask::Initialize()
{
if (!sIdleThread)
{
//sIdleThread = new IdleTaskThread();
sIdleThread = std::shared_ptr<IdleTaskThread>(new IdleTaskThread(), [&](IdleTaskThread* idle) { delete idle; idle = nullptr; });
sIdleThread->Start();
}
}
//实现
void
IdleTaskThread::Entry()
{
OSMutexLocker locker(&fHeapMutex);
while (true)
{
//if there are no events to process, block.
if (fIdleHeap.CurrentHeapSize() == 0)
fHeapCond.Wait(&fHeapMutex);
SInt64 msec = OS::Milliseconds();
//pop elements out of the heap as long as their timeout time has arrived
while ((fIdleHeap.CurrentHeapSize() > 0) && (fIdleHeap.PeekMin()->GetValue() <= msec))
{
IdleTask* elem = (IdleTask*)fIdleHeap.ExtractMin()->GetEnclosingObject();
Assert(elem != nullptr);
elem->Signal(Task::kIdleEvent);
}
//we are done sending idle events. If there is a lowest tick count, then
//we need to sleep until that time.
if (fIdleHeap.CurrentHeapSize() > 0)
{
SInt64 timeoutTime = fIdleHeap.PeekMin()->GetValue();
//because sleep takes a 32 bit number
timeoutTime -= msec;
Assert(timeoutTime > 0);
UInt32 smallTime = (UInt32)timeoutTime;
fHeapCond.Wait(&fHeapMutex, smallTime);
}
}
}
Task线程池
1、线程池的创建
通过bool TaskThreadPool::AddThreads(UInt32 numToAdd)来创建TaskThread任务线程池。
2、任务的投递
通过执行Task的方法Signal，把任务投递到线程池
3、任务的执行
通过void TaskThread::Entry()执行Task的Run方法，所以继承Task的Run方法为任务的执行代码
总结
整个DSS的程序结构还是比较清晰，里面存在大量的类的封装，通常使用几个类一起实现一个功能，主要是使用友元类和静态变量来实现。EventThread线程和IdleTaskThread线程都只有一个，处理多媒体并发时，应该勉强够用的。毕竟一台服务器的多媒体并发量并不大，瓶颈在于流量。里面的有些代码不够清晰，如果需要深入理解，还是要花费一定的功夫。
