MongoDB command命令处理模块源码实现二

//命令注册，所有注册的命令最终全部保存到_commands全局map表中  
//name和oldName实际上是同一个command，只是可能因为历史原因，命令名改名了  
Command::Command(StringData name, StringData oldName)  
  //命令名字符串  
  : _name(name.toString()),  
    //对应命令执行统计，total代表总的，failed代表执行失败的次数  
    _commandsExecutedMetric("commands." + _name + ".total", &_commandsExecuted),  
    _commandsFailedMetric("commands." + _name + ".failed", &_commandsFailed) {  
  //如果_commands map表还没有生成，则new一个  
  if (_commands == 0)  
      _commands = new CommandMap();  
  ......  
  //把name命令对应的command添加到map表中  
  Command*& c = (*_commands)[name];  
  if (c)  
      log() << "warning: 2 commands with name: " << _name;  
  c = this;  
  ......  
 
  //大部分命令name和oldName是一样的，所以在数组中只会记录一个  
  //如果改名过，则name和oldName就不一样，这时候都需要注册到map表，对应同一个command  
  if (!oldName.empty()) //也就是name和oldName两个命令对应的是同一个this类  
      (*_commands)[oldName.toString()] = this;  
}  

//insert命令初始化  
class CmdInsert : public WriteCommand { //  
public:  
  //insert命令初始化构造  
  CmdInsert() : WriteCommand("insert") {}  
  ......  
  //认证检查  
  Status checkAuthForRequest(...) final {  
      ......  
  }  
 
  //真正的Insert插入文档会走这里面  
  void runImpl(...);  
  }  
} cmdInsert; //直接定义一个cmdInsert全局变量  
 
//update命令初始化  
class CmdUpdate: public WriteCommand { //  
public:  
  //update命令初始化构造  
  CmdUpdate() : WriteCommand("update") {}  
  ......  
  //认证检查  
  Status checkAuthForRequest(...) final {  
      ......  
  }  
  //查询计划执行过程  
  Status explain(...) const override {  
        ......  
  }  
  //真正的update插入文档会走这里面  
  void runImpl(...);  
  }  
} cmdUpdate; //直接定义一个cmdUpdate全局变量  
 
//delete命令初始化  
class CmdDelete: public WriteCommand { //  
public:  
  //delete命令初始化构造  
  CmdDelete() : WriteCommand("delete") {}  
  ......  
  //认证检查  
  Status checkAuthForRequest(...) final {  
      ......  
  }  
  //查询计划执行过程  
  Status explain(...) const override {  
        ......  
  }  
 
  //真正的delete插入文档会走这里面  
  void runImpl(...);  
  }  
} cmdDelete; //直接定义一个cmdDelete全局变量

//find命令实现类  
class FindCmd : public BasicCommand {  
public:  
  //初始化构造  
  FindCmd() : BasicCommand("find") {}  
  ......  
     
  //查询计划执行过程  
  Status explain(...) const override {  
        ......  
  }  
} findCmd; //直接定义一个findCmd全局变量

//执行计划相关的几个command注册过程，通过new实现  
MONGO_INITIALIZER_WITH_PREREQUISITES(SetupPlanCacheCommands, MONGO_NO_PREREQUISITES)  
(InitializerContext* context) {  
  //执行计划相关的几个命令注册  
  new PlanCacheListQueryShapes();  
  new PlanCacheClear();  
  new PlanCacheListPlans();  
  return Status::OK();  
}  
 
//test命令相关的几个command注册过程，也是通过new实现  
MONGO_INITIALIZER(RegisterEmptyCappedCmd)(InitializerContext* context) {  
  //必须使能testCommandsEnabled，该命令才有效  
  if (Command::testCommandsEnabled) {  
      new CapTrunc();  
      new CmdSleep();  
      new EmptyCapped();  
      new GodInsert();  
  }  
  return Status::OK();  
}

