crtmpserver框架代码详解

Hank (http://blog.chinaunix.net/uid/26000296.html)版权所有，尊重他人劳动成果，
转载时请注明作者和原始出处及本声明。

一、简介
crtmpserver（C++ RTMP Server）是高性能的流媒体服务器，支持以下协议（直播或录制）:
 . 支持Flash（RTMP,RTMPE,RTMPS,RTMPS,RTMPT,RTMPTE）
 . 支持嵌入式设备：iPhone,Android
 . 支持监控摄像机
 . 支持IP-TV（MPEG-tS,RTSP/RTCP/RTP）


此外，crtmpserver可以作为高性能rendes-vous服务器，可以让你做：
 . 音视频会议
 . 在线游戏
 . 在线协作
 . 简单/复杂的聊天应用


crtmpserver不同之处
 . 支持多种流媒体技术之间的通信（Adobe flash, Apple streaming, Silverlight, etc）
 . 高性能，并发几千路连接
 . 占用资源少
 . 可移植性强，只要GCC支持，crtmpserver可以运行在： IP cameras, Android, ARM or MIPS based systems, SoC, etc
 . 依赖少：lua, openssl


如此强大和高性能的流媒体服务器，是如何实现肯定很让人期待，
我前面写了一些详细的如何应用的文章，
本文开始分析代码，剖析其是如何实现，并高性能处理的；


NOTE1: 本文基于linux Redhat运行环境，使用其默认配置文件crtmpserver/crtmpserver.lua;

二、Crtmpserver的框架层
先从main()函数开始，具体的分析时，将会配上相应的数据结构解析；


2.1 main()函数层
源文件: crtmpserver/src/crtmpserver.cpp 
代码:
01  int main(int argc, char **argv) { 


      //1. Pick up the startup parameters and hold them inside the running status 
02    Variant::DeserializeFromCmdLineArgs(argc, argv, gRs.commandLine);


03    string configFile = argv[argc - 1];
04    if (configFile.find("--") == 0)
05      configFile = "";   
06    NormalizeCommandLine(configFile);  


07    do {
       //2. Reset the run flag
08     gRs.run = false;


        //3. Initialize the running status
09      if (Initialize()) {
10        Run();
11      } else {
12        gRs.run = false;
13      }


        //5. Cleanup
14      Cleanup();
15    } while (gRs.run);


      //6. We are done
16    return 0;
17  }


分析:
02 :  命令行参数检查
03~06 : 命令行参数的基本处理；
09 :  初始化整个流媒体服务器；
10 :  处理各种事件;
14 :  关闭服务器的各项功能，并清空；


最重要的就是Initialize()，Run()这两个函数，下面具体分析它们;


2.2 Initialize()函数层
源文件: crtmpserver/src/crtmpserver.cpp 
代码:
01  bool Initialize() {
02    Logger::Init();


03    gRs.pConfigFile = new ConfigFile(NULL, NULL);
04    string configFilePath = gRs.commandLine["arguments"]["configFile"];
05    string fileName;
06    string extension;
07    splitFileName(configFilePath, fileName, extension);
08    if (lowerCase(extension) == "xml") {
        ...
09    } else if (lowerCase(extension) == "lua") {
10      gRs.pConfigFile->LoadLuaFile(configFilePath,(bool)gRs.commandLine["arguments"]["--daemon"]);
11    } else {
12      FATAL("Invalid file format: %s", STR(configFilePath));
13      return false;
14    }


15    gRs.pConfigFile->ConfigLogAppenders();


16    INFO("Initialize I/O handlers manager: %s", NETWORK_REACTOR);
17    IOHandlerManager::Initialize();


18    INFO("Configure modules");
19    gRs.pConfigFile->ConfigModules();


20    INFO("Plug in the default protocol factory");
21    gRs.pProtocolFactory = new DefaultProtocolFactory();
22    ProtocolFactoryManager::RegisterProtocolFactory(gRs.pProtocolFactory);


23    INFO("Configure factories");
24    gRs.pConfigFile->ConfigFactories();


25    INFO("Configure acceptors");
26    gRs.pConfigFile->ConfigAcceptors();


27    INFO("Configure instances");
28    gRs.pConfigFile->ConfigInstances();


29    INFO("Start I/O handlers manager: %s", NETWORK_REACTOR);
30    IOHandlerManager::Start();


31    INFO("Configure applications");
32    gRs.pConfigFile->ConfigApplications();


33    INFO("Install the quit signal");
34    installQuitSignal(QuitSignalHandler);


35    return true;
36  }


分析:
03~14: 依据配置文件的后缀(本文使用的是crtmpserver.lua)选择不同的函数来处理配置文件;
Lua 是一个小巧的脚本语言。作者是巴西人。该语言的设计目的是为了嵌入应用程序中，从而为应用程序提供灵活的扩展和定制功能。
Lua脚本可以很容易的被 C/C++ 代码调用，也可以反过来调用C/C++的函数，这使得Lua在应用程序中可以被广泛应用。
不仅仅作为扩展脚本，也可以作为普通的配置文件，代替XML,Ini等文件格式，并且更容易理解和维护。
Lua由标准C编写而成，代码简洁优美，几乎在所有操作系统和平台上都可以编译，运行.


15: 配置流媒体服务器日志模块；


17: 初始化类IOHandlerManager的成员变量
IOHandlerManager类定义:
// FILE: thelib/include/netio/epoll/iohandlermanager.h
#define EPOLL_QUERY_SIZE 1024


class IOHandlerManager {
private:
  static int32_t _eq; 
  static map _activeIOHandlers;
  static map _deadIOHandlers;


  static struct epoll_event _query[EPOLL_QUERY_SIZE];


  static vector _tokensVector1;
  static vector _tokensVector2;
  static vector *_pAvailableTokens;
  static vector *_pRecycledTokens;


  static TimersManager *_pTimersManager;
  static struct epoll_event _dummy;
public:
  ...
}


IOHandlerManager::Initialize()的函数定义如下:
// FILE: thelib/src/netio/epoll/iohandlermanager.cpp


void IOHandlerManager::Initialize() {
  _eq = 0;
  _pAvailableTokens = &_tokensVector1;
  _pRecycledTokens = &_tokensVector2;
  _pTimersManager = new TimersManager(ProcessTimer);
  memset(&_dummy, 0, sizeof (_dummy));
}


19: 以dlopen的方式加载各个模块的动态库
ConfigFile类定义:
// FILE: thelib/include/configuration/configfile.h
class DLLEXP ConfigFile {
private:
  Variant _configuration;
  Variant _logAppenders;
  string _rootAppFolder;
  Variant _applications;
  map _uniqueNames;
  GetApplicationFunction_t _staticGetApplicationFunction;
  GetFactoryFunction_t _staticGetFactoryFunction;
  map _modules;
  bool _isOrigin;
public:
  ...
}


函数调用:
gRs.pConfigFile->ConfigModules()
 |--> ConfigFile::ConfigModules 
       |--> FOR_MAP(_applications, string, Variant, i) {
              ConfigFile::ConfigModule(Variant &node) 
            }   |
                |
                |--> Module::Load()
                       |--> Module::LoadLibrary(){
                              libHandler = LOAD_LIBRARY(STR(path), LOAD_LIBRARY_FLAGS);
                            }
                              
ConfigFile::_modules的数据:
["flvplayback"]   => flvplayback/libflvplayback.so
["samplefactory"] => samplefactory/libsamplefactory.so
["vptests"]       => vptests/libvptests.so
["admin"]         => admin/libadmin.so
["appselector"]   => appselector/libappselector.so
["proxypublish"]  => proxypublish/libproxypublish.so
["stresstest"]    => stresstest/libstresstest.so
["applestreamingclient"] => applestreamingclient/libapplestreamingclient.so


21: 创建类DefaultProtocolFactory的对象；
22: 初始化协议工厂类的对象的成员变量；
类定义如下:
/*!   
  @class ProtocolFactoryManager
  @brief Class that manages protocol factories.
 */
class DLLEXP ProtocolFactoryManager {
private:
  static map _factoriesById;
  static map _factoriesByProtocolId;
  static map _factoriesByChainName;
public:
  ...
}


代码如下：
bool ProtocolFactoryManager::RegisterProtocolFactory(BaseProtocolFactory *pFactory) {
  ...
  检查协议是否已注册；
  //1. Test to see if this factory is already registered
  //2. Test to see if the protocol chains exported by this factory are already in use
  //3. Test to see if the protocols exported by this factory are already in use


  //4. Register everything
  FOR_VECTOR(protocolChains, i) {
    _factoriesByChainName[protocolChains[i]] = pFactory;
  }


  FOR_VECTOR(protocols, i) {
    _factoriesByProtocolId[protocols[i]] = pFactory;
  }


  _factoriesById[pFactory->GetId()] = pFactory;
  
  return true;
}


24: 将加载了动态库的模块注册到协议工厂的成员变量中去；
成员函数定义如下:
bool ConfigFile::ConfigFactories() {


  FOR_MAP(_modules, string, Module, i) {
    if (!MAP_VAL(i).ConfigFactory()) {
      FATAL("Unable to configure factory");
      return false; 
    } 
  }     
  return true;
}


26: 区分是UDP还是TCP,对所有要处理协议对应的端口进行socket绑定;
    并初始化每个协议事件响应函数；
bool ConfigFile::ConfigAcceptors(){ 
  // 对所有的模块进行处理
  FOR_MAP(_modules, string, Module, i) {
       MAP_VAL(i).BindAcceptors();
         |-->bool Module::BindAcceptors()
               |--> bool Module::BindAcceptor(Variant &node)
                     |--> IOHandler::IOHandler
                           |--> void IOHandlerManager::RegisterIOHandler(IOHandler* pIOHandler)
                                 |--> void IOHandlerManager::SetupToken(IOHandler *pIOHandler)
                                       |--   pResult->pPayload = pIOHandler;
                                               pResult->validPayload = true;
  }
}


28: 创建server的实例
30: 调用函数 int epoll_create(int size); 创建epoll句柄；
void IOHandlerManager::Start() {
  _eq = epoll_create(EPOLL_QUERY_SIZE);
}


32: 初始化各种应用, 并调用 epoll_ctl()进行epoll事件响应函数的注册； 
bool ConfigFile::ConfigApplications()
  FOR_MAP(_modules, string, Module, i) {
          |--> bool Module::ConfigApplication()
                |--> bool BaseClientApplication::ActivateAcceptors
                       |--> bool BaseClientApplication::ActivateAcceptor(IOHandler *pIOHandler) 
                             |--> bool TCPAcceptor::StartAccept() 
                                    |--> bool IOHandlerManager::EnableAcceptConnections(IOHandler *pIOHandler) 
                                           evt.events = EPOLLIN;
                                           evt.data.ptr = pIOHandler->GetIOHandlerManagerToken();
                                           epoll_ctl(_eq, EPOLL_CTL_ADD, pIOHandler->GetInboundFd(), &evt) ;
        
  }
}


