EasyDarwin的RTP和RTSP会话的超时机制

今天是2016.08.05(这篇wiki写于公司内网的wiki平台上,现转载于此。),星期五,周末双休,又能回家了,但是在今天上午发生了一件事情,还是让我很惊奇,这件事情就是RTP会话被超时结束了,但是设备(拉流)还是能建立RTSP会话,只不过拉取不到任何数据,从抓包来看,设备如果拉取不到数据,就会重新建立RTSP会话,具体可以看抓包日志和流媒体服务器的日志。

当推流端开始推送RTP包后,RTSP会话会调用到RTSPSession::HandleIncomingDataPacket(),在这个函数里调用了fRTPSession->RefreshTimeout(),于是我们可以看到fRTPSession里的fTimeoutTask任务,fTimeoutTask的超时时间是又配置文件里的部分的rtp_timeout字段的值,也就是120秒,如果120秒没有收到任何推送的RTP包,fTimeoutTask会触发超时(详情请见TimeoutTaskThread::Run()函数),也就是会给RTPSession这个任务发送一个Task::kTimeoutEvent事件。

1
2
3
<PREF NAME="rtsp_timeout" TYPE="UInt32" >0</PREF>
<PREF NAME="real_rtsp_timeout" TYPE="UInt32" >180</PREF>
<PREF NAME="rtp_timeout" TYPE="UInt32" >120</PREF>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
SInt64 TimeoutTaskThread::Run()
{
//ok, check for timeouts now. Go through the whole queue
OSMutexLocker locker(&fMutex);
SInt64 curTime = OS::Milliseconds();
SInt64 intervalMilli = kIntervalSeconds * 1000;//always default to 60 seconds but adjust to smallest interval > 0
SInt64 taskInterval = intervalMilli;
for (OSQueueIter iter(&fQueue); !iter.IsDone(); iter.Next())
{
TimeoutTask* theTimeoutTask = (TimeoutTask*)iter.GetCurrent()->GetEnclosingObject();

//if it's time to time this task out, signal it
if ((theTimeoutTask->fTimeoutAtThisTime > 0) && (curTime >= theTimeoutTask->fTimeoutAtThisTime))
{
#if TIMEOUT_DEBUGGING
qtss_printf("TimeoutTask %"_S32BITARG_" timed out. Curtime = %"_64BITARG_"d, timeout time = %"_64BITARG_"d\n",(SInt32)theTimeoutTask, curTime, theTimeoutTask->fTimeoutAtThisTime);
#endif
theTimeoutTask->fTask->Signal(Task::kTimeoutEvent);
}
else
{
taskInterval = theTimeoutTask->fTimeoutAtThisTime - curTime;
if ( (taskInterval > 0) && (theTimeoutTask->fTimeoutInMilSecs > 0) && (intervalMilli > taskInterval) )
intervalMilli = taskInterval + 1000; // set timeout to 1 second past this task's timeout
#if TIMEOUT_DEBUGGING
qtss_printf("TimeoutTask %"_S32BITARG_" not being timed out. Curtime = %"_64BITARG_"d. timeout time = %"_64BITARG_"d\n", (SInt32)theTimeoutTask, curTime, theTimeoutTask->fTimeoutAtThisTime);
#endif
}
}
(void)this->GetEvents();//we must clear the event mask!

OSThread::ThreadYield();

#if TIMEOUT_DEBUGGING
qtss_printf ("TimeoutTaskThread::Run interval seconds= %"_S32BITARG_"\n", (SInt32) intervalMilli/1000);
#endif

return intervalMilli;//don't delete me!
}

