AI服务器的设计与实现

经过一段时间的设计与完善，我们游戏的AI服务器已经达到了基本的性能要求，目前单个AI进程可同时运行4000+个频繁的AI对象。

在前面一篇博客中已经提到过，AI服务器的主逻辑循环是单线程的，这个线程上运行了数千个用户级线程，每个用户级线程运行一个AI对象。AI对象被激活之后就会运行一段lua脚本，以实现AI逻辑.

之所以采用用户级线程(windows下是fiber,linux下使用ucontext)的方案，是因为AI的实现使用了大量的远程调用，如果使用同步调用势必导致主线程的阻塞，从而影响AI服务器的性能。采用异步调用又导致了逻辑的过分复杂。而用户级线程正好解决了这些问题，向上提供了一个同步调用的接口，又不会导致主线程的阻塞（当一个用户级线程处于等待结果的状态下，调度器可以选择另一个用户级线程来运行）。

AI服务器的主要构件是用户级线程调度器，和一个用户级线程池，服务器启动后会产生一组用户级线程序，并且在每个线程上创建一个lua虚拟机。

 

基本的设计思路已经介绍完毕，下面介绍各个主要的组成部分：

首先是主循环：

 

void CAIApp::Process()
{
    psarmor l_pa(*this);
    Scheduler::Init();
    while(!GetExitTaskFlag() && l_pa(psobj::realtime))
    {
        //如果到game的连接断开，执行错误处理并尝试重连
        while(!m_flag2Game)
        {
            //连接断了，要清除所有已经绑定的Ai对象
            //g_AiObjMap为空的话不可能有任务在运行
            if(!g_AiObjMap.empty())
            {    
                //连接已经断开，停掉所有运行的AI
                {
                    std::map<uLong,rptr<AiAvatar> >::iterator it =  g_AiObjMap.begin();
                    std::map<uLong,rptr<AiAvatar> >::iterator end =  g_AiObjMap.end();
                    for( ; it != end; ++it)
                        it->second->StopAi();
                }
                //清理active列表
                Scheduler::ClearActiveList();
                //清理timeout列表
                Scheduler::ClearTimeOut();
                {
                    std::cout << "到gameserver的连接断开,清除所有绑定对象" << std::endl;
                    std::map<uLong,rptr<AiAvatar> >::iterator it =  g_AiObjMap.begin();
                    std::map<uLong,rptr<AiAvatar> >::iterator end =  g_AiObjMap.end();
                    for( ; it != end; ++it)
                        it->second = 0;
                    g_AiObjMap.clear();
                }
                //清理aigroup    
                {
                    std::map<long,rptr<AiGroup> >::iterator it =  g_GroupMap.begin();
                    std::map<long,rptr<AiGroup> >::iterator end =  g_GroupMap.end();
                    for( ; it != end; ++it)
                        it->second = 0;
                    g_GroupMap.clear();
                }
            }
            m_pToGame = 0;
            while(m_pToGame._nil())
            {
                rptr<DataSocket> l_sock    =g_aiapp->Connect(g_aiapp->m_config.m_gameip,g_aiapp->m_config.m_gameport);
                if(l_sock._nil())
                {
                    std::cout << "连接game失败!5秒后重试..." << std::endl;
                }
                else
                {
                    printf("连接game成功...");
                    WPacket    l_wpk    =g_aiapp->GetWPacket();
                    l_wpk.WriteCmd(CMD_AM_AILOGIN);
                    l_wpk.WriteShort(g_aiapp->m_config.m_mapcount);
                    for( int i = 0;i < g_aiapp->m_config.m_mapcount; ++i)
                    {
                        l_wpk.WriteString(g_aiapp->m_config.m_names[i].c_str());
                    }
                    l_sock->SendData(l_wpk);
                    m_pToGame = l_sock;
                    m_flag2Game = true;
                    break;
                }
                Sleep(5000);
            }
        }
        Scheduler::Schedule();
        PeekPacket(50);
    }
    Scheduler::Destroy();
}

上面代码的主要作用就是尝试连接gameserver，如果连接成功就在循环中调用调度器的调度函数以选择合适的用户级线程运行。PeekPacket(50);会从网络层提取网络包，如果没有网络包则会休眠最多50毫秒.

下面在来看看调度器：

