masterserver/server/masterserver/gsmgr.cc

#include "precompile.h"

#include <a8/mutable_xobject.h>

#include <f8/utils.h>
#include <f8/timer.h>
#include <f8/jsonhttprequest.h>

#include "gsmgr.h"
#include "app.h"
#include "GGListener.h"

#include "ss_proto.pb.h"

struct GSNode
{
    std::string key;
    long long node_idx = 0;
    int node_id = 0;
    int instance_id = 0;
    int room_num = 0;
    int alive_count = 0;
    int online_num = 0;
    std::string ip;
    int port = 0;
    bool servicing = false;
    int version = 0;
    f8::TimerWp timer_wp;
};

void GSMgr::Init()
{
}

void GSMgr::UnInit()
{

}

void GSMgr::_SS_WSP_RequestTargetServer(f8::MsgHdr& hdr, const ss::SS_WSP_RequestTargetServer& msg)
{
    ss::SS_MS_ResponseTargetServer respmsg;
    respmsg.set_context_id(msg.context_id());
    if (msg.is_reconnect()) {
        std::shared_ptr<GSNode> node = GetNodeByNodeKey(msg.server_info());
        if (node) {
            respmsg.set_host(node->ip);
            respmsg.set_port(node->port);
        } else {
            respmsg.set_error_code(1);
            respmsg.set_error_msg("cant alloc node");
        }
    } else {
        std::shared_ptr<GSNode> node = GetNodeByTeamId(msg.team_id());
        if (node) {
            respmsg.set_host(node->ip);
            respmsg.set_port(node->port);
        } else {
            node = AllocNode();
            if (node) {
                respmsg.set_host(node->ip);
                respmsg.set_port(node->port);
                team_hash_[msg.team_id()] = node;
            } else {
                respmsg.set_error_code(1);
                respmsg.set_error_msg("cant alloc node");
            }
         }
    }
    GGListener::Instance()->SendMsg(hdr.socket_handle, respmsg);
}

void GSMgr::_SS_Ping(f8::MsgHdr& hdr, const ss::SS_Ping& msg)
{
    ss::SS_Pong pongmsg;
    GGListener::Instance()->SendMsg(hdr.socket_handle, pongmsg);
}

void GSMgr::___GSReport(std::shared_ptr<f8::JsonHttpRequest> request)
{
    std::string ip = request->GetParams()->Get("ip");
    int port = request->GetParams()->Get("port");
    int alive_count = request->GetParams()->Get("alive_count");
    int online_num = request->GetParams()->Get("online_num");
    int room_num = request->GetParams()->Get("room_num");
    int instance_id = request->GetParams()->Get("instance_id");
    int node_id = request->GetParams()->Get("node_id");
    int version = request->GetParams()->Get("version");
    bool servicing = request->GetParams()->Get("servicing");
    std::string key = ip + ":" + a8::XValue(port).GetString();

    auto itr = node_key_hash_.find(key);
    if (itr != node_key_hash_.end()) {
        if (itr->second->online_num != online_num ||
            itr->second->room_num != room_num ||
            itr->second->alive_count != alive_count ||
            itr->second->servicing != servicing ||
            itr->second->version != version
            ) {
            itr->second->online_num = online_num;
            itr->second->room_num = room_num;
            itr->second->servicing = servicing;
            if (itr->second->version != version) {
                itr->second->version = version;
                OnVersionChange(itr->second);
            }
            RearrangeNode();
        }
        itr->second->alive_count = alive_count;
        if (!itr->second->timer_wp.expired()) {
            f8::Timer::Instance()->ModifyTime(itr->second->timer_wp, 1000 * 5);
        }
    } else {
        auto gs = std::make_shared<GSNode>();
        gs->key = key;
        gs->node_id = node_id;
        gs->node_idx = f8::App::Instance()->NewNodeUuid();
        gs->instance_id = instance_id;
        gs->alive_count = alive_count;
        gs->online_num = online_num;
        gs->room_num = room_num;
        gs->ip = ip;
        gs->port = port;
        gs->servicing = servicing;
        gs->version = version;
        gs->timer_wp = f8::Timer::Instance()->SetIntervalWp
            (
             1000 * 5,
             [this, gs] (int e, const a8::Args* args)
             {
                 if (a8::TIMER_EXEC_EVENT == e) {
                     RemoveNodeFromSortedNodes(gs);
                     {
                         std::vector<std::string> deleted_teams;
                         for (auto& pair : team_hash_) {
                             if (pair.second == gs) {
                                 deleted_teams.push_back(pair.first);
                             }
                         }
                         for (const std::string& team_id : deleted_teams) {
                             team_hash_.erase(team_id);
                         }
                     }
                     node_key_hash_.erase(gs->key);
                     RearrangeNode();
                 }
             });
        node_key_hash_[key] = gs;
        AddNodeToSortedNodes(gs);
        RearrangeNode();
    }

    request->GetResp()->SetVal("errcode", 0);
    request->GetResp()->SetVal("errmsg", "");
}