RTPSession收到kTimeoutEvent事件后会向其它模块发送QTSS_ClientSessionClosing_Role角色的事件,至此RTPSession会话任务结束掉。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
SInt64 RTPSession::Run()
{
//if we have been instructed to go away, then let's delete ourselves
if ((events & Task::kKillEvent) || (events & Task::kTimeoutEvent) || (fModuleDoingAsyncStuff))
{
if (!fModuleDoingAsyncStuff)
{
if (events & Task::kTimeoutEvent)
fClosingReason = qtssCliSesCloseTimeout;

//deletion is a bit complicated. For one thing, it must happen from within
//the Run function to ensure that we aren't getting events when we are deleting
//ourselves. We also need to make sure that we aren't getting RTSP requests
//(or, more accurately, that the stream object isn't being used by any other
//threads). We do this by first removing the session from the session map.

#if RTPSESSION_DEBUGGING
qtss_printf("RTPSession %"_S32BITARG_": about to be killed. Eventmask = %"_S32BITARG_"\n",(SInt32)this, (SInt32)events);
#endif
// We cannot block waiting to UnRegister, because we have to
// give the RTSPSessionTask a chance to release the RTPSession.
OSRefTable* sessionTable = QTSServerInterface::GetServer()->GetRTPSessionMap();
Assert(sessionTable != NULL);
if (!sessionTable->TryUnRegister(&fRTPMapElem))
{
this->Signal(Task::kKillEvent);// So that we get back to this place in the code
return kCantGetMutexIdleTime;
}

// The ClientSessionClosing role is allowed to do async stuff
fModuleState.curTask = this;
fModuleDoingAsyncStuff = true; // So that we know to jump back to the
fCurrentModule = 0; // right place in the code

// Set the reason parameter
theParams.clientSessionClosingParams.inReason = fClosingReason;

// If RTCP packets are being generated internally for this stream,
// Send a BYE now.
RTPStream** theStream = NULL;
UInt32 theLen = 0;

if (this->GetPlayFlags() & qtssPlayFlagsSendRTCP)
{
SInt64 byePacketTime = OS::Milliseconds();
for (int x = 0; this->GetValuePtr(qtssCliSesStreamObjects, x, (void**)&theStream, &theLen) == QTSS_NoErr; x++)
if (theStream && *theStream != NULL)
(*theStream)->SendRTCPSR(byePacketTime, true);
}
}

//at this point, we know no one is using this session, so invoke the
//session cleanup role. We don't need to grab the session mutex before
//invoking modules here, because the session is unregistered and
//therefore there's no way another thread could get involved anyway
UInt32 numModules = QTSServerInterface::GetNumModulesInRole(QTSSModule::kClientSessionClosingRole);
{
for (; fCurrentModule < numModules; fCurrentModule++)
{
fModuleState.eventRequested = false;
fModuleState.idleTime = 0;
QTSSModule* theModule = QTSServerInterface::GetModule(QTSSModule::kClientSessionClosingRole, fCurrentModule);
(void)theModule->CallDispatch(QTSS_ClientSessionClosing_Role, &theParams);
// If this module has requested an event, return and wait for the event to transpire
if (fModuleState.eventRequested)
return fModuleState.idleTime; // If the module has requested idle time...
}
}

return -1;//doing this will cause the destructor to get called.
}
}

RTSPSession的超时机制和RTPSession类似,RTSPSession的超时时间是配置文件的real_rtsp_timeout字段值,在kFilteringRequest状态时会去刷新fTimeoutTask的时间,如果超时,也会结束整个状态机的处理部分,同时调用fRTPSession的Teardown函数,结束RTPSession会话。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
SInt64 RTSPSession::Run()
{
...

//check for a timeout or a kill. If so, just consider the session dead
if ((events & Task::kTimeoutEvent) || (events & Task::kKillEvent))
{
fLiveSession = false;
}
...// 状态机的部分
//fObjectHolders--
if(!IsLiveSession()&& fObjectHolders > 0){

OSRefTable* theMap = QTSServerInterface::GetServer()->GetRTPSessionMap();
OSRef* theRef = theMap->Resolve(&fLastRTPSessionIDPtr);
if (theRef != NULL){
fRTPSession = (RTPSession*)theRef->GetObject();
if(fRTPSession) fRTPSession->Teardown();
theMap->Release(fRTPSession->GetRef());
fRTPSession = NULL;
}
}
// Make absolutely sure there are no resources being occupied by the session
// at this point.
this->CleanupRequest();
// Only delete if it is ok to delete!
if (fObjectHolders == 0)
return -1;
// If we are here because of a timeout, but we can't delete because someone
// is holding onto a reference to this session, just reschedule the timeout.
//
// At this point, however, the session is DEAD.
return 0;

于是从以上可知,RTP会话超时是在120秒内没有收到任何RTP包就结束这个会话,此时RTSP会话还是存在的,也就是拉流端还是能够来请求拉流,鉴权也能通过,当RTSP会话在超过180秒内没有收到任何数据包则会结束整个RTSPSession。