live555学习笔记4－计划任务(TaskScheduler)深入探讨

四　计划任务(TaskScheduler)深入探讨

我们且把三种任务命名为：socket handler,event handler,delay task。

这三种任务的特点是，前两个加入执行队列后会一直存在，而delay task在执行完一次后会立即弃掉。

socket handler保存在队列BasicTaskScheduler0::HandlerSet* fHandlers中;

event handler保存在数组BasicTaskScheduler0::TaskFunc * fTriggeredEventHandlers[MAX_NUM_EVENT_TRIGGERS]中;

delay task保存在队列BasicTaskScheduler0::DelayQueue fDelayQueue中。

下面看一下三种任务的执行函数的定义：

socket handler为

typedef void BackgroundHandlerProc(void* clientData, int mask);

event handler为

typedef void TaskFunc(void* clientData);

delay task 为

typedef void TaskFunc(void* clientData);//跟event handler一样。

再看一下向任务调度对象添加三种任务的函数的样子：

delay task为：

void setBackgroundHandling(int socketNum, int conditionSet　,BackgroundHandlerProc* handlerProc, void* clientData)

event handler为:

EventTriggerId createEventTrigger(TaskFunc* eventHandlerProc)

delay task为：

TaskToken scheduleDelayedTask(int64_t microseconds, TaskFunc* proc,void* clientData)

socket handler添加时为什么需要那些参数呢？socketNum是需要的，因为要select socket（socketNum即是socket()返回的那个socket对象）。conditionSet也是需要的，它用于表明socket在select时查看哪种装态，是可读？可写？还是出错？proc和clientData这两个参数就不必说了（真有不明白的吗？）。再看BackgroundHandlerProc的参数，socketNum不必解释，mask是什么呢？它正是对应着conditionSet，但它表明的是select之后的结果，比如一个socket可能需要检查其读/写状态，而当前只能读，不能写，那么mask中就只有表明读的位被设置。

event handler是被存在数组中。数组大小固定，是32项，用EventTriggerId来表示数组中的项，EventTriggerId是一个32位整数，因为数组是32项，所以用EventTriggerId中的第n位置１表明对应数组中的第n项。成员变量fTriggersAwaitingHandling也是EventTriggerId类型，它里面置1的那些位对应了数组中所有需要处理的项。这样做节省了内存和计算，但降低了可读性，呵呵，而且也不够灵活，只能支持32项或64项，其它数量不被支持。以下是函数体

[cpp]
view plaincopyprint?

EventTriggerId BasicTaskScheduler0::createEventTrigger( TaskFunc* eventHandlerProc)
{
unsigned i = fLastUsedTriggerNum;
EventTriggerId mask = fLastUsedTriggerMask;

//在数组中寻找一个未使用的项，把eventHandlerProc分配到这一项。

do {
i = (i + 1) % MAX_NUM_EVENT_TRIGGERS;
mask >>= 1;
if (mask == 0)
mask = 0x80000000;

if (fTriggeredEventHandlers[i] == NULL) {
// This trigger number is free; use it:

fTriggeredEventHandlers[i] = eventHandlerProc;
fTriggeredEventClientDatas[i] = NULL; // sanity


fLastUsedTriggerMask = mask;
fLastUsedTriggerNum = i;

return mask; //分配成功，返回值表面了第几项

}
} while (i != fLastUsedTriggerNum);//表明在数组中循环一圈


//数组中的所有项都被占用，返回表明失败。

// All available event triggers are allocated; return 0 instead:

return 0;
}

[cpp] 
view plaincopyprint?

void BasicTaskScheduler0::triggerEvent(EventTriggerId eventTriggerId,void* clientData)   
{  
// First, record the "clientData":
  
if (eventTriggerId == fLastUsedTriggerMask) {   
// common-case optimization:直接保存下clientData
  
fTriggeredEventClientDatas[fLastUsedTriggerNum] = clientData;  
} else {  
//从头到尾查找eventTriggerId对应的项，保存下clientData
  
EventTriggerId mask = 0x80000000;  
for (unsigned i = 0; i < MAX_NUM_EVENT_TRIGGERS; ++i) {  
if ((eventTriggerId & mask) != 0) {  
fTriggeredEventClientDatas[i] = clientData;  
  
fLastUsedTriggerMask = mask;  
fLastUsedTriggerNum = i;  
}  
mask >>= 1;  
}  
}  
  
// Then, note this event as being ready to be handled.
  
// (Note that because this function (unlike others in the library) 
  
// can be called from an external thread, we do this last, to
  
// reduce the risk of a race condition.)
  
//利用fTriggersAwaitingHandling以bit mask的方式记录需要响应的事件handler们。
  
fTriggersAwaitingHandling |= eventTriggerId;  
}  