命令模块基类基础接口初始化实现  
ass Command : public CommandInterface {  
blic:  
//获取集合名collection  
static std::string parseNsFullyQualified(...);  
//获取DB.COLLECTION  
static NamespaceString parseNsCollectionRequired(...);  
//map表结构  
using CommandMap = StringMap<Command*>;  
......  
//获取命令名  
const std::string& getName() const final {  
    return _name;  
}  
......  
//应答保留填充字段长度  
std::size_t reserveBytesForReply() const override {  
    return 0u;  
}  
//该命令是否只能在admin库执行，默认不可以  
bool adminOnly() const override {  
    return false;  
}  
//该命令是否需要权限认证检查？默认不需要  
bool localHostOnlyIfNoAuth() override {  
    return false;  
}  
//该命令执行后是否进行command操作计数  
bool shouldAffectCommandCounter() const override {  
    return true;  
}  
//该命令是否需要认证  
bool requiresAuth() const override {  
    return true;  
}  
//help帮助信息  
void help(std::stringstream& help) const override;  
//执行计划信息  
Status explain(...) const override;  
//日志信息相关  
void redactForLogging(mutablebson::Document* cmdObj) override;  
BSONObj getRedactedCopyForLogging(const BSONObj& cmdObj) override;  
//该命令是否为maintenance模式，默认false  
bool maintenanceMode() const override {  
    return false;  
}  
//maintenance是否支持，默认支持  
bool maintenanceOk() const override {  
    return true;  
}  
//本地是否支持非本地ReadConcern,默认不支持  
bool supportsNonLocalReadConcern(...) const override {  
    return false;  
}  
//是否允许AfterClusterTime，默认允许  
bool allowsAfterClusterTime(const BSONObj& cmdObj) const override {  
    return true;  
}  
//3.6版本默认opCode=OP_MSG,所以对应逻辑操作op为LogicalOp::opCommand  
LogicalOp getLogicalOp() const override {  
    return LogicalOp::opCommand;  
}  
//例如find就是kRead，update delete insert就是kWrite，非读写操作就是kCommand  
ReadWriteType getReadWriteType() const override {  
    return ReadWriteType::kCommand;  
}  
//该命令执行成功统计  
void incrementCommandsExecuted() final {  
    _commandsExecuted.increment();  
}  

//该命令执行失败统计  
void incrementCommandsFailed() final {  
    _commandsFailed.increment();  
}  

//真正得命令运行  
bool publicRun(OperationContext* opCtx, const OpMsgRequest& request, BSONObjBuilder& result);  

//获取支持的所有命令信息 ListCommandsCmd获取所有支持的命令 db.listCommands()  
static const CommandMap& allCommands() {  
    return *_commands;  
}  

//没用  
static const CommandMap& allCommandsByBestName() {  
    return *_commandsByBestName;  
}  

//收到不支持命令的统计，例如mongo shell敲一个mongodb无法识别得命令，这里就会统计出来  
static Counter64 unknownCommands;  
//根据命令字符串名查找对应命令  
static Command* findCommand(StringData name);  
//执行结果  
static void appendCommandStatus(...);  
//是否启用了command test功能  
static bool testCommandsEnabled;  
//help帮助信息  
static bool isHelpRequest(const BSONElement& helpElem);  
  static const char kHelpFieldName[];  
  //认证检查，检查是否有执行该命令得权限  
  static Status checkAuthorization(Command* c,  
                                    OperationContext* opCtx,  
                                    const OpMsgRequest& request);  
  ......  
private:  
  //添加地方见Command::Command(    
  //所有的command都在_commands中保存  
  static CommandMap* _commands;  
  //暂时没用  
  static CommandMap* _commandsByBestName;  
  //执行对应命令run接口  
  virtual bool enhancedRun(OperationContext* opCtx,  
                            const OpMsgRequest& request,  
                            BSONObjBuilder& result) = 0;  
  //db.serverStatus().metrics.commands命令查看，本命令的执行统计,包括执行成功和执行失败的  
  Counter64 _commandsExecuted;  
  Counter64 _commandsFailed;  
  //命令名，如"find" "insert" "update" "createIndexes" "deleteIndexes"  
  const std::string _name;  
 
  //每个命令执行是否成功通过MetricTree管理起来，也就是db.serverStatus().metrics.commands统计信息  
  //通过MetricTree特殊二叉树管理起来  
  ServerStatusMetricField<Counter64> _commandsExecutedMetric;  
  ServerStatusMetricField<Counter64> _commandsFailedMetric;  
}

