Apple公司Darwin流式服务器源代码分析

Apple公司Darwin流式服务器源代码分析（尉明）当前，伴随着Internet的飞速发展，计算机网络已经进入到每一个普通人的家庭。在这个过程中，一个值得我们关注的现象是：Internet中存储和传输内容的构成已经发生了本质的改变，从传统的基于文本或少量图像的主页变为大容量、富信息量的流式媒体信息。一份早在1998年提交的研究报告就曾指出，流式媒体统治Internet的潮流是不可抗拒的，该报告估计到2003年，存储在网络服务器上的内容超过50%的将是流式媒体信息。但今天看来，这个估计还是有些保守了。所谓的流式媒体简单的讲就是指人们通过网络实时的收看多媒体信息：如音频流、视频流等。与流式媒体对应的传统工作方式是下载+播放模式，即用户首先下载多媒体文件，然后再在本地播放，这种方法的一个主要缺点是启动延迟较大，例如一个30分钟长的MPEG-I文件（相当于VCD质量），即使使用1.5Mbps的速率下载，也需要半个小时才能完成，这样一个漫长的等待时间实在是无法忍受。在窄带网络环境中，几乎所有基于Internet的流式媒体产品都有着类似的工作原理：首先需要开发高效的压缩编码技术，并通过一套完整有效的传输体系将其发布到用户的桌面上。目前在流式媒体领域，有三种占有主导地位的产品，它们分别是Apple公司的Quick Time、Microsoft公司的Media Server以及Real公司的Real System。本文将介绍QuickTime技术及其开放源代码的Darwin流化服务器。1QuickTime技术介绍Apple公司近日发布了QuickTime 5及QuickTime Streaming Server 3（简称QTSS）。作为客户端的QuickTime 5是用于在Internet上对高质量音频和视频内容进行创建、播放及提供数字流的软件，目前QuickTime在全世界的使用量已经超过1亿5千万份。QuickTime Streaming Server 3是Apple基于标准的、开放式源代码的流式服务器软件的新版本，它包括以下新功能：跳读保护（Skip Protection），一项获得专利的特性组合，它可以保证Internet上数字流的质量，防止中断；全新的易于使用、基于Web的界面，用户可以在本地或远程进行管理，实现服务器配置。作为Internet流媒体联盟(ISMA)的创建者之一，Apple不断致力于开弦到绫曜嫉牟泛图际酰岣呋僮餍岳从呕没氖褂锰逖椋壳癚uickTime已被国际标准组织(ISO)选为MPEG-4的基本文件格式，可预见Apple将有更多MPEG-4产品和技术的推出。QuickTime正迅速成为世界领先的跨平台多媒体技术，而且是迄今为止唯一的开放式源代码、基于标准的数字流解决方案。ZDNet在2000年9月对于三种流式媒体服务器的特征比较说明了QTSS不仅仅被技术开发者关注，而且可以通过简单的定制成为成熟强大的产品，评测结果可见表1。表1ZDNet对三类产品的评测结果服务器模块QTSS 2.01Media Server 7RealServer Basic 7操作系统支持Windows NT, 2000；FreeBSD；Linux；Mac OS；SolarisWindows NT, 2000Windows NT, 2000并发流个数2,0002,00025 free/3000 pro现场直播和广播YesYesYes在线广告支持YesYesYesPPV/流加密No / NoYes / YesYes / Yes分配流能力NoYesYesSMIL标准支持YesNoYesRTSP标准支持YesNoYes多播支持YesYesYes状态报告YesYesYes服务器日志YesYesYes防火墙和代理支持YesYesYes远程监控YesYesYes客户可以使用QuickTime Player或其他支持QuickTime的应用程序在Windows或Macintosh平台上接收视频流，而且QuickTime Player可以从苹果公司的网站上下载免费使用。如果安装了QuickTime的插件，客户还可以直接通过浏览器收看。客户希望点播一个节目时，QuickTime Player或插件将向QTSS发送请求，指明要点播的节目名。如果该节目存在，QTSS将向客户发送相应的视频流。当客户希望收看现场直播（或实时广播）时，它首先从QTSS获得关于当前频道的编码格式、地址等相关信息，然后再接受该频道的媒体流。对于那些希望在Internet上实时流化视频或音频信息的用户，QTSS服务器将是一个很好的选择，通过它可实现多项任务，例如：创建一个24小时在线的Internet广播电台；现场实况转播：如公司会议、体育节目等；创建远程学习站点：如能够点播视频和演讲；图1是一个利用QTSS服务器建立的现场直播场景。2Darwin流化服务器介绍Darwin Streaming Server（简称DSS）是QuickTime Streaming Server开放式源代码的版本，同时支持FreeBSD、Linux、Solaris、Windows NT和Windows 2000等多个操作系统，是当前所有同类产品中支持平台最多的一个。DSS的源代码和相关文档可从以下站点获得：DSS源代码完全采用标准C+语言写成，编程风格非常优秀，每个C+类都对应着一对和类同名的.h/.cpp文件。但是由于大量采用了面向对象的概念，如继承、多态等等；而且源文件和类相当多，所以不太容易讲清楚。因此，读者最好事先把代码完整的过滤一两遍，再配合本文，就能看得更清楚点。整个服务器包括多个子系统，分别存放在独立的工程内，如图2所示。其中，最为重要的是基础功能类库（CommonUtilitiesLib）和流化服务器（StreamingServer）两个工程，前者是整个系统的通用代码工具箱，包括了线程管理、数据结构、网络和文本分析等多个功能模块。DSS和其他相关的工具使用基础功能类库工程中定义的功能类实现以下三个目标：（1）抽象出系统中相同或类似的功能，用于降低代码的冗余度；（2）封装基本功能，简化高层编码的复杂度；（3）隔离开操作系统平台相关的代码。而流化服务器工程中包含了DSS对多个国际标准的实现，是整个服务器的主工程。在本文中，我们将重点分析这两个工程中的核心代码和模块。另外，我们还将简单介绍利用DSS提供的开发接口（Module）扩展和定制服务器的方法。DSS实现了四种IETF制定的国际标准，分别是：实时流传输协议RTSP（Real-time Streaming Protocol, RFC 2326）、实时传输协议（RTP Real-time Transfer Protocol，RFC 1889）、实时传输控制协议RTCP（Real-time Transport Control Protocol，RFC 1889）、会话描述协议SDP（Session Description Protocol，RFC 2327）。这四个标准是开发所有流式媒体产品都必须掌握的，因此在对相关代码进行分析和二次开发之前，希望读者了解上述四种协议的基本思想，上述协议样本可从以下网站获得：3基础功能类库（Common Utilities）3.1OS类Darwin Streaming Server支持包括Windows，Linux以及Solaris在内的多种操作系统平台。我们知道，Windows和Unix（或Unix-like）操作系统之间无论从内核还是编程接口上都有着本质的区别，即使是Linux和Solaris，在编程接口上也大为不同。为此，DSS开发了多个用于处理时间、临界区、信号量、事件、互斥量和线程等操作系统相关的类，这些类为上层提供了统一的使用接口，但在内部却需要针对不同的操作系统采用不同的方法实现。表2罗列出了DSS中的主要OS类和数据结构。表2DSS中的主要OS类和数据结构类（数据结构）名主要功能OS平台相关的功能类，如内存分配、时间等OSCond状态变量的基本功能和操作OSMutex互斥量的基本功能和操作OSThread线程类OSFileSource简单文件类OSQueue队列类OSHashTable哈希表类OSHeap堆类OSRef参考引用类3.1.1OSMutex/OSCond Class在有多个线程并发运行的环境中，能同步不同线程的活动是很重要的，DSS开发了OSMutex和OSCond两个类用以封装不同操作系统对线程同步支持的差异。我们首先分析OSMutex类，这个类定义了广义互斥量的基本操作，类定义如下：class OSMutex1public:2OSMutex()；/构造函数3OSMutex()；/析构函数4inline void Lock()；/加锁5inline void Unlock()；/解锁6inline Bool16 TryLock()；/异步锁，无论是否成功立即返回7private:8#ifdef _Win32_9CRITICAL_SECTION fMutex；/临界区10DWORDfHolder；/拥有临界区的线程id11UInt32fHolderCount；/进入临界区线程数/其他略在Windows平台上，OSMutex类是通过临界区（CRITICAL_SECTION）来实现的，第10行定义了临界区变量fMutex。类实例化时构造函数调用InitializeCriticalSection(&fMutex)初始化临界区变量，对应的在析构函数中调用DeleteCriticalSection(&fMutex)清除。Lock()函数用于对互斥量加锁，它调用私有方法RecursiveLock实现：voidOSMutex:RecursiveLock()/当前线程已经拥有互斥量，只需增加引用计数1if (OSThread:GetCurrentThreadID() = fHolder)23fHolderCount+；/增加引用计数4return；56#ifdef _Win32_7:EnterCriticalSection(&fMutex)；/申请进入临界区8#else9(void)pthread_mutex_lock(&fMutex)；10#endif11Assert(fHolder = 0)；12fHolder = OSThread:GetCurrentThreadID()；/更新临界区拥有者标志13fHolderCount+；14Assert(fHolderCount = 1)；第1行检测如果当前线程已经拥有互斥量，就只需将内部计数fHolderCount加1，以便纪录正在使用互斥量的方法数。如果当前线程还没有得到互斥量，第7行调用EnterCriticalSection()函数申请进入临界区；如果当前已经有其他线程进入临界区，该函数就会阻塞，使得当前线程进入睡眠状态，直到占用临界区的线程调用LeaveCriticalSection(&fMutex)离开临界区后才可能被唤醒。一旦线程进入临界区后，它将首先更新临界区持有者标志（第12行），同时将临界区引用计数加1。注意到另外一个函数TryLock()，该函数也是用于为互斥量加锁，但与Lock()不同的是，TryLock()函数为用户提供了异步调用互斥量的功能，这是因为它调用:TryEnterCriticalSection(&fMutex)函数申请进入缓冲区：如果临界区没有被任何线程拥有，该函数将临界区的访问区给予调用的线程，并返回TRUE，否则它将立刻返回FALSE。TryEnterCriticalSection()和EnterCriticalSection()函数的本质区别在于前者从不挂起线程。接着分析OSCond类，该类定义了状态变量（Condition Variable）的基本操作，类定义如下：class OSCond1public:2OSCond()；/构造函数3OSCond()；/析构函数4inline voidSignal()；/传信函数5inline voidWait(OSMutex* inMutex, SInt32 inTimeoutInMilSecs = 0)；/等待传信函数6inline voidBroadcast()；/广播传信函数7private:8#ifdef _Win32_9HANDLEfCondition；/事件句柄10UInt32fWaitCount；/等待传信用户数/其他略虽然同是用于线程同步，但OSCond类与OSMutex大不相同，后者用来控制对关键数据的访问，而前者则通过发信号表示某一操作已经完成。在Windows平台中，OSCond是通过事件（event）来实现的；构造函数调用CreateEvent()函数初始化事件句柄fCondition，而析构函数则调用CloseHandle()关闭句柄。OSCond的使用流程是这样的：线程调用Wait(OSMutex* inMutex, SInt32 inTimeoutInMilSecs = 0)函数等待某个事件的发生，其中inTimeoutInMilSecs是最长等待时间，0代表无限长。Wait()函数内部调用了WaitForSingleObject (fCondition, theTimeout)函数，该函数告诉系统线程在等待由事件句柄fCondition标识的内核对象变为有信号，参数theTimeout告诉系统线程最长愿意等待多少毫秒。如果指定的内核对象在规定时间内没有变为有信号，系统就会唤醒该线程，让它继续执行。而函数Signal()正是用来使事件句柄fCondition有信号的。Signal()函数内部实现很简单，只是简单调用SetEvent函数将事件句柄设置为有信号状态。使用OSCond的过程中存在一种需求，就是希望通知所有正在等待的用户事件已经完成，而Signal()函数每次只能通知一个用户，因此又开发了另外一个广播传信函数如下：inline void OSCond:Broadcast()/提示：本函数相当循环调用Signal()函数1#ifdef _Win32_2UInt32 waitCount = fWaitCount；/等待传信的用户数3for (UInt32 x = 0；x fRunning = true），当Entry()函数运行完后再设为非运行状态；在运行过程中，用户可以通过StopAndWaitForThread()、join()、Detach()以及ThrowStopRequest()等函数改变线程其他状态变量。3.1.3OSHashTable/OSQueue/OSHeap/OSRef ClassDSS定义了几个通用的较为复杂的数据结构，它们都以类的方式封装。这些数据结构不但贯穿于DSS的所有源代码，而且由于其封装的十分好，读者可以在看懂源代码的基础上很容易的将它们从DSS的工程中抽取出来，构建自己的基础类库，为将来的开发工作打下良好的基础。另外，对这些基础数据结构源代码的研究将提高我们对于面向对象技术的掌握和领会。最主要的数据结构有四种：哈希表（OSHashTable）、队列（OSQueue）、堆（OSHeap）和对象引用表（OSRef）。前三种是我们在编程中大量使用的数据结构，而对象引用表则是类似于COM/DCOM组件编程中IUNKOWN接口功能的数据结构，它首先为每个对象建立了一个字符串形式的ID，以便于通过这个ID找到对象（类似于QueryInterface）；另外OSRef类还为每个对象实例建立了引用计数，只有一个对象不再被任何人引用，才可能被释放（类似于AddRef和Release）。鉴于这几个类在结构上有相似之处，下面我们将分析OSHashTable的源代码，以便能够帮助读者更好的理解其他几个类。OSHashTable的代码如下：templateclass OSHashTable /*提示：OSHashTable被设计成为一个类模版，两个输入参数分别为：class T：实际的对象类；class K：用于为class T计算哈希表键值的功能类。*/1public:2OSHashTable( UInt32 size )/构造函数，入参是哈希表中对象的最大个数34fHashTable = new ( T*size )；/申请分配size个哈希对象class T的空间5Assert( fHashTable )；6memset( fHashTable, 0, sizeof(T*) * size )；/初始化7fSize = size；/*下面的代码决定用哪种方式为哈希表的键值计算索引；如果哈希表的大小不是2的幂，只好采用对fSize求余的方法；否则可以直接用掩码的方式，这种方式相对速度更快*/8fMask = fSize - 1；9if (fMask & fSize) != 0)/fSize不是2的幂10fMask = 0；11fNumEntries = 0；/当前对象数1213OSHashTable()/析构函数1415delete fHashTable；/释放空间16/下面介绍向哈希表中添加一个class T对象的源代码17void Add( T* entry ) 18Assert( entry-fNextHashEntry = NULL )；/*利用功能类class K，计算class T对象的哈希键值，其计算方法由用户在class K中定义*/19K key( entry )；20UInt32 theIndex = ComputeIndex( key.GetHashKey() )；/利用键值计算索引21entry-fNextHashEntry = fHashTable theIndex ；/在新加对象中存储索引值22fHashTable theIndex = entry；/将该对象插入到索引指定的位置23fNumEntries+；/24/下面介绍从哈希表中删除一个class T对象的源代码25void Remove( T* entry )26/首先从哈希表中找到待删除的对象/1、计算哈希键值和其对应的对象索引27key( entry )；28UInt32 theIndex = ComputeIndex( key.GetHashKey() )；29T* elem = fHashTable theIndex ；30T* last = NULL；/*2、通过对象索引查找对象，如果不是要找的对象，接着找下一个，直到找到为止。这是因为，存放的时候就是按照这种模式计算索引的。*/31while (elem & elem != entry) 32last = elem；33elem = elem-fNextHashEntry；34/找到该对象，将其删除35if ( elem )3637Assert(elem)；38if (last)39last-fNextHashEntry = elem-fNextHashEntry；40else /elem在头部41fHashTable theIndex = elem-fNextHashEntry；42elem-fNextHashEntry = NULL；43fNumEntries-；4445/下面介绍从哈希表中查找一个class T对象的方法46T* Map( K* key )/入参为哈希键值4748UInt32 theIndex = ComputeIndex( key-GetHashKey() )；/计算索引49T* elem = fHashTable theIndex ；/找到索引对应的对象50while (elem) 51K elemKey( elem )；52if (elemKey =*key)/检查是否找对53break；54elem = elem-fNextHashEntry；/如果不是，继续找下一个5556return elem；57/以下略以上介绍了哈希表的构造以及三种基本操作：添加、删除和查询。另外，DSS还定义了OSHashTableIter类用于枚举OSHashTable中的class T对象；其中主要的操作有First和Next等，限于篇幅，此处就不再详述。3.2Tasks类因为服务器从整体上采用了异步的运行模式，这就需要一种用于事件通信的机制。举例来说：一个RTSP连接对应的Socket端口监测到网络上有数据到达，此时必须有一个模块（或代码）被通知（notify）去处理这些数据。为此，DSS定义了Task及其相关类作为实现这一通信机制的核心。在Task.h/cpp文件中，定义了三个主要的类，分别是：任务线程池类（TaskThreadPool Class）、任务线程类（TaskThread Class）以及任务类（Task Class）。每个Task对象有两个主要的方法：Signal和Run。当服务器希望发送一个事件给某个Task对象时，就会调用Signal()方法；而Run()方法是在Task对象获得处理该事件的时间片后运行的，服务器中的大部分工作都是在不同Task对象的Run()函数中进行的。每个Task对象的目标就是利用很小的且不会阻塞的时间片完成服务器指定某个工作。任务线程类是上文介绍的OSThread类的一个子类，代表专门用于运行任务类的一个线程。在每个任务线程对象内部都有一个OSQueue_Blocking类型的任务队列，存储该线程需要执行的任务。后面的分析可以看到，服务器调用一个任务的Signal函数，实际上就是将该任务加入到某个任务线程类的任务队列中去。另外，为了统一管理这些任务线程，DSS还开发了任务线程池类，该类负责生成、删除以及维护内部的任务线程列表。图4描述了任务类的运行。下面我们首先分析TashThread类，该类的定义如下：class TaskThread : public OSThread/OSThread的子类/提示：所有的Task对象都将在TaskThread中运行1public:2TaskThread() :OSThread(), fTaskThreadPoolElem(this)/构造函数3virtualTaskThread() this-StopAndWaitForThread()； /析构函数4private:5virtual voidEntry()；/从OSThread重载的执行函数，仍然能够被子类重载6Task*WaitForTask()；/检测是否有该执行的任务7OSQueueElemfTaskThreadPoolElem；/对应的线程池对象8OSHeapfHeap；/纪录任务运行时间的堆，用于WaitForTask函数/*关键数据结构：任务队列；在Task的Signal函数中直接调用fTaskQueue对象的EnQueue函数将自己加入任务队列*/9OSQueue_BlockingfTaskQueue；/此处略作为OSThread的子类，TaskThread重载了Entry函数，一旦TaskThread的对象被实例化，便运行该函数。Entry()函数的主要任务就是调用WaitForTask()函数监测任务队列，如果发现新任务，就在规定时间内执行；否则，就被阻塞。下面我们简要分析Entry()函数的流程：void TaskThread:Entry()1Task* theTask = NULL；/空任务2while (true) /线程循环执行3/监测是否有需要执行的任务，如果有就返回该任务；否则阻塞；4theTask = this-WaitForTask()；5Assert(theTask != NULL)；6Bool16 doneProcessingEvent = false；/尚未处理事件7while (!doneProcessingEvent)89theTask-fUseThisThread = NULL；/对任务的调度独立于线程10SInt64 theTimeout = 0；/Task中Run函数的返回值，重要/核心部分：运行任务，根据返回值判断任务进度11if (theTask-fWriteLock)12/如果任务中有写锁，需要使用写互斥量，否则可能造成死锁13OSMutexWriteLocker mutexLocker(&TaskThreadPool:sMutexRW)；14theTimeout = theTask-Run()；/运行任务，得到返回值15theTask-fWriteLock = false；1617else18/使用读互斥量19OSMutexReadLocker mutexLocker(&TaskThreadPool:sMutexRW)；20theTimeout = theTask-Run()；/运行任务，得到返回值2122/监测Task中Run()函数的返回值，共有三种情况23/1、返回负数，表明任务已经完全结束24if (theTimeout fEvents)；34if (doneProcessingEvent)35theTask = NULL；36/3、返回正数，表明任务希望在等待theTimeout时间后再次执行37else38/*将该任务加入到Heap中，并且纪录它希望等待的时间。Entry()函数将通过waitfortask()函数进行检测，如果等待的时间到了，就再次运行该任务*/39theTask-fTimerHeapElem.SetValue(OS:Milliseconds() + theTimeout)；40fHeap.Insert(&theTask-fTimerHeapElem)；41(void)atomic_or(&theTask-fEvents, Task:kIdleEvent)；/设置Idle事件42doneProcessingEvent = true；43/此处略注意，如果Task的Run()函数返回值TimeOut为正数，意味着该任务是一个周期性的工作，例如发送数据的视频泵（pump），需要每隔一定时间就发出一定量的视频数据，直至整个节目结束。为此，在第3843行，将该任务加入到堆fHeap中去，并且标记该任务下次运行的时间为TimeOut毫秒之后。将来通过调用WaitForTask()函数就能检测到该任务是否到达规定的运行时间，WaitForTask()函数的代码如下：Task* TaskThread:WaitForTask()1while (true)2/得到当前时间，该函数为静态函数，定义见OS.h3SInt64 theCurrentTime = OS:Milliseconds()；/*如果堆中有任务，且任务已经到执行时间，返回该任务。PeekMin函数见OSHeap.h，窃听堆中第一个元素（但不取出）*/4if (fHeap.PeekMin() != NULL) & (fHeap.PeekMin()-GetValue() GetEnclosingObject()；/如果堆中有任务，但是尚未到执行时间，计算需要等待的时间6SInt32 theTimeout = 0；7if (fHeap.PeekMin() != NULL)/计算还需等待的时间8theTimeout = fHeap.PeekMin()-GetValue() - theCurrentTime；9Assert(theTimeout = 0)；/等待theTimeout时间后从堆中取出任务返回10OSQueueElem* theElem = fTaskQueue.DeQueueBlocking(this, theTimeout)；11if (theElem != NULL)12return (Task*)theElem-GetEnclosingObject()；13上文曾经提到，Task对象内有两个方法：Signal和Run。Run函数是一个虚函数，由Task的子类重载，它的用法我们在分析TaskThread的Entry()函数和WaitForTask()函数中已经讨论了。而另一个Signal()函数也十分重要：服务器通过调用该函数将Task加入TaskThread，并且执行Run()函数。Signal()函数的核心部分如下：void Task:Signal(EventFlags events)/ fUseThisThread用于指定该任务运行的任务线程1if (fUseThisThread != NULL)/存在指定任务