void GSMgr::___GSList(std::shared_ptr<f8::JsonHttpRequest> request)
{
    {
        auto node_list = a8::MutableXObject::CreateArray();

        for (auto& pair : node_key_hash_) {
            auto node = a8::MutableXObject::CreateObject();
            node->SetVal("node_id", pair.second->node_id);
            node->SetVal("instance_id", pair.second->instance_id);
            node->SetVal("room_num", pair.second->room_num);
            node->SetVal("alive_count", pair.second->alive_count);
            node->SetVal("online_num", pair.second->online_num);
            node->SetVal("ip", pair.second->ip);
            node->SetVal("port", pair.second->port);
            node->SetVal("servicing", pair.second->servicing);
            node->SetVal("version", pair.second->version);
            node_list->Push(*node);
        }

        request->GetResp()->SetVal("errcode", 0);
        request->GetResp()->SetVal("errmsg", "");
        request->GetResp()->SetVal("node_list", *node_list);
    }
    {
        auto node_list = a8::MutableXObject::CreateArray();

        for (auto& pair : sorted_node_hash_) {
            for (std::shared_ptr<GSNode> gs_node : pair.second) {
                auto node = a8::MutableXObject::CreateObject();
                node->SetVal("node_id", gs_node->node_id);
                node->SetVal("instance_id", gs_node->instance_id);
                node->SetVal("room_num", gs_node->room_num);
                node->SetVal("online_num", gs_node->online_num);
                node->SetVal("ip", gs_node->ip);
                node->SetVal("port", gs_node->port);
                node->SetVal("servicing", gs_node->servicing);
                node->SetVal("version", gs_node->version);
                node_list->Push(*node);
            }
        }

        request->GetResp()->SetVal("errcode", 0);
        request->GetResp()->SetVal("errmsg", "");
        request->GetResp()->SetVal("sorted_node_list", *node_list);
    }
}

std::shared_ptr<GSNode> GSMgr::GetNodeByTeamId(const std::string& team_id)
{
    auto itr = team_hash_.find(team_id);
    return itr != team_hash_.end() ? itr->second : nullptr;
}

std::shared_ptr<GSNode> GSMgr::GetNodeByNodeKey(const std::string& node_key)
{
    auto itr = node_key_hash_.find(node_key);
    return itr != node_key_hash_.end() ? itr->second : nullptr;
}

std::shared_ptr<GSNode> GSMgr::AllocNode()
{
    std::vector<std::shared_ptr<GSNode>>* sorted_nodes = GetSortedNodes();
    if (sorted_nodes && !sorted_nodes->empty()) {
        size_t rnd = std::min((size_t)2, sorted_nodes->size());
        if (rnd >= 2) {
            std::shared_ptr<GSNode> n1 = sorted_nodes->at(sorted_nodes->size() - 1);
            std::shared_ptr<GSNode> n2 = sorted_nodes->at(sorted_nodes->size() - 2);
            if (n1->online_num < 100 && n2->online_num < 100) {
                if (n1->instance_id < n2->instance_id) {
                    if (n1->servicing) {
                        return n1;
                    }
                } else {
                    if (n2->servicing) {
                        return n2;
                    }
                }
            }
        }
        int idx = rand() % rnd;
        while (idx >= 0) {
            if (sorted_nodes->at(idx)->servicing) {
                return sorted_nodes->at(idx);
            }
            --idx;
        }
    }
    f8::UdpLog::Instance()->Warning
        ("节点分配失败 sorted_node_hashlist.size:%d node_list.size:%d sorted_node.size:%d",
         {
          sorted_node_hash_.size(),
          node_key_hash_.size(),
          sorted_nodes ? sorted_nodes->size() : 0,
         });
    return nullptr;
}

void GSMgr::RearrangeNode()
{
    for (auto& pair : sorted_node_hash_) {
        std::sort(pair.second.begin(), pair.second.end(),
                  [] (const std::shared_ptr<GSNode> a, const std::shared_ptr<GSNode> b)
                  {
                      if (a->servicing && b->servicing) {
                          if (a->online_num < b->online_num) {
                              return true;
                          }
                          if (a->online_num > b->online_num) {
                              return false;
                          }
                          if (a->room_num < b->room_num) {
                              return true;
                          }
                          if (a->room_num > b->room_num) {
                              return false;
                          }
                          return a->node_idx > b->node_idx;
                      }
                      if (a->servicing) {
                          return true;
                      }
                      if (b->servicing) {
                          return false;
                      }
                      return a->node_idx > b->node_idx;
                  }
                  );
    }
}

void GSMgr::AddNodeToSortedNodes(std::shared_ptr<GSNode> node)
{
    auto itr = sorted_node_hash_.find(node->version);
    if (itr != sorted_node_hash_.end()) {
        itr->second.push_back(node);
    } else {
        sorted_node_hash_[node->version] = std::vector<std::shared_ptr<GSNode>>({node});
    }
}

void GSMgr::RemoveNodeFromSortedNodes(std::shared_ptr<GSNode> node)
{
    for (auto& pair : sorted_node_hash_) {
        std::vector<std::shared_ptr<GSNode>>& node_list = pair.second;
        for (size_t i = 0; i < node_list.size(); ++i) {
            if (node_list[i] == node) {
                node_list.erase(node_list.begin() + i);
                break;
            }
        }
    }
}

std::vector<std::shared_ptr<GSNode>>* GSMgr::GetSortedNodes()
{
    if (sorted_node_hash_.empty()) {
        return nullptr;
    }
    std::vector<std::shared_ptr<GSNode>>* result = nullptr;
    for (auto& pair : sorted_node_hash_) {
        result = &pair.second;
    }
    return result;
}

void GSMgr::OnVersionChange(std::shared_ptr<GSNode> node)
{
    RemoveNodeFromSortedNodes(node);
    AddNodeToSortedNodes(node);
}