//该命令执行成功统计  
void incrementCommandsExecuted() final {  
  _commandsExecuted.increment();  
}  
 
//该命令执行失败统计  
void incrementCommandsFailed() final {  
  _commandsFailed.increment();  
}  

//命令入口  
void execCommandDatabase(...)  
{  
  ......  
  //该命令执行次数统计 db.serverStatus().metrics.commands可以获取统计信息  
  command->incrementCommandsExecuted();  
  ......  
  //真正的命令执行在这里面  
  retval = runCommandImpl(opCtx, command, request, replyBuilder, startOperationTime);  
 
  //该命令失败次数统计  
  if (!retval) {  
      command->incrementCommandsFailed();  
  }  
  ......  
}  

bool insertBatchAndHandleErrors(...) {  
  ......  
  //一次性一条一条插入，上面的固定集合是一次性插入  
  for (auto it = batch.begin(); it != batch.end(); ++it) {  
      //insert操作计数  
      globalOpCounters.gotInsert();  
  }  
  ......  
}

//mongod代理insert统计核心流程  
bool ClusterWriteCmd::enhancedRun(...) {  
  ......  
  if (_writeType == BatchedCommandRequest::BatchType_Insert) {  
      //insert计数  
      for (size_t i = 0; i < numAttempts; ++i) {  
          globalOpCounters.gotInsert();  
      }  
  }  
  ......  
}

  bool ClusterFindCmd::run(...) {  
  //find操作统计，也就是query统计  
  globalOpCounters.gotQuery();  
  ......  
}

bool FindCmd::run(...) {  
  //find操作统计，也就是query统计  
  globalOpCounters.gotQuery();  
  ......  
}

bool ClusterWriteCmd::enhancedRun(...) {  
  //update操作统计
  globalOpCounters.gotUpdate();  
  ......  
}

//mongos代理update统计核心流程  
bool ClusterWriteCmd::enhancedRun(...) {  
  ......  
  if (_writeType == BatchedCommandRequest::BatchType_Update) {  
      //insert计数  
  for (size_t i = 0; i < numAttempts; ++i) {  
        globalOpCounters.gotUpdate();  
      }  
  }  
  ......  
}

//mongos代理update统计核心流程  
bool ClusterWriteCmd::enhancedRun(...) {  
  ......  
  if (_writeType == BatchedCommandRequest::BatchType_Update) {  
      //insert计数  
  for (size_t i = 0; i < numAttempts; ++i) {  
        globalOpCounters.gotUpdate();  
      }  
  }  
  ......  
}

static SingleWriteResult performSingleDeleteOp(...) {  
  ......  
  //分片mongod实例delete操作统计  
  globalOpCounters.gotDelete();  
  ......  
}

//mongos代理getMore统计核心流程  
bool ClusterGetMoreCmd::enhancedRun(...) {  
  //代理getMore统计  
  globalOpCounters.gotGetMore();  
  ......  
}

//mongod分片存储节点getMore统计  
bool ClusterGetMoreCmd::enhancedRun(...) {  
  //存储节点mongod getMore统计  
  globalOpCounters.gotGetMore();  
  ......  
}

//mongos代理commands统计  
void execCommandDatabase(...) {  
  ......  
  if (command->shouldAffectCommandCounter()) {  
      OpCounters* opCounters = &globalOpCounters;  
      opCounters->gotCommand();  
  }  
  ......  
}