34: 依据不同的平台, 如linux ,windows, android等安装不同的信号处理机;
linux平台下：调用 sigemptyset() , sigaction(); 进行清空并注册自己的信号处理机；


2.3 Run()函数
函数定义:
01  void Run() {


02    INFO("\n%s", STR(gRs.pConfigFile->GetServicesInfo()));
03    INFO("GO! GO! GO! (%u)", (uint32_t) getpid());


04    while (IOHandlerManager::Pulse()) {
05      IOHandlerManager::DeleteDeadHandlers();
06      ProtocolManager::CleanupDeadProtocols();
07    }
08  }


代码分析:
02: 打印显示流媒体服务器所有开启的应用，以及对应的端口;
03: 显示流媒体服务器的进程ID；
04: 调用 epoll_wait()等待事件发生；
    当对应端口的事件发生时，就调用事件处理函数进行处理；


bool IOHandlerManager::Pulse() {
  int32_t eventsCount = 0;
  if ((eventsCount = epoll_wait(_eq, _query, EPOLL_QUERY_SIZE, 1000)) < 0) {
    int32_t err = errno;
    FATAL("Unable to execute epoll_wait: (%d) %s", err, strerror(err));
    return false;
  }
  
  _pTimersManager->TimeElapsed(time(NULL));
    
  for (int32_t i = 0; i < eventsCount; i++) {
    //1. Get the token
    IOHandlerManagerToken *pToken =
        (IOHandlerManagerToken *) _query[i].data.ptr;
  
    //2. Test the fd
    if ((_query[i].events & EPOLLERR) != 0) {
      if (pToken->validPayload) {
        if ((_query[i].events & EPOLLHUP) != 0) {
          DEBUG("***Event handler HUP: %p", (IOHandler *) pToken->pPayload);
          ((IOHandler *) pToken->pPayload)->OnEvent(_query[i]);
        } else {
          DEBUG("***Event handler ERR: %p", (IOHandler *) pToken->pPayload);
        }
        IOHandlerManager::EnqueueForDelete((IOHandler *) pToken->pPayload);
      }
      continue;
    }


    //3. Do the damage
    if (pToken->validPayload) {
      if (!((IOHandler *) pToken->pPayload)->OnEvent(_query[i])) {
        EnqueueForDelete((IOHandler *) pToken->pPayload);
      }
    } else {
      FATAL("Invalid token");
    }
  }


  if (_tokensVector1.size() > _tokensVector2.size()) {
    _pAvailableTokens = &_tokensVector1;
    _pRecycledTokens = &_tokensVector2;
  } else {
    _pAvailableTokens = &_tokensVector2;
    _pRecycledTokens = &_tokensVector1;
  }


  return true;
}

05,06: 事件处理完，对资源进行回收；