void Scheduler::Schedule()
{
    //将所有等待添加到m_activeList中的纤程都添加进去
    {
        for(unsigned int i = 0; i < pending_index; ++i)
        {
            uthread *ut = m_uthreads[m_pendingAdd[i]];
            ut->SetNext(0);
            if(m_active_tail)
            {
                m_active_tail->SetNext(ut);
                m_active_tail = ut;
            }
            else
            {
                m_active_head = m_active_tail = ut;
            }
        }
        pending_index = 0;
    }
    uthread *cur = m_active_head;
    uthread *pre = NULL;
    while(cur)
    {
        g_aiapp->PeekPacket(0);
        m_curuid = cur->GetUid();
        SwitchToFiber(cur->GetUContext());
        m_curuid = -1;
        unsigned char status = cur->GetStatus();
        //当纤程处于以下状态时需要从可运行队列中移除
        if(status == DEAD || status == SLEEP || status == WAIT4EVENT || status == UNACTIVED || status == YIELD)
        {
            //删除首元素
            if(cur == m_active_head)
            {
                //同时也是尾元素
                if(cur == m_active_tail)
                    m_active_head = m_active_tail = NULL;
                else
                    m_active_head = cur->Next();
            }
            else if(cur == m_active_tail)
            {
                    pre->SetNext(NULL);
                    m_active_tail = pre;
            }
            else
                pre->SetNext(cur->Next());
            uthread *tmp = cur;
            cur = cur->Next();
            tmp->SetNext(0);
            //如果仅仅是让出处理器，需要重新投入到可运行队列中
            if(status == YIELD)
                Add2Active(tmp);
            
        }
        else
        {
            pre = cur;
            cur = cur->Next();
        }
    }
    //看看有没有timeout的纤程
    {
        uLong now = dbc::GetTickCount();
        while(m_timeoutlist.Min() !=0 && m_timeoutlist.Min() <= now)
        {
            st_timeout *timeout = m_timeoutlist.PopMin();
            if(timeout->ut->GetStatus() == WAIT4EVENT || timeout->ut->GetStatus() == SLEEP)
            {
                timeout->ut->wakeuptick = timeout->_timeout;
                Add2Active(timeout->ut);
            }
        }
    }
}

 

调度器首先将重新处于激活态的线程投入到运行队列中，然后遍历可运行队列，运行其中的线程，调度器的最后将处理所有处于休眠状态的线程，如果线程的休眠时间到了，则将线程重新投入到可运行队列中。在这里使用了一个极小堆来处理超时。

从上面的代码可以看出，当调度器挑选了一个线程运行之后，代码路径就跳转到线程中，当线程需要阻塞时，就会设置一个状态（YIELD, WAIT4EVENT或SLEEP）并将运行权又重新交回给调度器，当调度器重新获得运行权后，代码会从SwitchToFiber(cur->GetUContext());中返回，调度器需要根据上次运行的线程的状态，或者将线程投入休眠队列（SLEEP）,或者重新将线程投入到队列的末尾（YIELD）或者从运行队列中删除(WAIT4EVENT).

      

下面再看看一个同步调用的例子:

以移动为例，假设AI请求移动到某给位置，则需要向gameserver发送移动请求，直到到达目标点，或者发现移动失败才会从调用中返回:

int AiAvatar::Move(Point3D &pt,short cntx,uLong ms)
{
    class PosBlock : public BlockStruct
    {    
    public:
        PosBlock(Point3D &pos,AiAvatar *ava)
            :m_ava(ava),m_targetpos(pos){}
        
        //返回true则纤程从阻塞中恢复
        bool WakeUp()
        {
            //到达了请求点，恢复
            if(m_targetpos.x == m_ava->GetPos().x &&
               m_targetpos.y == m_ava->GetPos().y )
            {
                return true;
            }
            return false;
        }
    private:
        Point3D m_targetpos;
        AiAvatar *m_ava;
    };
    //printf("开始移动/n");
    //向GameServer发送移动请求
    WPacket    l_wpk = g_aiapp->GetWPacket();
    l_wpk.WriteCmd(CMD_AM_BEGMOV);
    l_wpk.WriteLong(pt.x);
    l_wpk.WriteLong(pt.y);
    l_wpk.WriteLong(pt.z);
    l_wpk.WriteShort(cntx);
    Send2Game(this,l_wpk);
    //阻塞所在fiber直到pos到达要求的值/或者收到移动失败消息/或则AI被请求停止
    PosBlock pb(pt,this);
    Scheduler::Block(&pb,ms);
    //接到了停止AI的命令
    if(!isAiRunning())
        return -1;
    bool ret = (pt.x == m_pos.x && pt.y == m_pos.y);
    return ret ? 1:0;
}

 

函数首先创建了一个阻塞条件的结构，然后阻塞在这个条件上，在这里是判断AI对象是否到达了目标点。然后将移动请求发送出去并阻塞在条件上。当gameserver把对象移动到正确的点之后，会把对象的坐标通过网络同步到AI服务器，处理网络包的时候发现那个对象对应的线程正被阻塞，就会调用阻塞条件的WakeUp函数尝试唤醒线程，此时如果条件满足,WakeUp就会返回true,线程被重新投入到可运行队列中，否则线程就会继续被阻塞。

 

最用来看一段AI脚本，当一个AI对象被激活（进入玩家的视野），就会为这个对象分配一个线程，这个线程就会马上运行与这个对象相关的lua入口函数:

function monster_routine(this)        
    --出生点
    local start_pos = {}
    start_pos.x,start_pos.y,start_pos.z = getbegpos(this)
        
    local c = 1
    
    --巡逻点
    local points = {
        {x=start_pos.x+300,y=start_pos.y,z=start_pos.z},
        {x=start_pos.x,y=start_pos.y,z=start_pos.z}
    }
    
    
    --生成状态机
    stateMachine = AiStateMachine:new()
    stateMachine.owner = this
    --初始化trace
    stateMachine.state_trace = trace:new():init(this,stateMachine,start_pos)
    --stateMachine.state_trace:init(this,stateMachine,start_pos)
    --初始化partol
    stateMachine.state_partol = partol:new():init(this,stateMachine,start_pos,points)
    --stateMachine.state_partol:init(this,stateMachine,start_pos,points)
    --初始化attack
    stateMachine.state_attack = attack:new():init(this,stateMachine)
    --stateMachine.state_attack:init(this,stateMachine)
    --初始化goback
    stateMachine.state_goback = goback:new():init(this,stateMachine,start_pos)
    --stateMachine.state_goback:init(this,stateMachine,start_pos)
    --初始化help
    stateMachine.state_help = help:new():init(this,stateMachine)
    --stateMachine.state_help:init(this,stateMachine)
    
    stateMachine.cur_state = stateMachine.state_partol
    
    while isAiRunning(this) == true do
                    
        if isdead(this) == true then
            sc_yield()
        else    
            stateMachine.cur_pos.x,stateMachine.cur_pos.y,stateMachine.cur_pos.z = getpos(this)
            stateMachine.target = get_target(this)
            if stateMachine.target == nil then
                stateMachine.target = select_target(this)
            end
            
            
            --查看是否有消息要处理
            local sender
            local recver
            local msg
            local sendtick
            sender,recver,msg,sendtick = PopMsg(this)
            if sender ~= nil then
                print("消息队列非空")
                if msg == "help" then
                    --如果自己没有目标才处理帮助请求
                    if stateMachine.target == nil then
                        stateMachine.target = sender
                        stateMachine.cur_state = stateMachine.state_help
                    end
                end
            end
            
            local ret = 0
            ret,stateMachine.cur_state = stateMachine.cur_state:execute()    
            if ret == -1 then
                return
            end
            sc_yield()
        end
        
    end
end

AI主入口函数首先创建了一个状态机，并选择一个初始状态运行。下面再看看追击状态的处理:

trace = {
owner = 0,
StateMachine = 0,
start_pos = 0
}
 
function trace:init(owner,statemachine,start_pos)
    self.owner = owner
    self.StateMachine = statemachine
    self.start_pos = start_pos
    return self
end 
 
--追击 
function trace:execute()
    
    if self.StateMachine.target == nil then
        --判断离出生点的距离，太远了就回出生点
        local dis2begpos = calDistance(self.start_pos.x,self.start_pos.y,self.StateMachine.cur_pos.x,self.StateMachine.cur_pos.y)
        if dis2begpos >= 500 then
            return 0,self.StateMachine.state_goback
        else
              --没有目标，巡逻
              return 0,self.StateMachine.state_partol
        end
    else
        local dis2begpos = calDistance(self.start_pos.x,self.start_pos.y,self.StateMachine.cur_pos.x,self.StateMachine.cur_pos.y)
        if dis2begpos >= 4000 then
            return 0,self.StateMachine.state_goback
        else
              --取得目标当前点            
            local target_pos ={}
            target_pos.x,target_pos.y,target_pos.z = getpos(self.StateMachine.target)
            
            local dis = calDistance(self.StateMachine.cur_pos.x,self.StateMachine.cur_pos.y,target_pos.x,target_pos.y)    
            if dis <= 200 then
                --print("选择攻击点")
                --选择攻击点
                --if cur_pos.x ~= self.cur_pos and cur_pos.y ~= self.cur_pos.y then
                    local d_x,d_y = gen_pos_circle(target_pos.x,target_pos.y,200)
        
                    if -1 == mov(self.owner,d_x,d_y,target_pos.z,804,1000) then
                        return -1,nil
                    end
                    --面向目标
                      turnface(self.owner,self.StateMachine.target)
                    --切换到攻击态
                  --end
                return 0,self.StateMachine.state_attack
            else
                --在目标半径2米内随机选择一个点,作为目标点
                local d_x,d_y = gen_pos_line(self.StateMachine.cur_pos.x,self.StateMachine.cur_pos.y,target_pos.x,target_pos.y,200,100)
                if dis <= 300 then
                    --离目标点小于3米直接过去
                    if -1 == mov(self.owner,d_x,d_y,target_pos.z,804,1000) then
                          return -1,nil
                      end
                else                            
                    local ttx,tty = forword(self.owner,d_x,d_y,300)
                      if -1 == mov(self.owner,ttx,tty,target_pos.z,804,1000) then
                          return -1,nil
                      end
                end
            end
        end
    end
    return 0,self.StateMachine.state_trace
end
        
function trace:new(o)
  o = o or {}   
  setmetatable(o, self)
  self.__index = self
  return o
end

 

在追击状态下，根据各种条件或者执行追击，或者返回下一个状态.