void execCommandDatabase(...) {  
  ......  
  //是否进行command统计  
  if (command->shouldAffectCommandCounter()) {  
      OpCounters* opCounters = &globalOpCounters;  
      //commands计数自增
      opCounters->gotCommand();  
  }  
  ......  
}  

class Command : {  
  ......  
  //该命令是否进行command操作计数，默认需要。如果不需要进行command统计，可在命令继承类中置为false  
  bool shouldAffectCommandCounter() const override {  
      return true;  
  }  
  ......  
}

DbResponse ServiceEntryPointMongod::handleRequest(...) {  
  ......  
  //记录开始时间  
  //获取当前操作对应curop  
  CurOp& currentOp = *CurOp::get(opCtx);  
 
  ......  
  //执行请求对应命令  
  runCommands(opCtx, m);  
  ......  
 
  //记录结束时间  
  currentOp.ensureStarted();  
  currentOp.done();  
  //获取开始和结束时间差，也就是命令执行时间  
  debug.executionTimeMicros = durationCount<Microseconds>  
                (currentOp.elapsedTimeExcludingPauses());  
  //记录超过阀值的慢日志到日志文件  
  if (shouldLogOpDebug || (shouldSample && debug.executionTimeMicros > logThresholdMs * 1000LL)) {  
      ......  
  //记录慢日志到日志文件  
      log() << debug.report(&c, currentOp, lockerInfo.stats);  
  }  
  //记录慢日志到system.profile集合  
  if (currentOp.shouldDBProfile(shouldSample)) {  
      ......  
      //记录慢日志到system.profile集合中  
      profile(opCtx, op);  
  }  
  ......  
}  

//获取操作类型  
Command::ReadWriteType CurOp::getReadWriteType() const {  
  ......  
  switch (_logicalOp) {  
  //getmore find归纳为读  
      case LogicalOp::opGetMore:  
      case LogicalOp::opQuery:  
          return Command::ReadWriteType::kRead;  
  //增删改统一归纳为写  
      case LogicalOp::opUpdate:  
      case LogicalOp::opInsert:  
      case LogicalOp::opDelete:  
          return Command::ReadWriteType::kWrite;  
  //增删改以外的归纳为command  
      default:  
          return Command::ReadWriteType::kCommand;  
  }  
}  

//Top::_incrementHistogram调用  
//操作和时延计数操作  
void OperationLatencyHistogram::increment(uint64_t latency, Command::ReadWriteType type) {  
  int bucket = _getBucket(latency);  
  switch (type) {  
      //读时延累加，读计数自增  
      case Command::ReadWriteType::kRead:  
          _incrementData(latency, bucket, &_reads);  
          break;  
      //写时延累加，写计数自增    
      case Command::ReadWriteType::kWrite:  
          _incrementData(latency, bucket, &_writes);  
          break;  
      //command时延累加，command计数自增    
      case Command::ReadWriteType::kCommand:  
          _incrementData(latency, bucket, &_commands);  
          break;  
      default:  
          MONGO_UNREACHABLE;  
  }  
}

DbResponse ServiceEntryPointMongod::handleRequest(...) {  
  ......  
  //记录开始时间  
  //获取当前操作对应curop  
  CurOp& currentOp = *CurOp::get(opCtx);  
 
  ......  
  //执行请求对应命令  
  runCommands(opCtx, m);  
  ......  
 
  //记录结束时间  
  currentOp.ensureStarted();  
  currentOp.done();  
  //获取开始和结束时间差，也就是命令执行时间  
  debug.executionTimeMicros = durationCount<Microseconds>  
                (currentOp.elapsedTimeExcludingPauses());  
  ......  
  //记录慢日志到system.profile集合  
  //mongod读写的时间延迟统计 db.serverStatus().opLatencies获取  
  Top::get(opCtx->getServiceContext())  
      .incrementGlobalLatencyStats(   //读写统计
          opCtx,  
            //时延            
          durationCount<Microseconds>(currentOp.elapsedTimeExcludingPauses()),  
          currentOp.getReadWriteType()); //读写类型
  ......  
}