如果你真的有认真阅读live555的源码,你会发现,无论是它的服务器还是客户端,最终都会走到一个env->taskScheduler().doEventLoop()函数内,看名字应该就可以看出来,这是个消息循环处理函数。实际上,无论是客户端还是服务器,我们可以通过watchVariable变量来控制服务器或客户端的正常退出,否则它们永远都不会退出。
void BasicTaskScheduler0::doEventLoop(char volatile* watchVariable) { // Repeatedly loop, handling readble sockets and timed events: while (1) { //将传入参数watchVariable 在外部置位非0,则循环结束 if (watchVariable != NULL && *watchVariable != 0) break; SingleStep(); } }
下面我们来看看void BasicTaskScheduler::SingleStep(unsigned maxDelayTime)函数里到底做了哪些工作!这里我先总结一下,如何再认真阅读一下他的源码:
总结为以下四步:
1为所有需要操作的socket执行select。
2找出第一个应执行的socket任务(handler)并执行之。
3找到第一个应响应的事件,并执行之。
4找到第一个应执行的延迟任务并执行之。
可见,每一步中只执行三个任务队列中的一项。下面详细分析函数SingleStep():
void BasicTaskScheduler::SingleStep(unsigned maxDelayTime) { fd_set readSet = fReadSet; // make a copy for this select() call fd_set writeSet = fWriteSet; // ditto fd_set exceptionSet = fExceptionSet; // ditto //防止等待时间过长,导致select出错 DelayInterval const& timeToDelay = fDelayQueue.timeToNextAlarm(); struct timeval tv_timeToDelay; tv_timeToDelay.tv_sec = timeToDelay.seconds(); tv_timeToDelay.tv_usec = timeToDelay.useconds(); // Very large "tv_sec" values cause select() to fail. // Don't make it any larger than 1 million seconds (11.5 days) const long MAX_TV_SEC = MILLION; if (tv_timeToDelay.tv_sec > MAX_TV_SEC) { tv_timeToDelay.tv_sec = MAX_TV_SEC; } // Also check our "maxDelayTime" parameter (if it's > 0): if (maxDelayTime > 0 && (tv_timeToDelay.tv_sec > (long)maxDelayTime/MILLION || (tv_timeToDelay.tv_sec == (long)maxDelayTime/MILLION && tv_timeToDelay.tv_usec > (long)maxDelayTime%MILLION))) { tv_timeToDelay.tv_sec = maxDelayTime/MILLION; tv_timeToDelay.tv_usec = maxDelayTime%MILLION; } int selectResult = select(fMaxNumSockets, &readSet, &writeSet, &exceptionSet, &tv_timeToDelay); //select出错 if (selectResult < 0) { #if defined(__WIN32__) || defined(_WIN32) int err = WSAGetLastError(); // For some unknown reason, select() in Windoze sometimes fails with WSAEINVAL if // it was called with no entries set in "readSet". If this happens, ignore it: if (err == WSAEINVAL && readSet.fd_count == 0) { err = EINTR; // To stop this from happening again, create a dummy socket: if (fDummySocketNum >= 0) closeSocket(fDummySocketNum); fDummySocketNum = socket(AF_INET, SOCK_DGRAM, 0); FD_SET((unsigned)fDummySocketNum, &fReadSet); } if (err != EINTR) { #else if (errno != EINTR && errno != EAGAIN) { #endif // Unexpected error - treat this as fatal: #if !defined(_WIN32_WCE) perror("BasicTaskScheduler::SingleStep(): select() fails"); // Because this failure is often "Bad file descriptor" - which is caused by an invalid socket number (i.e., a socket number // that had already been closed) being used in "select()" - we print out the sockets that were being used in "select()", // to assist in debugging: fprintf(stderr, "socket numbers used in the select() call:"); for (int i = 0; i < 10000; ++i) { if (FD_ISSET(i, &fReadSet) || FD_ISSET(i, &fWriteSet) || FD_ISSET(i, &fExceptionSet)) { fprintf(stderr, " %d(", i); if (FD_ISSET(i, &fReadSet)) fprintf(stderr, "r"); if (FD_ISSET(i, &fWriteSet)) fprintf(stderr, "w"); if (FD_ISSET(i, &fExceptionSet)) fprintf(stderr, "e"); fprintf(stderr, ")"); } } fprintf(stderr, "\n"); #endif internalError(); } } //找到上次执行成功的socket节点 // Call the handler function for one readable socket: HandlerIterator iter(*fHandlers); HandlerDescriptor* handler; // To ensure forward progress through the handlers, begin past the last // socket number that we handled: if (fLastHandledSocketNum >= 0) { while ((handler = iter.next()) != NULL) { if (handler->socketNum == fLastHandledSocketNum) break; } if (handler == NULL) { fLastHandledSocketNum = -1; iter.reset(); // start from the beginning instead } } //判断socket触发的事件是否和套接字属性一致,如果一致,则执行 while ((handler = iter.next()) != NULL) { int sock = handler->socketNum; // alias int resultConditionSet = 0; if (FD_ISSET(sock, &readSet) && FD_ISSET(sock, &fReadSet)/*sanity check*/) resultConditionSet |= SOCKET_READABLE; if (FD_ISSET(sock, &writeSet) && FD_ISSET(sock, &fWriteSet)/*sanity check*/) resultConditionSet |= SOCKET_WRITABLE; if (FD_ISSET(sock, &exceptionSet) && FD_ISSET(sock, &fExceptionSet)/*sanity check*/) resultConditionSet |= SOCKET_EXCEPTION; if ((resultConditionSet&handler->conditionSet) != 0 && handler->handlerProc != NULL) { fLastHandledSocketNum = sock; // Note: we set "fLastHandledSocketNum" before calling the handler, // in case the handler calls "doEventLoop()" reentrantly. (*handler->handlerProc)(handler->clientData, resultConditionSet); break; } } //没有找到可执行的套接字,则从头开始遍历 if (handler == NULL && fLastHandledSocketNum >= 0) { // We didn't call a handler, but we didn't get to check all of them, // so try again from the beginning: iter.reset(); while ((handler = iter.next()) != NULL) { int sock = handler->socketNum; // alias int resultConditionSet = 0; if (FD_ISSET(sock, &readSet) && FD_ISSET(sock, &fReadSet)/*sanity check*/) resultConditionSet |= SOCKET_READABLE; if (FD_ISSET(sock, &writeSet) && FD_ISSET(sock, &fWriteSet)/*sanity check*/) resultConditionSet |= SOCKET_WRITABLE; if (FD_ISSET(sock, &exceptionSet) && FD_ISSET(sock, &fExceptionSet)/*sanity check*/) resultConditionSet |= SOCKET_EXCEPTION; if ((resultConditionSet&handler->conditionSet) != 0 && handler->handlerProc != NULL) { fLastHandledSocketNum = sock; // Note: we set "fLastHandledSocketNum" before calling the handler, // in case the handler calls "doEventLoop()" reentrantly. (*handler->handlerProc)(handler->clientData, resultConditionSet); break; } } if (handler == NULL) fLastHandledSocketNum = -1;//because we didn't call a handler } // Also handle any newly-triggered event (Note that we do this *after* calling a socket handler, // in case the triggered event handler modifies The set of readable sockets.) if (fTriggersAwaitingHandling != 0) { if (fTriggersAwaitingHandling == fLastUsedTriggerMask) { // Common-case optimization for a single event trigger: fTriggersAwaitingHandling &=~ fLastUsedTriggerMask; if (fTriggeredEventHandlers[fLastUsedTriggerNum] != NULL) { (*fTriggeredEventHandlers[fLastUsedTriggerNum])(fTriggeredEventClientDatas[fLastUsedTriggerNum]); } } else { // Look for an event trigger that needs handling (making sure that we make forward progress through all possible triggers): unsigned i = fLastUsedTriggerNum; EventTriggerId mask = fLastUsedTriggerMask; do { i = (i+1)%MAX_NUM_EVENT_TRIGGERS; mask >>= 1; if (mask == 0) mask = 0x80000000; if ((fTriggersAwaitingHandling&mask) != 0) { fTriggersAwaitingHandling &=~ mask; if (fTriggeredEventHandlers[i] != NULL) { (*fTriggeredEventHandlers[i])(fTriggeredEventClientDatas[i]); } fLastUsedTriggerMask = mask; fLastUsedTriggerNum = i; break; } } while (i != fLastUsedTriggerNum); } } //执行DelayQueue事件 // Also handle any delayed event that may have come due. fDelayQueue.handleAlarm(); }