Mongodb源码分析--主程序入口main()

长平狐 发布于 2012/11/06 18:42
阅读 115
收藏 0
      作为这个系列的开篇,本人特此声明,因为本人技术功力有限,且对mongodb源码目前也在研究探索中,可能会对mongodb内部某些实现机制及原作者的意图领会不够精确,因此错误再所难免,希望大家批评指正。另外本文所使用的mongodb源码为1.8 rc1,同时如果有条件的话,大家可以安装vs2010,用C++来编译调试mongodb源码,以便通过运行过程中的数据和流程来验证自己的判断。
 
      VS2010 C++下编译调试MongoDB源码
     http://www.cnblogs.com/daizhj/archive/2011/03/07/1973764.html

      好了,开始今天的正文吧。

    为了理解mongodb整体的运行机制,首先我们需要对其主要运行流程有一个大概的理解,而主入口函数main无疑是最佳突破口。首先我们在VS2010中打开db.sln文件,并打开db.cpp文件,找到主入口函数(位于文件613行),如下:

int  main( int  argc,  char *  argv[]) {
    
static  StaticObserver staticObserver;
    getcurns 
=  ourgetns;

    po::options_description general_options(
" General options " ); // 常规选项
#if  defined(_WIN32)
    po::options_description windows_scm_options(
" Windows Service Control Manager options " ); // windows服务控制管理选项仅限windows平台
#endif
    po::options_description replication_options(
" Replication options " ); // Replication选项
    po::options_description ms_options( " Master/slave options " ); // 主从选项
    po::options_description rs_options( " Replica set options " ); // Replica设置选项
    po::options_description sharding_options( " Sharding options " ); // 数据分片选项
    po::options_description visible_options( " Allowed options " ); // 可见选项
    po::options_description hidden_options( " Hidden options " ); // 隐藏选项

    po::positional_options_description positional_options;

    general_options.add_options()
    (
" auth " " run with security " )
    (
" cpu " " periodically show cpu and iowait utilization " )
    (
" dbpath " , po::value < string > () ,  " directory for datafiles " )
    (
" diaglog " , po::value < int > (),  " 0=off 1=W 2=R 3=both 7=W+some reads " )
    (
" directoryperdb " " each database will be stored in a separate directory " )
    .....



    该方法的开头代码(上面)主要是绑定一个配置操作选项的说明,包括命令行模式下的参数说明,因为内容较长,这里就不做过多描述了,需要说明options_description的是这些内容被放到了boost库(一个C++开源库)的options_description对象中,其类型结构可以理解为key/value模式,主要用于记录一系列的选项描述(符)信息,以便于通过名称查询相应选项信息。同时mongodb将选项大致归为8类,如上所述。

    接下说看一下其初始化时命令行参数的操作,如下:
  
  if ( argc  ==   1  )
        cout 
<<  dbExecCommand  <<   "  --help for help and startup options "   <<  endl;
    {
        po::variables_map 
params ;

        
string  error_message  =  arg_error_check(argc, argv);
        
if  (error_message  !=   "" ) {
            cout 
<<  error_message  <<  endl  <<  endl;
            show_help_text(visible_options);
            
return   0 ;
        }

        
if  (  !  CmdLine::store( argc , argv , visible_options , hidden_options , positional_options ,  params  ) )
            
return   0 ;

     上面方法对main主函数参数argc,argv及上面的那些选项实例进行存储并以此绑定到params实例上,因为接下来会通过params来设置cmdLine对象(CmdLine类型),并最终以该对象做为最终在mongodb内部标记相应启动命令参数信息的对象。形如:
 
     if  ( params .count( " version " )) {
            cout 
<<  mongodVersion()  <<  endl;
            printGitVersion();
            
return   0 ;
        }
        
if  (  params .count(  " dbpath "  ) ) {
            dbpath 
=   params [ " dbpath " ]. as < string > ();
            
if  (  params .count(  " fork "  )  &&  dbpath[ 0 !=   ' / '  ) {
                
//  we need to change dbpath if we fork since we change
                
//  cwd to "/"
                
//  fork only exists on *nix
                
//  so '/' is safe
                dbpath  =  cmdLine.cwd  +   " / "   +  dbpath;
            }
        }
        
else  {
            dbpath 
=   " d:/data/db/ " ;//我在此处改了源码
        }
    
        
if  (  params .count( " directoryperdb " )) {
            directoryperdb 
=   true ;
        }
        
if  ( params .count( " cpu " )) {
            cmdLine.cpu 
=   true ;
        }
        ......

    当搜集到足够的启动信息(参数)后,mongodb开启执行下面两行代码:
    Module::configAll(  params  );

 
    上面用params来配置加载的模块信息,而就目前而言,mongodb中的模块有两个:其类模式和MMS模块,后者是当mongodb监视服务有效情况下,以后台线程方式(BackgroundJob)运行的程序,类定义如下:

     /* * Mongo Monitoring Service
        if enabled, this runs in the background ands pings mss
    
*/
    
class  MMS :  public  BackgroundJob , Module {
    ....
    }


      因为相关代码比较简单,这里就不多作说明了,如果大家感兴趣的话,以后会专门写一篇介绍Module,BackgroundJob的文章 。

     回到正文,模块实始化完成了,就会运行如下代码:

    dataFileSync.go();


    这里要说明的是dataFileSync类也派生自BackgroundJob类,而BackgroundJob的功能就是生成一个后台线程并执行相应任务。而当前dataFileSync的任务就是在一段时间后(cmdLine.syncdelay)将内存中的数据flash到磁盘上(因为mongodb使用mmap方式将数据先放入内存中),代码如下:
class  DataFileSync :  public  BackgroundJob {
  ......
  
void  run() {
            
if ( cmdLine.syncdelay  ==   0  )
                log() 
<<   " warning: --syncdelay 0 is not recommended and can have strange performance "   <<  endl;
            
else   if ( cmdLine.syncdelay  ==   1  )
                log() 
<<   " --syncdelay 1 "   <<  endl;
            
else   if ( cmdLine.syncdelay  !=   60  )
                log(
1 <<   " --syncdelay  "   <<  cmdLine.syncdelay  <<  endl;
            
int  time_flushing  =   0 ;
            
while  (  !  inShutdown() ) {
                flushDiagLog();
                
if  ( cmdLine.syncdelay  ==   0  ) {
                    
//  in case at some point we add an option to change at runtime
                    sleepsecs( 5 );
                    
continue ;
                }

                sleepmillis( (
long   long ) std::max( 0.0 , (cmdLine.syncdelay  *   1000 -  time_flushing) );

                
if  ( inShutdown() ) {
                    
//  occasional issue trying to flush during shutdown when sleep interrupted
                     break ;
                }

                Date_t start 
=  jsTime();
                
int  numFiles  =  MemoryMappedFile::flushAll(  true  ); // 使用系统提供的内存映射文件方法
                time_flushing  =  ( int ) (jsTime()  -  start);

                globalFlushCounters.flushed(time_flushing);

                log(
1 <<   " flushing mmap took  "   <<  time_flushing  <<   " ms  "   <<   "  for  "   <<  numFiles  <<   "  files "   <<  endl;
            }
        }
......


     main主函数完成上面方法后,就会启动侦听方法,开始侦听客户端的链接请求,如下:

     initAndListen(cmdLine.port, appsrvPath);


     该侦听方法会最终调用db.cpp (467行)的如下方法,我们来看一下该方法做了些什么:

      void  _initAndListen( int  listenPort,  const   char   * appserverLoc  =  NULL) {


     首先是初始化一个名称“initandlisten”线程用于侦听客户端传来的操作信息(可能有误):
 
    Client::initThread( " initandlisten " );

     接着判断当前系统是32或64位系统?并获取当前进程ID并输出进程ID及数据库路径,端口信息以及当前mongodb及系统信息(这些信息也就是我们在命令行下经常看到的启动mongodb信息)
 
    bool  is32bit  =   sizeof ( int * ==   4 ;

    {
#if  !defined(_WIN32)
            pid_t pid 
=  getpid();
#else
            DWORD pid
= GetCurrentProcessId();
#endif
        Nullstream
&  l  =  log();
        l 
<<   " MongoDB starting : pid= "   <<  pid  <<   "  port= "   <<  cmdLine.port  <<   "  dbpath= "   <<  dbpath;
        
if ( replSettings.master ) l  <<   "  master= "   <<  replSettings.master;
        
if ( replSettings.slave )  l  <<   "  slave= "   <<  ( int ) replSettings.slave;
        l 
<<  ( is32bit  ?   "  32 "  :  "  64 "  )  <<   " -bit  "   <<  endl;
    }
    DEV log() 
<<   " _DEBUG build (which is slower) "   <<  endl;
    show_warnings();
    log() 
<<  mongodVersion()  <<  endl;
    printGitVersion();
    printSysInfo();

    完成这一步之后,接下来mongodb就会对相应路径下的数据文件进行检查,如出现文件错误(文件不存在等):

    stringstream ss;
    ss 
<<   " dbpath ( "   <<  dbpath  <<   " ) does not exist " ;
    uassert( 
10296  ,  ss.str().c_str(), boost::filesystem::exists( dbpath ) );
    
    stringstream ss;
    ss 
<<   " repairpath ( "   <<  repairpath  <<   " ) does not exist " ;
    uassert( 
12590  ,  ss.str().c_str(), boost::filesystem::exists( repairpath ) );

        
   同时使用"路径锁"方式来移除指定路径下的临时文件夹信息,如下:
   acquirePathLock();
   remove_all( dbpath 
+   " /_tmp/ "  );


   接着,mongodb还会启动持久化功能,该功能貌似是1.7版本后引入到系统中的,主要用于解决因系统宕机时,内存中的数据未写入磁盘而造成的数据丢失。其机制主要是通过log方式定时将操作日志(如cud操作等)记录到db的journal文件夹下,这样当系统再次重启时从该文件夹下恢复丢失的(内存)数据。有关这部分内容我会专门写文章加以介绍。

     dur::startup();

     
if ( cmdLine.durOptions  &  CmdLine::DurRecoverOnly )
         
return


     注:其命令行枚举定义如下
      enum  {  //  bits to be ORed
            DurDumpJournal  =   1 ,    //  dump diagnostics on the journal during recovery
            DurScanOnly  =   2 ,       //  don't do any real work, just scan and dump if dump specified
            DurRecoverOnly  =   4 ,    //  terminate after recovery step
            DurParanoid  =   8 ,       //  paranoid mode enables extra checks
            DurAlwaysCommit  =   16    //  do a group commit every time the writelock is released
        };
     
int  durOptions;           //  --durOptions <n> for debugging



   完成这一步之后,系统还会初始化脚本引擎,因为mongodb支持脚本语法做为其操作数据库的语言,如下:
    if  ( scriptingEnabled ) {
            ScriptEngine::setup();
            globalScriptEngine
-> setCheckInterruptCallback( jsInterruptCallback );
            globalScriptEngine
-> setGetInterruptSpecCallback( jsGetInterruptSpecCallback );
   }


   当这些主要工作做完之后,最后系统会调用下面方法正式启动侦听操作:

 
   void  listen( int  port) {
        log() 
<<   " waiting for connections on port  "   <<  port  <<  endl;
        OurListener l(cmdLine.bind_ip, port);
        l.setAsTimeTracker();
        startReplication();
        
if  (  ! noHttpInterface )
            boost::thread web( boost::bind(
& webServerThread,  new  RestAdminAccess()  /*  takes ownership  */ ));

#if (TESTEXHAUST)
        boost::thread thr(testExhaust);
#endif
        l.initAndListen();
    }

    注意上面的OurListener类其initAndListen()方法位于message.cpp中,因为mongodb采用message相关类来封装c/s双在的数据和操作:

void  Listener::initAndListen() {
     checkTicketNumbers();
        vector
< SockAddr >  mine  =  ipToAddrs(_ip.c_str(), _port);
        vector
< int >  socks;
        SOCKET maxfd 
=   0 //  needed for select()

        
for  (vector < SockAddr > ::iterator it = mine.begin(), end = mine.end(); it  !=  end;  ++ it) {
            SockAddr
&  me  =   * it;

            SOCKET sock 
=  ::socket(me.getType(), SOCK_STREAM,  0 );
            
if  ( sock  ==  INVALID_SOCKET ) {
                log() 
<<   " ERROR: listen(): invalid socket?  "   <<  errnoWithDescription()  <<  endl;
            }

            
if  (me.getType()  ==  AF_UNIX) {
#if  !defined(_WIN32)
                
if  (unlink(me.getAddr().c_str())  ==   - 1 ) {
                    
int  x  =  errno;
                    
if  (x  !=  ENOENT) {
                        log() 
<<   " couldn't unlink socket file  "   <<  me  <<  errnoWithDescription(x)  <<   "  skipping "   <<  endl;
                        
continue ;
                    }
                }
#endif
            }
            
else   if  (me.getType()  ==  AF_INET6) {
                
//  IPv6 can also accept IPv4 connections as mapped addresses (::ffff:127.0.0.1)
                
//  That causes a conflict if we don't do set it to IPV6_ONLY
                 const   int  one  =   1 ;
                setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (
const   char * & one,  sizeof (one));
            }

            bindOptions( sock );

            
if  ( ::bind(sock, me.raw(), me.addressSize)  !=   0  ) {
                
int  x  =  errno;
                log() 
<<   " listen(): bind() failed  "   <<  errnoWithDescription(x)  <<   "  for socket:  "   <<  me.toString()  <<  endl;
                
if  ( x  ==  EADDRINUSE )
                    log() 
<<   "   addr already in use "   <<  endl;
                closesocket(sock);
                
return ;
            }

#if  !defined(_WIN32)
            
if  (me.getType()  ==  AF_UNIX) {
                
if  (chmod(me.getAddr().c_str(),  0777 ==   - 1 ) {
                    log() 
<<   " couldn't chmod socket file  "   <<  me  <<  errnoWithDescription()  <<  endl;
                }

                ListeningSockets::
get () -> addPath( me.getAddr() );
            }
#endif

            
if  ( ::listen(sock,  128 !=   0  ) {
                log() 
<<   " listen(): listen() failed  "   <<  errnoWithDescription()  <<  endl;
                closesocket(sock);
                
return ;
            }

            ListeningSockets::
get () -> add( sock );

            socks.push_back(sock);
            
if  (sock  >  maxfd)
                maxfd 
=  sock;
        }

        
static   long  connNumber  =   0 ;
        
struct  timeval maxSelectTime;
        
while  (  !  inShutdown() ) {
            fd_set fds[
1 ];
            FD_ZERO(fds);

            
for  (vector < int > ::iterator it = socks.begin(), end = socks.end(); it  !=  end;  ++ it) {
                FD_SET(
* it, fds);
            }

            maxSelectTime.tv_sec 
=   0 ;
            maxSelectTime.tv_usec 
=   10000 ;
            
const   int  ret  =  select(maxfd + 1 , fds, NULL, NULL,  & maxSelectTime);

            
if  (ret  ==   0 ) {
#if  defined(__linux__)
                _elapsedTime 
+=  (  10000   -  maxSelectTime.tv_usec )  /   1000 ;
#else
                _elapsedTime 
+=   10 ;
#endif
                
continue ;
            }
            _elapsedTime 
+=  ret;  //  assume 1ms to grab connection. very rough

            
if  (ret  <   0 ) {
                
int  x  =  errno;
#ifdef EINTR
                
if  ( x  ==  EINTR ) {
                    log() 
<<   " select() signal caught, continuing "   <<  endl;
                    
continue ;
                }
#endif
                
if  (  !  inShutdown() )
                    log() 
<<   " select() failure: ret= "   <<  ret  <<   "   "   <<  errnoWithDescription(x)  <<  endl;
                
return ;
            }

            
for  (vector < int > ::iterator it = socks.begin(), end = socks.end(); it  !=  end;  ++ it) {
                
if  ( !  (FD_ISSET( * it, fds)))
                    
continue ;

                SockAddr from;
                
int  s  =  accept( * it, from.raw(),  & from.addressSize);
                
if  ( s  <   0  ) {
                    
int  x  =  errno;  //  so no global issues
                     if  ( x  ==  ECONNABORTED  ||  x  ==  EBADF ) {
                        log() 
<<   " Listener on port  "   <<  _port  <<   "  aborted "   <<  endl;
                        
return ;
                    }
                    
if  ( x  ==   0   &&  inShutdown() ) {
                        
return ;    //  socket closed
                    }
                    
if ! inShutdown() )
                        log() 
<<   " Listener: accept() returns  "   <<  s  <<   "   "   <<  errnoWithDescription(x)  <<  endl;
                    
continue ;
                }
                
if  (from.getType()  !=  AF_UNIX)
                    disableNagle(s);
                
if  ( _logConnect  &&   !  cmdLine.quiet )
                    log() 
<<   " connection accepted from  "   <<  from.toString()  <<   "  # "   <<   ++ connNumber  <<  endl;
                accepted(s, from);
            }
        }
    }


    上面方法基本上就是一个无限循环( while ( ! inShutdown() ) )的侦听服务端,它调用操作系统的底层socket api接口,并将侦听到的结果使用accepted()方法进行接收。这里要注意的是因为最终我们使用的是OurListener进行的侦听,所以最终系统会调用OurListener所实现的虚(virtual)方法,如下:

class  OurListener :  public  Listener {
    
public :
        OurListener(
const   string   & ip,  int  p) : Listener(ip, p) { }
        
virtual   void  accepted(MessagingPort  * mp) {

            
if  (  !  connTicketHolder.tryAcquire() ) {
                log() 
<<   " connection refused because too many open connections:  "   <<  connTicketHolder.used()  <<   "  of  "   <<  connTicketHolder.outof()  <<  endl;
                
//  TODO: would be nice if we notified them...
                mp -> shutdown();
                delete mp;
                
return ;
            }

            
try  {
                boost::thread thr(boost::bind(
& connThread,mp));
            }
            
catch  ( boost::thread_resource_error &  ) {
                log() 
<<   " can't create new thread, closing connection "   <<  endl;
                mp
-> shutdown();
                delete mp;
            }
            
catch  ( ... ) {
                log() 
<<   " unkonwn exception starting connThread "   <<  endl;
                mp
-> shutdown();
                delete mp;
            }
        }
    };

    上面方法中的try{}语句中包含的是boost库中的thread方法,其主要提供了跨操作系统的线程创建方式及相关并行操作(相关信息参数boost官方网站),我们这里只要知道,通过该语句,我们最终用一个线程来运行connThread方法及其所需参数mp即可。下面看一下connThread方法的代码:

 
void  connThread( MessagingPort  *  inPort ) {
        TicketHolderReleaser connTicketReleaser( 
& connTicketHolder );

        
/*  todo: move to Client object  */
        LastError 
* le  =   new  LastError();
        lastError.reset(le);

        inPort
-> _logLevel  =   1 ;
        auto_ptr
< MessagingPort >  dbMsgPort( inPort );
        Client
&  c  =  Client::initThread( " conn " , inPort);

        
try  {

            c.getAuthenticationInfo()
-> isLocalHost  =  dbMsgPort -> farEnd.isLocalHost();

            Message m;
            
while  (  1  ) {
                inPort
-> clearCounters();

                
if  (  ! dbMsgPort -> recv(m) ) {
                    
if ! cmdLine.quiet )
                        log() 
<<   " end connection  "   <<  dbMsgPort -> farEnd.toString()  <<  endl;
                    dbMsgPort
-> shutdown();
                    
break ;
                }
sendmore:
                
if  ( inShutdown() ) {
                    log() 
<<   " got request after shutdown() "   <<  endl;
                    
break ;
                }

                lastError.startRequest( m , le );

                DbResponse dbresponse;
                assembleResponse( m, dbresponse, dbMsgPort
-> farEnd );

                
if  ( dbresponse.response ) {
                    dbMsgPort
-> reply(m,  * dbresponse.response, dbresponse.responseTo);
                    
if ( dbresponse.exhaust ) {
                        ...出现问题时
                    }
                }

                networkCounter.hit( inPort
-> getBytesIn() , inPort -> getBytesOut() );

                m.reset();
            }

        }
        ......

        
//  thread ending...
        {
            Client 
*  c  =  currentClient. get ();
            
if ( c ) c -> shutdown();
        }
        globalScriptEngine
-> threadDone();
    }

    上面代码主要工作就是不断循环[while ( 1 )]获取当前客户端发来的信息(上面已封装成了message)并将其信息进行分析,并根据相应操作标志位确定当前操作是CRUD或构建索引等[ assembleResponse()],如果一些正常,则向客户端发送应答信息:

void  connThread( MessagingPort  *  inPort ) {
        TicketHolderReleaser connTicketReleaser( 
& connTicketHolder );

        
/*  todo: move to Client object  */
        LastError 
* le  =   new  LastError();
        lastError.reset(le);

        inPort
-> _logLevel  =   1 ;
        auto_ptr
< MessagingPort >  dbMsgPort( inPort );
        Client
&  c  =  Client::initThread( " conn " , inPort);

        
try  {

            c.getAuthenticationInfo()
-> isLocalHost  =  dbMsgPort -> farEnd.isLocalHost();

            Message m;
            
while  (  1  ) {
                inPort
-> clearCounters();

                
if  (  ! dbMsgPort -> recv(m) ) {
                    
if ! cmdLine.quiet )
                        log() 
<<   " end connection  "   <<  dbMsgPort -> farEnd.toString()  <<  endl;
                    dbMsgPort
-> shutdown();
                    
break ;
                }
sendmore:
                
if  ( inShutdown() ) {
                    log() 
<<   " got request after shutdown() "   <<  endl;
                    
break ;
                }

                lastError.startRequest( m , le );

                DbResponse dbresponse;
                assembleResponse( m, dbresponse, dbMsgPort
-> farEnd );

                
if  ( dbresponse.response ) {
                    dbMsgPort
-> reply(m,  * dbresponse.response, dbresponse.responseTo);
                    
if ( dbresponse.exhaust ) {
                        ...出现问题时
                    }
                }

                networkCounter.hit( inPort
-> getBytesIn() , inPort -> getBytesOut() );

                m.reset();
            }

        }
        ......

        
//  thread ending...
        {
            Client 
*  c  =  currentClient. get ();
            
if ( c ) c -> shutdown();
        }
        globalScriptEngine
-> threadDone();
    }


  
    运行到这里,main函数的使命就完成了,本来想用一张时序图来大致回顾一下,只有等有时间再补充了。

    好了,今天的内容到这里就告一段落了,在接下来的文章中,将会介绍客户端发起查询操作时,Mongodb的执行流程和运行机制。

    原文链接: http://www.cnblogs.com/daizhj/archive/2011/03/17/1987311.html
    作者: daizhj, 代震军   
    微博:http://t.sina.com.cn/daizhj
   
Tags: mongodb,c++,source code
Tags: mongodb,c++,source code

原文链接:http://www.cnblogs.com/daizhj/archive/2011/03/17/1987311.html
加载中
返回顶部
顶部