// whoever calling this function should NOT get latch
int pmdEDUMgr::_createNewEDU ( EDU_TYPES type, void* arg, EDUID *eduid)
{
   int rc               = EDB_OK ;
   unsigned int probe   = 0 ;
   pmdEDUCB *cb         = NULL ;
   EDUID myEDUID        = 0 ;
   if ( isQuiesced () )
   {
      rc = EDB_QUIESCED ;
      goto done ;
   }

   if ( !getEntryFuncByType ( type ) )
   {
      PD_LOG ( PDERROR, "The edu[type:%d] not exist or function is null", type ) ;
      rc = EDB_INVALIDARG ;
      probe = 30 ;
      goto error ;
   }

   cb = new(std::nothrow) pmdEDUCB ( this, type ) ;
   EDB_VALIDATE_GOTOERROR ( cb, EDB_OOM,
            "Out of memory to create agent control block" )
   // set to creating status
   cb->setStatus ( PMD_EDU_CREATING ) ;

   /***********CRITICAL SECTION*********************/
   _mutex.get () ;
   // if the EDU exist in runqueue
   if ( _runQueue.end() != _runQueue.find ( _EDUID )  )
   {
      _mutex.release () ;
      rc = EDB_SYS ;
      probe = 10 ;
      goto error ;
   }
   // if the EDU exist in idle queue
   if ( _idleQueue.end() != _idleQueue.find ( _EDUID )  )
   {
      _mutex.release () ;
      rc = EDB_SYS ;
      probe = 15 ;
      goto error ;
   }
   // assign EDU id and increment global EDUID
   cb->setID ( _EDUID ) ;
   if ( eduid )
      *eduid = _EDUID ;
   // place cb into runqueue
   _runQueue [ _EDUID ] = ( pmdEDUCB* ) cb ;
   myEDUID = _EDUID ;
   ++_EDUID ;
   _mutex.release () ;
   /***********END CRITICAL SECTION****************/

   // create a new thread here, pass agent CB and other arguments
   try
   {
      boost::thread agentThread ( pmdEDUEntryPoint,
                                  type, cb, arg ) ;
      // detach the agent so that he's all on his own
      // we only track based on CB
      agentThread.detach () ;
   }
   catch ( std::exception e )
   {
      // if we failed to create thread, make sure to clean runqueue
      _runQueue.erase ( myEDUID ) ;
      rc = EDB_SYS ;
      probe = 20 ;
      goto error ;
   }

   //The edu is create, need post a resum event
   cb->postEvent( pmdEDUEvent( PMD_EDU_EVENT_RESUME, false, arg ) ) ;
done :
   return rc ;
error :
   // clean out memory if it's allocated
   if ( cb )
      delete cb ;
   PD_LOG ( PDERROR, "Failed to create new agent, probe = %d", probe ) ;
   goto done ;
}

   //PD_TRACE_DECLARE_FUNCTION ( COORD_OPERATORDEL_EXE, "_coordDeleteOperator::execute" )
   INT32 _coordDeleteOperator::execute( MsgHeader *pMsg,
                                        pmdEDUCB *cb,
                                        INT64 &contextID,
                                        rtnContextBuf *buf )
   {
      INT32 rc = SDB_OK ;
      INT32 rcTmp = SDB_OK ;
      PD_TRACE_ENTRY ( COORD_OPERATORDEL_EXE ) ;

      coordSendOptions sendOpt( TRUE ) ;
      coordSendMsgIn inMsg( pMsg ) ;
      coordProcessResult result ;
      ROUTE_RC_MAP nokRC ;
      result._pNokRC = &nokRC ;

      coordCataSel cataSel ;
      MsgRouteID errNodeID ;

      BSONObj boDeletor ;

      MsgOpDelete *pDelMsg             = (MsgOpDelete *)pMsg ;
      INT32 oldFlag                    = pDelMsg->flags ;
      pDelMsg->flags                  |= FLG_DELETE_RETURNNUM ;
      contextID                        = -1 ;

      INT32 flag = 0;
      CHAR *pCollectionName = NULL ;
      CHAR *pDeletor = NULL ;
      CHAR *pHint = NULL ;
      rc = msgExtractDelete( (CHAR*)pMsg, &flag, &pCollectionName,
                             &pDeletor, &pHint ) ;
      if( rc )
      {
         PD_LOG( PDERROR,"Failed to parse delete request, rc: %d", rc ) ;
         pCollectionName = NULL ;
         goto error ;
      }

      try
      {
         boDeletor = BSONObj( pDeletor ) ;
      }
      catch ( std::exception &e )
      {
         PD_RC_CHECK( SDB_INVALIDARG, PDERROR,
                      "Delete failed, received unexpected error: %s",
                      e.what() ) ;
      }

      MON_SAVE_OP_DETAIL( cb->getMonAppCB(), pMsg->opCode,
                          "Collection:%s, Deletor:%s, Hint:%s, "
                          "Flag:0x%08x(%u)",
                          pCollectionName,
                          boDeletor.toString().c_str(),
                          BSONObj(pHint).toString().c_str(),
                          oldFlag, oldFlag ) ;

      rc = cataSel.bind( _pResource, pCollectionName, cb, FALSE, TRUE ) ;
      if ( rc )
      {
         PD_LOG( PDERROR, "Get or update collection[%s]'s catalog info "
                 "failed, rc: %d", pCollectionName, rc ) ;
         goto error ;
      }

   retry:
      pDelMsg->version = cataSel.getCataPtr()->getVersion() ;
      pDelMsg->w = 0 ;
      rcTmp = doOpOnCL( cataSel, boDeletor, inMsg, sendOpt, cb, result ) ;

      if ( SDB_OK == rcTmp && nokRC.empty() )
      {
         goto done ;
      }
      else if ( checkRetryForCLOpr( rcTmp, &nokRC, cataSel, inMsg.msg(),
                                    cb, rc, &errNodeID, TRUE ) )
      {
         nokRC.clear() ;
         _groupSession.getGroupCtrl()->incRetry() ;
         goto retry ;
      }
      else
      {
         PD_LOG( PDERROR, "Delete failed on node[%s], rc: %d",
                 routeID2String( errNodeID ).c_str(), rc ) ;
         goto error ;
      }

   done:
      if ( oldFlag & FLG_DELETE_RETURNNUM )
      {
         contextID = _recvNum ;
      }
      if ( pCollectionName )
      {
         PD_AUDIT_OP( AUDIT_DML, MSG_BS_DELETE_REQ, AUDIT_OBJ_CL,
                      pCollectionName, rc,
                      "DeletedNum:%u, Deletor:%s, Hint:%s, Flag:0x%08x(%u)",
                      _recvNum, boDeletor.toString().c_str(),
//.........这里部分代码省略.........

   // PD_TRACE_DECLARE_FUNCTION ( SDB_RTNQUERY, "rtnQuery" )
   INT32 rtnQuery ( const CHAR *pCollectionName,
                    const BSONObj &selector,
                    const BSONObj &matcher,
                    const BSONObj &orderBy,
                    const BSONObj &hint,
                    SINT32 flags,
                    pmdEDUCB *cb,
                    SINT64 numToSkip,
                    SINT64 numToReturn,
                    SDB_DMSCB *dmsCB,
                    SDB_RTNCB *rtnCB,
                    SINT64 &contextID,
                    rtnContextBase **ppContext,
                    BOOLEAN enablePrefetch )
   {
      INT32 rc = SDB_OK ;
      PD_TRACE_ENTRY ( SDB_RTNQUERY ) ;
      dmsStorageUnitID suID = DMS_INVALID_CS ;
      contextID             = -1 ;

      SDB_ASSERT ( pCollectionName, "collection name can't be NULL" ) ;
      SDB_ASSERT ( cb, "educb can't be NULL" ) ;
      SDB_ASSERT ( dmsCB, "dmsCB can't be NULL" ) ;
      SDB_ASSERT ( rtnCB, "rtnCB can't be NULL" ) ;

      dmsStorageUnit *su = NULL ;
      dmsMBContext *mbContext = NULL ;
      rtnContextData *dataContext = NULL ;
      const CHAR *pCollectionShortName = NULL ;
      rtnAccessPlanManager *apm = NULL ;
      optAccessPlan *plan = NULL ;

      BSONObj hintTmp = hint ;
      BSONObj blockObj ;
      BSONObj *pBlockObj = NULL ;
      const CHAR *indexName = NULL ;
      const CHAR *scanType  = NULL ;
      INT32 indexLID = DMS_INVALID_EXTENT ;
      INT32 direction = 0 ;

      if ( FLG_QUERY_EXPLAIN & flags )
      {
         rc = rtnExplain( pCollectionName,
                          selector,
                          matcher,
                          orderBy,
                          hint,
                          flags, numToSkip,
                          numToReturn,
                          cb, dmsCB, rtnCB,
                          contextID,
                          ppContext ) ;
         if ( SDB_OK != rc )
         {
            PD_LOG( PDERROR, "failed to explain query:%d", rc ) ;
            goto error ;
         }
         else
         {
            goto done ;
         }
      }

      rc = rtnResolveCollectionNameAndLock ( pCollectionName, dmsCB, &su,
                                             &pCollectionShortName, suID ) ;
      PD_RC_CHECK( rc, PDERROR, "Failed to resolve collection name %s",
                   pCollectionName ) ;

      rc = su->data()->getMBContext( &mbContext, pCollectionShortName, -1 ) ;
      PD_RC_CHECK( rc, PDERROR, "Failed to get dms mb context, rc: %d", rc ) ;

      rc = rtnCB->contextNew ( ( flags & FLG_QUERY_PARALLED ) ?
                               RTN_CONTEXT_PARADATA : RTN_CONTEXT_DATA,
                               (rtnContext**)&dataContext,
                               contextID, cb ) ;
      PD_RC_CHECK( rc, PDERROR, "Failed to create new data context" ) ;

      if ( Object == hint.getField( FIELD_NAME_META ).type() )
      {
         BSONObjBuilder build ;
         rc = _rtnParseQueryMeta( hint.getField( FIELD_NAME_META ).embeddedObject(),
                                  scanType, indexName, indexLID, direction,
                                  blockObj ) ;
         PD_RC_CHECK( rc, PDERROR, "Failed to parase query meta[%s], rc: %d",
                      hint.toString().c_str(), rc ) ;

         pBlockObj = &blockObj ;

         if ( indexName )
         {
            build.append( "", indexName ) ;
         }
         else
         {
            build.appendNull( "" ) ;
         }
         hintTmp = build.obj () ;
      }

//.........这里部分代码省略.........

   INT32 migWorker::_getBsonFromQueue( pmdEDUCB *eduCB, BSONObj &obj )
   {
      INT32 rc = SDB_OK ;
      PD_TRACE_ENTRY ( SDB__MIGWORKER__GETBSON );
      INT32 tempRc = SDB_OK ;
      UINT32 offset = 0 ;
      UINT32 size   = 0 ;
      UINT32 line   = 0 ;
      UINT32 column = 0 ;
      UINT32 startBlock = 0 ;
      UINT32 endBlock   = 0 ;
      //CHAR  *pJsonBuffer = NULL ;

      _master->popFromQueue ( eduCB,
                              offset, size,
                              line, column ) ;
      if ( 0 == offset && 0 == size &&
           0 == line && 0 == column )
      {
         rc = SDB_MIG_END_OF_QUEUE ;
         goto done ;
      }

      if ( MIG_PARSER_JSON == _master->_fileType )
      {
         tempRc = fromjson ( _master->getBuffer() + offset, obj ) ;
      }
      else if ( MIG_PARSER_CSV == _master->_fileType )
      {
         rc = _csvParser.csv2bson( _master->getBuffer() + offset,
                                   size, &obj ) ;
         if ( rc )
         {
            rc = SDB_UTIL_PARSE_JSON_INVALID ;
            PD_LOG ( PDERROR, "Failed to convert Bson, rc=%d", rc ) ;
            goto error ;
         }
      }
      else
      {
         rc = SDB_MIG_UNKNOW_FILE_TYPE ;
         PD_LOG ( PDERROR, "unknow file type" ) ;
         goto error ;
      }
      if ( tempRc )
      {
         //PD_LOG ( PDERROR, "Failed to json convert bson, json: %s , rc=%d",
         //         _pJsonBuffer, tempRc ) ;
         _master->sendMsgToClient ( "Error: error "
                                    "in json format, line %u, column %u",
                                    line, column ) ;
      }
      rc = _master->getBlockFromPointer ( offset, size,
                                          startBlock, endBlock ) ;
      if ( rc )
      {
         PD_LOG ( PDERROR, "Failed to get block from pointer, rc=%d", rc ) ;
         goto error ;
      }
      for ( UINT32 i = startBlock; i <= endBlock; ++i )
      {
         _master->bucketDec( i ) ;
      }
   done:
      if ( tempRc )
      {
         rc = tempRc ;
      }
      PD_TRACE_EXITRC ( SDB__MIGWORKER__GETBSON, rc );
      return rc ;
   error:
      goto done ;
   }

   INT32 _omTaskStrategyInfo::fromBSON( const BSONObj &obj )
   {
      INT32 rc = SDB_OK ;
      BSONElement beField ;
      BSONObj ipsObj ;

      beField = obj.getField( OM_REST_FIELD_RULE_ID ) ;
      if ( !beField.isNumber() )
      {
         PD_LOG( PDERROR, "Field[%s] must be number",
                 beField.toString( TRUE, TRUE ).c_str() ) ;
         rc = SDB_INVALIDARG ;
         goto error ;
      }
      setID( beField.numberLong() ) ;

      beField = obj.getField( OM_REST_FIELD_TASK_ID ) ;
      if ( !beField.isNumber() )
      {
         PD_LOG( PDERROR, "Field[%s] must be number",
                 beField.toString( TRUE, TRUE ).c_str() ) ;
         rc = SDB_INVALIDARG ;
         goto error ;
      }
      setTaskID( beField.numberLong() ) ;

      beField = obj.getField( OM_REST_FIELD_TASK_NAME ) ;
      if ( String != beField.type() )
      {
         PD_LOG( PDERROR, "Field[%s] must be string",
                 beField.toString( TRUE, TRUE ).c_str() ) ;
         rc = SDB_INVALIDARG ;
         goto error ;
      }
      setTaskName( beField.str() ) ;

      beField = obj.getField( OM_REST_FIELD_NICE ) ;
      if ( !beField.isNumber() )
      {
         PD_LOG( PDERROR, "Field[%s] must be number",
                 beField.toString( TRUE, TRUE ).c_str() ) ;
         rc = SDB_INVALIDARG ;
         goto error ;
      }
      setNice( beField.numberInt() ) ;

      beField = obj.getField( OM_REST_FIELD_USER_NAME ) ;
      if ( String != beField.type() )
      {
         PD_LOG( PDERROR, "Field[%s] must be string",
                 beField.toString( TRUE, TRUE ).c_str() ) ;
         rc = SDB_INVALIDARG ;
         goto error ;
      }
      setUserName( beField.str() ) ;

      beField = obj.getField( OM_REST_FIELD_IPS ) ;
      if ( Array != beField.type() )
      {
         PD_LOG( PDERROR, "Field[%s] must be string array",
                 beField.toString( TRUE, TRUE ).c_str() ) ;
         rc = SDB_INVALIDARG ;
         goto error ;
      }
      else
      {
         clearIPSet() ;

         string tmpStr ;
         BSONElement e ;
         BSONObjIterator itr( beField.embeddedObject() ) ;
         while( itr.more() )
         {
            e = itr.next() ;
            if ( String != beField.type() )
            {
               PD_LOG( PDERROR, "Field[%s] must be string array",
                       beField.toString( TRUE, TRUE ).c_str() ) ;
               rc = SDB_INVALIDARG ;
               goto error ;
            }
            tmpStr = e.str() ;

            if ( !tmpStr.empty() )
            {
               addIP( tmpStr ) ;
            }
         }
      }

   done:
      return rc ;
   error:
      goto done ;
   }

   INT32 utilWriteConfigFile( const CHAR * pFile, const CHAR * pData,
                              BOOLEAN createOnly )
   {
      INT32 rc = SDB_OK ;
      std::string tmpFile = pFile ;
      tmpFile += ".tmp" ;
      OSSFILE file ;
      BOOLEAN isOpen = FALSE ;
      BOOLEAN isBak = FALSE ;

      if ( SDB_OK == ossAccess( tmpFile.c_str() ) )
      {
         ossDelete( tmpFile.c_str() ) ;
      }

      if ( SDB_OK == ossAccess( pFile ) )
      {
         if ( createOnly )
         {
            rc = SDB_FE ;
            goto error ;
         }
         if ( SDB_OK == ossRenamePath( pFile, tmpFile.c_str() ) )
         {
            isBak = TRUE ;
         }
      }

      rc = ossOpen ( pFile, OSS_READWRITE|OSS_SHAREWRITE|OSS_REPLACE,
                     OSS_RWXU, file ) ;
      if ( rc )
      {
         goto error ;
      }
      isOpen = TRUE ;

      {
         SINT64 written = 0 ;
         SINT64 len = ossStrlen( pData ) ;
         while ( 0 < len )
         {
            SINT64 tmpWritten = 0 ;
            rc = ossWrite( &file, pData + written , len, &tmpWritten ) ;
            if ( rc && SDB_INTERRUPT != rc )
            {
               PD_LOG( PDERROR, "Failed to write file[%s]:%d", pFile, rc ) ;
               goto error ;
            }
            written += tmpWritten ;
            len -= tmpWritten ;
            rc = SDB_OK ;
         }
      }

      if ( SDB_OK == ossAccess( tmpFile.c_str() ) )
      {
         ossDelete( tmpFile.c_str() ) ;
      }

   done:
      if ( isOpen )
      {
         ossClose( file ) ;
      }
      return rc ;
   error:
      if ( isBak )
      {
         if ( isOpen )
         {
            ossClose( file ) ;
            isOpen = FALSE ;
            ossDelete( pFile ) ;
         }
         ossRenamePath( tmpFile.c_str(), pFile ) ;
      }
      goto done ;
   }

static INT32 _ossEnumFiles( const string &dirPath,
                            map<string, string> &mapFiles,
                            const CHAR *filter, UINT32 filterLen,
                            OSS_MATCH_TYPE type, UINT32 deep )
{
   INT32 rc = SDB_OK ;
   const CHAR *pFind = NULL ;

   try
   {
      fs::path dbDir ( dirPath ) ;
      fs::directory_iterator end_iter ;

      if ( 0 == deep )
      {
         goto done ;
      }

      if ( fs::exists ( dbDir ) && fs::is_directory ( dbDir ) )
      {
         for ( fs::directory_iterator dir_iter ( dbDir );
               dir_iter != end_iter; ++dir_iter )
         {
            try
            {
               if ( fs::is_regular_file ( dir_iter->status() ) )
               {
                  const std::string fileName =
                     dir_iter->path().filename().string() ;

                  if ( ( OSS_MATCH_NULL == type ) ||
                       ( OSS_MATCH_LEFT == type &&
                         0 == ossStrncmp( fileName.c_str(), filter,
                                          filterLen ) ) ||
                       ( OSS_MATCH_MID == type &&
                         ossStrstr( fileName.c_str(), filter ) ) ||
                       ( OSS_MATCH_RIGHT == type &&
                         ( pFind = ossStrstr( fileName.c_str(), filter ) ) &&
                         pFind[filterLen] == 0 ) ||
                       ( OSS_MATCH_ALL == type &&
                         0 == ossStrcmp( fileName.c_str(), filter ) )
                     )
                  {
                     mapFiles[ fileName ] = dir_iter->path().string() ;
                  }
               }
               else if ( fs::is_directory( dir_iter->path() ) && deep > 1 )
               {
                  _ossEnumFiles( dir_iter->path().string(), mapFiles,
                                 filter, filterLen, type, deep - 1 ) ;
               }
            }
            catch( std::exception &e )
            {
               PD_LOG( PDWARNING, "File or dir[%s] occur exception: %s",
                       dir_iter->path().string().c_str(),
                       e.what() ) ;
            }
         }
      }
      else
      {
         rc = SDB_INVALIDARG ;
         goto error ;
      }
   }
   catch( std::exception &e )
   {
      PD_LOG( PDERROR, "Occur exception: %s", e.what() ) ;
      rc = SDB_SYS ;
      goto error ;
   }

done:
   return rc ;
error:
   goto done ;
}

static INT32 _ossEnumSubDirs( const string &dirPath,
                              const string &parentSubDir,
                              vector< string > &subDirs,
                              UINT32 deep )
{
   INT32 rc = SDB_OK ;

   try
   {
      fs::path dbDir ( dirPath ) ;
      fs::directory_iterator end_iter ;

      string subDir ;

      if ( 0 == deep )
      {
         goto done ;
      }

      if ( fs::exists ( dbDir ) && fs::is_directory ( dbDir ) )
      {
         for ( fs::directory_iterator dir_iter ( dbDir );
               dir_iter != end_iter; ++dir_iter )
         {
            try
            {
               if ( fs::is_directory( dir_iter->path() ) )
               {
                  if ( parentSubDir.empty() )
                  {
                     subDir = dir_iter->path().leaf().string() ;
                  }
                  else
                  {
                     string subDir = parentSubDir ;
                     subDir += OSS_FILE_SEP ;
                     subDir += dir_iter->path().leaf().string() ;
                  }
                  subDirs.push_back( subDir ) ;

                  if ( deep > 1 )
                  {
                     _ossEnumSubDirs( dir_iter->path().string(), subDir,
                                      subDirs,deep - 1 ) ;
                  }
               }
            }
            catch( std::exception &e )
            {
               PD_LOG( PDWARNING, "File or dir[%s] occur exception: %s",
                       dir_iter->path().string().c_str(), e.what() ) ;
            }
         }
      }
      else
      {
         rc = SDB_INVALIDARG ;
         goto error ;
      }
   }
   catch( std::exception &e )
   {
      PD_LOG( PDERROR, "Occur exception: %s", e.what() ) ;
      rc = SDB_SYS ;
      goto error ;
   }

done:
   return rc ;
error:
   goto done ;
}

   INT32 _sptInvoker::_callbackDone( JSContext *cx, JSObject *obj,
                                     _sptReturnVal &rval,
                                     const bson::BSONObj &detail,
                                     jsval *rvp )
   {
      INT32 rc = SDB_OK ;
      const sptProperty &rpro = rval.getVal() ;
      jsval val = JSVAL_VOID ;

      if ( EOO == rpro.getType() )
      {
         *rvp = JSVAL_VOID ;
         goto done ;
      }
      else if ( Object == rpro.getType() )
      {
         JSObject *jsObj = JS_NewObject ( cx, (JSClass *)(rval.getClassDef()),
                                          0 , 0 ) ;
         if ( NULL == jsObj )
         {
            PD_LOG( PDERROR, "faile to new js object" ) ;
            rc = SDB_OOM ;
            rval.releaseObj() ;
            goto error ;
         }

         JS_SetPrivate( cx, jsObj, rpro.getValue() ) ;

         if ( !rval.getValProperties().empty() )
         {
            rc = _sptInvoker::setProperty( cx, jsObj,
                                           rval.getValProperties() ) ;
            if ( SDB_OK != rc )
            {
               goto error ;
            }
         }

         val = OBJECT_TO_JSVAL( jsObj ) ;
      }
      else
      {
         rc = _getValFromProperty( cx, rpro, val ) ;
         if ( SDB_OK != rc )
         {
            goto error ;
         }
      }

      if ( !rpro.getName().empty() && NULL != obj )
      {
         if ( !JS_SetProperty( cx, obj,
                               rpro.getName().c_str(),
                               &val ))
         {
            PD_LOG( PDERROR, "failed to set obj to parent obj" ) ;
            rc = SDB_SYS ;
            goto error ;
         }
      }

      *rvp = val ;
   done:
      return rc ;
   error:
      goto done ;
   }

   /*
      _sptInvoker implement
   */
   INT32 _sptInvoker::_getValFromProperty( JSContext *cx,
                                           const sptProperty &pro,
                                           jsval &val )
   {
      INT32 rc = SDB_OK ;
      if ( String == pro.getType() )
      {
         JSString *jsstr = JS_NewStringCopyN( cx, pro.getString(),
                                              ossStrlen( pro.getString() ) ) ;
         if ( NULL == jsstr )
         {
            ossPrintf( "%s\n", pro.getString() ) ;
            PD_LOG( PDERROR, "failed to create a js string" ) ;
            rc = SDB_SYS ;
            goto error ;
         }

         val = STRING_TO_JSVAL( jsstr ) ;
      }
      else if ( Bool == pro.getType() )
      {
         BOOLEAN v = TRUE ;
         rc = pro.getNative( Bool, &v ) ;
         if ( SDB_OK != rc )
         {
            goto error ;
         }

         val = BOOLEAN_TO_JSVAL( v ) ;
      }
      else if ( NumberInt == pro.getType() )
      {
         INT32 v = 0 ;
         rc = pro.getNative( NumberInt, &v ) ;
         if ( SDB_OK != rc )
         {
            goto error ;
         }

         val = INT_TO_JSVAL( v ) ;
      }
      else if ( NumberDouble == pro.getType() )
      {
         FLOAT64 v = 0 ;
         rc = pro.getNative( NumberDouble, &v ) ;
         if ( SDB_OK != rc )
         {
            goto error ;
         }
         val = DOUBLE_TO_JSVAL( v ) ;
      }
      else
      {
         PD_LOG( PDERROR, "the type %d is not surpported yet.",
                 pro.getType() ) ;
         rc = SDB_SYS ;
         goto error ;
      }
   done:
      return rc ;
   error:
      goto done ;
   }

   // initialize a log file, file size max 4GB
   // PD_TRACE_DECLARE_FUNCTION ( SDB__DPSLOGFILE_INIT, "_dpsLogFile::init" )
   INT32 _dpsLogFile::init( const CHAR *path, UINT32 size, UINT32 fileNum )
   {
      INT32 rc = SDB_OK ;
      PD_TRACE_ENTRY ( SDB__DPSLOGFILE_INIT ) ;
      BOOLEAN created = FALSE ;

      SDB_ASSERT ( 0 == ( _fileSize % DPS_DEFAULT_PAGE_SIZE ),
                   "Size must be multiple of DPS_DEFAULT_PAGE_SIZE bytes" ) ;

      _fileSize = size ;
      _fileNum  = fileNum ;
      _idleSize = _fileSize ;

      // allocate OSS_FILE, free in destructor
      _file = SDB_OSS_NEW _OSS_FILE();
      if ( !_file )
      {
         rc = SDB_OOM;
         PD_LOG ( PDERROR, "new _OSS_FILE failed!" );
         goto error;
      }

      // if the file exist, restore
      if ( SDB_OK == ossAccess( path ) )
      {
         rc = ossOpen ( path, OSS_READWRITE|OSS_SHAREWRITE, OSS_RWXU, *_file ) ;
         if ( rc == SDB_OK )
         {
            rc = _restore () ;
            if ( rc == SDB_OK )
            {
               UINT32 startOffset = 0 ;
               if ( DPS_INVALID_LSN_OFFSET != _logHeader._firstLSN.offset )
               {
                  startOffset = (UINT32)( _logHeader._firstLSN.offset %
                                          _fileSize ) ;
               }
               PD_LOG ( PDEVENT, "Restore dps log file[%s] succeed, "
                        "firstLsn[%lld], idle space: %u, start offset: %d",
                        path, getFirstLSN().offset, getIdleSize(),
                        startOffset ) ;
               goto done ;
            }
            else
            {
               close () ;
               PD_LOG ( PDEVENT, "Restore dps log file[%s] failed[rc:%d]",
                        path, rc ) ;
               goto error ;
            }
         }
      }

      if ( SDB_OK == ossAccess( path ) )
      {
         rc = ossDelete ( path );
         if ( SDB_IO == rc )
         {
            PD_LOG ( PDERROR, "Failed to delete file at %s", path ) ;
            goto error;
         }
      }

      // open the file with "create only" and "read write" mode, for rx-r-----
      rc = ossOpen( path, OSS_CREATEONLY |OSS_READWRITE | OSS_SHAREWRITE,
                    OSS_RWXU, *_file );

      if ( rc )
      {
         PD_LOG ( PDERROR, "Failed to open log file %s, rc = %d", path, rc ) ;
         goto error;
      }

      created = TRUE ;

      // increase the file size to the given size plus log file header
      rc = ossExtendFile( _file, (SINT64)_fileSize + DPS_LOG_HEAD_LEN );
      if ( rc )
      {
         close() ;
         PD_LOG ( PDERROR, "Failed to extend log file size to %d, rc = %d",
                 size + DPS_LOG_HEAD_LEN, rc ) ;
         goto error;
      }

      _initHead ( DPS_INVALID_LOG_FILE_ID ) ;
      rc = _flushHeader () ;
      if ( rc )
      {
         close () ;
         PD_LOG ( PDERROR, "Failed to flush header, rc = %d", rc ) ;
         goto error ;
      }
      // Currently let's just skip head
      rc = ossSeek ( _file, DPS_LOG_HEAD_LEN, OSS_SEEK_SET ) ;
      if ( rc )
      {
         close() ;
//.........这里部分代码省略.........

   // PD_TRACE_DECLARE_FUNCTION ( SDB_RTNINSERT2, "rtnInsert" )
   INT32 rtnInsert ( const CHAR *pCollectionName, BSONObj &objs, INT32 objNum,
                     INT32 flags, pmdEDUCB *cb, SDB_DMSCB *dmsCB,
                     SDB_DPSCB *dpsCB, INT16 w )
   {
      INT32 rc = SDB_OK ;
      PD_TRACE_ENTRY ( SDB_RTNINSERT2 ) ;
      SDB_ASSERT ( pCollectionName, "collection name can't be NULL" ) ;
      SDB_ASSERT ( cb, "educb can't be NULL" ) ;
      SDB_ASSERT ( dmsCB, "dmsCB can't be NULL" ) ;
      dmsStorageUnit *su = NULL ;
      dmsStorageUnitID suID = DMS_INVALID_CS ;
      const CHAR *pCollectionShortName = NULL ;
      UINT32 insertCount = 0 ;
      BOOLEAN writable = FALSE ;

      ossValuePtr pDataPos = 0 ;
      rc = dmsCB->writable( cb ) ;
      if ( rc )
      {
         PD_LOG ( PDERROR, "Database is not writable, rc = %d", rc ) ;
         goto error;
      }
      writable = TRUE;

      rc = rtnResolveCollectionNameAndLock ( pCollectionName, dmsCB, &su,
                                             &pCollectionShortName, suID ) ;
      if ( rc )
      {
         PD_LOG ( PDERROR, "Failed to resolve collection name %s",
                  pCollectionName ) ;
         goto error ;
      }

      if ( objs.isEmpty () )
      {
         PD_LOG ( PDERROR, "Insert record can't be empty" ) ;
         rc = SDB_INVALIDARG ;
         goto error ;
      }

      pDataPos = (ossValuePtr)objs.objdata() ;
      for ( INT32 i = 0 ; i < objNum ; ++i )
      {
         if ( ++insertCount > RTN_INSERT_ONCE_NUM )
         {
            insertCount = 0 ;
            if ( cb->isInterrupted() )
            {
               rc = SDB_APP_INTERRUPT ;
               goto error ;
            }
         }

         try
         {
            BSONObj record ( (const CHAR*)pDataPos ) ;
            rc = su->insertRecord ( pCollectionShortName, record, cb, dpsCB ) ;
            if ( rc )
            {
               if ( ( SDB_IXM_DUP_KEY == rc ) &&
                    ( FLG_INSERT_CONTONDUP & flags ) )
               {
                  rc = SDB_OK ;
               }
               else
               {
                  PD_LOG ( PDERROR, "Failed to insert record %s into "
                           "collection: %s, rc: %d", record.toString().c_str(),
                           pCollectionName, rc ) ;
                  goto error ;
               }
            }
            pDataPos += ossAlignX ( (ossValuePtr)record.objsize(), 4 ) ;
         }
         catch ( std::exception &e )
         {
            PD_LOG ( PDERROR, "Failed to convert to BSON and insert to "
                     "collection: %s", e.what() ) ;
            rc = SDB_INVALIDARG ;
            goto error ;
         }
      }

   done :
      if ( DMS_INVALID_CS != suID )
      {
         dmsCB->suUnlock ( suID ) ;
      }
      if ( writable )
      {
         dmsCB->writeDown( cb );
      }
      if ( cb )
      {
         if ( SDB_OK == rc && dpsCB )
         {
            rc = dpsCB->completeOpr( cb, w ) ;
         }
      }
//.........这里部分代码省略.........

 // PD_TRACE_DECLARE_FUNCTION ( SDB__DMSROUNIT__INIT, "_dmsReorgUnit::_init" )
 INT32 _dmsReorgUnit::_init ( BOOLEAN createNew )
 {
    INT32 rc = SDB_OK ;
    PD_TRACE_ENTRY ( SDB__DMSROUNIT__INIT );
    class _reorgUnitHead *unitHead = NULL ;
    INT32 bufSize = ossRoundUpToMultipleX (
                          sizeof ( class _reorgUnitHead ),
                          DMS_REORG_UNIT_HEAD_SIZE_UNIT ) ;
    INT32 restSize = bufSize ;
    _headSize = bufSize ;
    CHAR *pBuffer = (CHAR*)SDB_OSS_MALLOC (bufSize) ;
    if ( !pBuffer )
    {
       PD_LOG ( PDERROR, "Failed to allocate %d bytes of memory", bufSize ) ;
       rc = SDB_OOM ;
       goto error ;
    }
    unitHead = (class _reorgUnitHead*)pBuffer ;
    ossMemset ( unitHead, 0, bufSize ) ;
    if ( createNew )
    {
       SINT64 writeSize = 0 ;
       _readOnly = FALSE ;
       ossMemcpy ( unitHead->_eyeCatcher, DMS_REORG_UNIT_EYECATCHER,
                   DMS_REORG_UNIT_EYECATCHER_LEN ) ;
       unitHead->_headerSize = bufSize ;
       ossMemcpy ( unitHead->_fileName, _fileName, OSS_MAX_PATHSIZE ) ;
       unitHead->_pageSize = _pageSize ;
       while ( restSize != 0 )
       {
          rc = ossWrite ( &_file, &pBuffer[bufSize-restSize], restSize,
                          &writeSize ) ;
          if ( rc && SDB_INTERRUPT != rc )
          {
             PD_LOG ( PDERROR, "Failed to write into file: %s, rc = %d",
                      _fileName, rc ) ;
             goto error ;
          }
          restSize -= writeSize ;
          rc = SDB_OK ;
       }
    }
    else
    {
       SINT64 readSize = 0 ;
       _readOnly = TRUE ;
       while ( restSize > 0 )
       {
          rc = ossRead ( &_file, &pBuffer[bufSize-restSize], restSize, &readSize ) ;
          if ( rc && SDB_INTERRUPT != rc )
          {
             PD_LOG ( PDERROR, "Failed to read from file: %s, rc = %d",
                      _fileName, rc ) ;
             goto error ;
          }
          restSize -= readSize ;
          rc = SDB_OK ;
       }
       if ( ossMemcmp ( unitHead->_eyeCatcher, DMS_REORG_UNIT_EYECATCHER,
                        DMS_REORG_UNIT_EYECATCHER_LEN ) ||
            unitHead->_headerSize != bufSize )
       {
          PD_LOG ( PDERROR, "Invalid reorg file is detected" ) ;
          rc = SDB_DMS_INVALID_REORG_FILE ;
          goto error ;
       }
    }
 done :
    if ( pBuffer )
    {
       SDB_OSS_FREE ( pBuffer ) ;
    }
    PD_TRACE_EXITRC ( SDB__DMSROUNIT__INIT, rc );
    return rc ;
 error :
    goto done ;
 }

INT32 _ossModuleHandle::init ()
{
   INT32 rc = SDB_OK ;
   CHAR  strPath [ 2*OSS_MAX_PATHSIZE + 1 ] = {0} ;
   CHAR  strModule [ OSS_MAX_PATHSIZE + 1 ] = {0} ;
   OSS_MHANDLE handle = 0 ;
   CHAR *p = NULL ;
#if defined (_WINDOWS)
   UINT32 errorMode ;
#endif
   PD_TRACE_ENTRY ( SDB_OSSMODULEHANDLE_INIT ) ;
   if ( _moduleName[0] == '\0' )
   {
      PD_LOG ( PDERROR, "Module name can't be empty" ) ;
      rc = SDB_INVALIDARG ;
      goto error ;
   }
   PD_TRACE3 ( SDB_OSSMODULEHANDLE_INIT, PD_PACK_STRING(_moduleName),
               PD_PACK_STRING(_libPath), PD_PACK_UINT(_flags) ) ;
   ossStrncpy ( strModule, _moduleName, sizeof(strModule) ) ;
   p = ossStrchr ( strModule, '(' ) ;
   if ( p )
   {
      *p = '\0' ;
   }
   rc = patchModuleName( strModule, _moduleName, sizeof(_moduleName) );
   PD_RC_CHECK ( rc, PDERROR, "Failed to patch module name, rc = %d", rc ) ;
   if ( _libPath[0] )
   {
      INT32 pathLen = 0 ;
      rc = ossAccess ( _libPath,
#if defined (_LINUX)
                       F_OK
#elif defined (_WINDOWS)
                       0
#endif
      ) ;
      PD_RC_CHECK ( rc, PDERROR, "Failed to access path %s, rc = %d",
                    _libPath, rc ) ;
      ossStrncat ( strPath, _libPath, sizeof(strPath) ) ;
      pathLen = ossStrlen ( strPath ) ;
      if ( strPath[pathLen-1] != OSS_FILE_SEP_CHAR )
      {
         if ( pathLen >= OSS_MAX_PATHSIZE )
         {
            PD_LOG ( PDERROR, "library path is too long: %s",
                     _libPath ) ;
            rc = SDB_INVALIDARG ;
            goto error ;
         }
         strPath[pathLen-1] = OSS_FILE_SEP_CHAR ;
         strPath[pathLen]   = '\0' ;
      }
   }
   if ( ossStrlen ( strPath ) + ossStrlen ( _moduleName ) >= sizeof(strPath) )
   {
      PD_LOG ( PDERROR, "path + module name is too long: %s:%s",
               strPath, _moduleName ) ;
      rc = SDB_INVALIDARG ;
      goto error ;
   }
   ossStrncat ( strPath, _moduleName, OSS_MAX_PATHSIZE ) ;
#if defined (_LINUX)
   handle = dlopen ( strPath, _flags | RTLD_NOW ) ;
   if ( !handle )
   {
      PD_LOG ( PDERROR, "Failed to load module %s, error = %s",
               strPath, dlerror() ) ;
      rc = SDB_SYS ;
      goto error ;
   }
   _isInitialized = TRUE ;
   _moduleHandle = handle ;
   dlerror() ;
#elif defined (_WINDOWS)
   errorMode = SetErrorMode ( SEM_NOOPENFILEERRORBOX |
                              SEM_FAILCRITICALERRORS ) ;
   _moduleHandle = LoadLibrary ( (LPCTSTR)strPath ) ;
   SetErrorMode ( errorMode ) ;
   if ( NULL == _moduleHandle )
   {
      rc = ossGetLastError () ;
      PD_LOG ( PDERROR, "Failed to load module %s, error = %d",
               _moduleName, rc ) ;
      OSSMODULEHANDLE_ERR(rc) ;
      goto error ;
   }
#endif
done :
   PD_TRACE_EXITRC ( SDB_OSSMODULEHANDLE_INIT, rc ) ;
   return rc ;
error :
   _isInitialized = FALSE ;
   _moduleHandle = OSS_MINVALIDHANDLE ;
   goto done ;
}

int pmdEDUMgr::_createNewEDU(EDU_TYPES type, void* arg, EDUID *eduid)
{
	int rc = EDB_OK;
	unsigned int probe	= 0;
	pmdEDUCB *cb		= NULL;
	EDUID myEDUID		= 0;
	if (isQuiesced()) {
		rc = EDB_QUIESCED;
		goto done;	
	}

	if (!getEntryFuncByType(type)) {
		PD_LOG(PDERROR, "The edu[type:%d] not exist or function is null", type);
		rc = EDB_INVALIDARG;
		probe = 30;
		goto error;	
	}

	cb = new(std::nothrow) pmdEDUCB(this, type);
	EDB_VALIDATE_GOTOERROR(cb, EDB_OOM, "Out of memery to create agent control block");
	cb->setStatus(PMD_EDU_CREATING);

	_mutex.get();
	if (_runQueue.end() != _runQueue.find(_EDUID)) {
		_mutex.release();
		rc = EDB_SYS;
		probe = 10;
		goto error;	
	}

	if (_idleQueue.end() != _idleQueue.find(_EDUID)) {
		_mutex.release();
		rc = EDB_SYS;
		probe = 15;
		goto error;	
	}

	cb->setId(_EDUID);
	if (eduid) {
		*eduid = _EDUID;	
	}

	_runQueue[_EDUID] = (pmdEDUCB*) cb;
	myEDUID = _EDUID;
	++_EDUID;
	_mutex.release();

	try {
		boost::thread agentThread(pmdEDUEntryPoint, type, cb, arg);
		agentThread.detach();	
	} catch (std::exception e) {
		_runQueue.erase(myEDUID);
		rc = EDB_SYS;
		probe = 20;
		goto error;
	}


	cb->postEvent(pmdEDUEvent(PMD_EDU_EVENT_RESUME, false, arg));

done:
	return rc;
error:
	if (cb) {
		delete cb;	
	}
	PD_LOG(PDERROR, "Failed to create new agent, probe = %d", probe);
	goto done;
}

   INT32 omRemoveBusinessCommand::doCommand()
   {
      INT32 rc = SDB_OK ;
      INT64 taskID = 0 ;
      BSONObj buzInfo ;
      BSONObj taskConfig ;
      BSONArray resultInfo ;
      omArgOptions option( _restAdaptor, _restSession ) ;
      omRestTool restTool( _restAdaptor, _restSession ) ;

      _setFileLanguageSep() ;

      pmdGetThreadEDUCB()->resetInfo( EDU_INFO_ERROR ) ;

      rc = option.parseRestArg( "s",
                                OM_REST_FIELD_BUSINESS_NAME, &_businessName ) ;
      if ( rc )
      {
         _errorMsg.setError( TRUE, option.getErrorMsg() ) ;
         PD_LOG( PDERROR, "failed to parse rest arg: rc=%d", rc ) ;
         goto error ;
      }

      rc = _check( buzInfo ) ;
      if ( rc )
      {
         PD_LOG( PDERROR, "failed to check: rc=%d", rc ) ;
         goto error ;
      }

      rc = _generateRequest( buzInfo, taskConfig, resultInfo ) ;
      if ( rc )
      {
         PD_LOG( PDERROR, "failed to generate task request: rc=%d", rc ) ;
         goto error ;
      }

      rc = _createTask( taskConfig, resultInfo, taskID ) ;
      if ( rc )
      {
         PD_LOG( PDERROR, "failed to create task: rc=%d", rc ) ;
         goto error ;
      }

      {
         BSONObj result = BSON( OM_BSON_TASKID << taskID ) ;

         rc = restTool.appendResponeContent( result ) ;
         if ( rc )
         {
            _errorMsg.setError( TRUE, "failed to append respone content: rc=%d",
                                rc ) ;
            PD_LOG( PDERROR, _errorMsg.getError() ) ;
            goto error ;
         }
      }

      restTool.sendOkRespone() ;

   done:
      return rc ;
   error:
      restTool.sendRespone( rc, _errorMsg.getError() ) ;
      goto done ;
   }

   INT32 omRemoveBusinessCommand::_check( BSONObj &buzInfo )
   {
      INT32 rc = SDB_OK ;
      INT64 taskID = -1 ;
      omDatabaseTool dbTool( _cb ) ;

      if ( FALSE == dbTool.isBusinessExist( _businessName ) )
      {
         rc = SDB_INVALIDARG ;
         _errorMsg.setError( TRUE, "business does not exist: name=%s",
                             _businessName.c_str() ) ;
         PD_LOG( PDERROR, _errorMsg.getError() ) ;
         goto error ;
      }

      if ( TRUE == dbTool.isRelationshipExistByBusiness( _businessName ) )
      {
         rc = SDB_INVALIDARG ;
         _errorMsg.setError( TRUE, "business has relationship: name=%s",
                             _businessName.c_str() ) ;
         PD_LOG( PDERROR, _errorMsg.getError() ) ;
         goto error ;
      }

      taskID = dbTool.getTa