void BasicTaskScheduler0::triggerEvent(EventTriggerId eventTriggerId,void* clientData) 
{
// First, record the "clientData":
if (eventTriggerId == fLastUsedTriggerMask) { 
// common-case optimization:直接保存下clientData
fTriggeredEventClientDatas[fLastUsedTriggerNum] = clientData;
} else {
//从头到尾查找eventTriggerId对应的项，保存下clientData
EventTriggerId mask = 0x80000000;
for (unsigned i = 0; i < MAX_NUM_EVENT_TRIGGERS; ++i) {
if ((eventTriggerId & mask) != 0) {
fTriggeredEventClientDatas[i] = clientData;

fLastUsedTriggerMask = mask;
fLastUsedTriggerNum = i;
}
mask >>= 1;
}
}

// Then, note this event as being ready to be handled.
// (Note that because this function (unlike others in the library) 
// can be called from an external thread, we do this last, to
// reduce the risk of a race condition.)
//利用fTriggersAwaitingHandling以bit mask的方式记录需要响应的事件handler们。
fTriggersAwaitingHandling |= eventTriggerId;
}

看，clientData被传入了，这表明clientData在每次触发事件时是可以变的。

此时再回去看SingleStep()是不是更明了了？

delay task添加时，需要传入task延迟等待的微秒（百万分之一秒）数(第一个参数)，这个弱智也可以理解吧？嘿嘿。分析一下介个函数：

[cpp]
view plaincopyprint?

TaskToken BasicTaskScheduler0::scheduleDelayedTask(int64_t microseconds,TaskFunc* proc, void* clientData)
{
if (microseconds < 0)
microseconds = 0;
//DelayInterval 是表示时间差的结构

DelayInterval timeToDelay((long) (microseconds / 1000000),(long) (microseconds % 1000000));
//创建delayQueue中的一项

AlarmHandler* alarmHandler = new AlarmHandler(proc, clientData,timeToDelay);
//加入DelayQueue

fDelayQueue.addEntry(alarmHandler);
//返回delay task的唯一标志

return (void*) (alarmHandler->token());
}

delay task的执行都在函数fDelayQueue.handleAlarm()中，handleAlarm()在类DelayQueue中实现。看一下handleAlarm():
void DelayQueue::handleAlarm()
{
//如果第一个任务的执行时间未到，则同步一下（重新计算各任务的等待时间）。

if (head()->fDeltaTimeRemaining != DELAY_ZERO)
synchronize();
//如果第一个任务的执行时间到了，则执行第一个，并把它从队列中删掉。

if (head()->fDeltaTimeRemaining == DELAY_ZERO) {
// This event is due to be handled:

DelayQueueEntry* toRemove = head();
removeEntry(toRemove); // do this first, in case handler accesses queue

//执行任务，执行完后会把这一项销毁。

toRemove->handleTimeout();
}
}

[cpp] 
view plaincopyprint?

void DelayQueue::addEntry(DelayQueueEntry* newEntry)   
{  
//重新计算各项的等待时间
  
synchronize();  
  
//取得第一项 
  
DelayQueueEntry* cur = head();  
//从头至尾循环中将新项与各项的等待时间进行比较
  
while (newEntry->fDeltaTimeRemaining >= cur->fDeltaTimeRemaining) {  
//如果新项等待时间长于当前项的等待时间，则减掉当前项的等待时间。
  
//也就是后面的等待时几只是与前面项等待时间的差，这样省掉了记录插入时的时间的变量。
  
newEntry->fDeltaTimeRemaining -= cur->fDeltaTimeRemaining;  
//下一项
  
cur = cur->fNext;  
}  
  
//循环完毕，cur就是找到的应插它前面的项，那就插它前面吧
  
cur->fDeltaTimeRemaining -= newEntry->fDeltaTimeRemaining;  
  
// Add "newEntry" to the queue, just before "cur":
  
newEntry->fNext = cur;  
newEntry->fPrev = cur->fPrev;  
cur->fPrev = newEntry->fPrev->fNext = newEntry;  
}  

void DelayQueue::addEntry(DelayQueueEntry* newEntry) 
{
//重新计算各项的等待时间
synchronize();

//取得第一项
DelayQueueEntry* cur = head();
//从头至尾循环中将新项与各项的等待时间进行比较
while (newEntry->fDeltaTimeRemaining >= cur->fDeltaTimeRemaining) {
//如果新项等待时间长于当前项的等待时间，则减掉当前项的等待时间。
//也就是后面的等待时几只是与前面项等待时间的差，这样省掉了记录插入时的时间的变量。
newEntry->fDeltaTimeRemaining -= cur->fDeltaTimeRemaining;
//下一项
cur = cur->fNext;
}

//循环完毕，cur就是找到的应插它前面的项，那就插它前面吧
cur->fDeltaTimeRemaining -= newEntry->fDeltaTimeRemaining;

// Add "newEntry" to the queue, just before "cur":
newEntry->fNext = cur;
newEntry->fPrev = cur->fPrev;
cur->fPrev = newEntry->fPrev->fNext = newEntry;
}

有个问题，while循环中为什么没有判断是否到达最后一下的代码呢？难道肯定能找到大于新项的等待时间的项吗？是的！第一个加入项的等待时间是无穷大的，而且这一项永远存在于队列中。