wsproxy/server/wsproxy/app.cc

#include "precompile.h"

#include <string.h>

#include <a8/xtimer.h>
#include <a8/uuid.h>
#include <a8/udplistener.h>

#include <f8/netmsghandler.h>
#include <f8/udplog.h>
#include <f8/app.h>
#include <f8/msgqueue.h>

#include "app.h"
#include "GCListener.h"
#include "jsondatamgr.h"
#include "handlermgr.h"
#include "downstream.h"
#include "downstreammgr.h"

#include "upstream.h"
#include "upstreammgr.h"
#include "master.h"
#include "mastermgr.h"

#include "longsessionmgr.h"

#include "ss_msgid.pb.h"
#include "ss_proto.pb.h"

struct UdpMsgNode
{
    a8::UdpPacket* pkt;
    UdpMsgNode* next;
};

static void SavePerfLog()
{
    f8::UdpLog::Instance()->Info
        ("max_run_delay_time:%d max_timer_idle:%d "
         "in_data_size:%d out_data_size:%d msgnode_size:%d udp_msgnode_size:%d "
         "read_count:%d max_login_time:%d "
         "max_join_time:%d tcp_count:%d udp_count:%d down_stream_count:%d",
         {
             App::Instance()->GetPerf().max_run_delay_time,
                 App::Instance()->GetPerf().max_timer_idle,
                 App::Instance()->GetPerf().in_data_size,
                 App::Instance()->GetPerf().out_data_size,
                 f8::App::Instance()->GetMsgNodeSize(),
                 App::Instance()->GetUdpMsgNodeSize(),
                 App::Instance()->GetPerf().read_count,
                 App::Instance()->GetPerf().max_login_time,
                 App::Instance()->GetPerf().max_join_time,
                 GCListener::Instance()->GetSocketCount(),
                 LongSessionMgr::Instance()->GetLongSessionCount(),
                 DownStreamMgr::Instance()->GetDownStreamCount()
                 });
    if (f8::App::Instance()->HasFlag(2)) {
        a8::XPrintf("mainloop_time:%d netmsg_time:%d send_node_num:%d sent_bytes_num:%d\n",
                    {
                     App::Instance()->GetPerf().max_run_delay_time,
                     App::Instance()->GetPerf().max_dispatchmsg_time,
                     GCListener::Instance()->GetSendNodeNum(),
                     GCListener::Instance()->GetSentBytesNum()
                    });
    }
    App::Instance()->GetPerf().max_run_delay_time = 0;
    App::Instance()->GetPerf().max_timer_idle = 0;
    App::Instance()->GetPerf().max_login_time = 0;
    App::Instance()->GetPerf().max_join_time = 0;
}

const std::string App::GetPkgName()
{
    return a8::Format("game%d_wsproxy", {App::Instance()->GetGameId()});
}

void App::Init()
{
    udp_msg_mutex_ = std::make_shared<std::mutex>();

    HandlerMgr::Instance()->Init();
    JsonDataMgr::Instance()->Init();
    DownStreamMgr::Instance()->Init();
    MasterMgr::Instance()->Init();
    UpStreamMgr::Instance()->Init();
    LongSessionMgr::Instance()->Init();
    GCListener::Instance()->Init();

    f8::UdpLog::Instance()->Info("wsproxy starting instance_id:%d pid:%d",
                                 {
                                  f8::App::Instance()->GetInstanceId(),
                                  f8::App::Instance()->GetPid(),
                                 });
    {
        int perf_log_time = 1000 * 60 * 5;
        if (getenv("is_dev_env")) {
            perf_log_time = 1000 * 10;
        }
        f8::Timer::Instance()->SetInterval
            (perf_log_time,
             [] (int event, const a8::Args* args)
             {
                 if (a8::TIMER_EXEC_EVENT == event) {
                     SavePerfLog();
                 }
             });
    }
    if (f8::App::Instance()->HasFlag(1)) {
        f8::Timer::Instance()->SetTimeout
            (
             1000 * 60,
             [] (int event, const a8::Args* args)
             {
                 if (a8::TIMER_EXEC_EVENT == event) {
                     f8::App::Instance()->Terminated();
                     f8::App::Instance()->NotifyLoopCond();
                 }
             }
             );
    }
}

void App::UnInit()
{
    a8::XPrintf("wsproxy terminating instance_id:%d pid:%d\n",
                {f8::App::Instance()->GetInstanceId(),
                 f8::App::Instance()->GetPid()});
    GCListener::Instance()->UnInit();
    LongSessionMgr::Instance()->UnInit();
    MasterMgr::Instance()->UnInit();
    UpStreamMgr::Instance()->UnInit();
    DownStreamMgr::Instance()->UnInit();
    JsonDataMgr::Instance()->UnInit();
    HandlerMgr::Instance()->UnInit();

    FreeUdpMsgQueue();
    udp_msg_mutex_ = nullptr;
    a8::XPrintf("wsproxy terminated instance_id:%d pid:%d\n",
                {f8::App::Instance()->GetInstanceId(),
                 f8::App::Instance()->GetPid()});
}

void App::Update(int delta_time)
{
    QuickExecute();
    SlowerExecute();
}

void App::QuickExecute()
{
    DispatchUdpMsg();
    LongSessionMgr::Instance()->Update();
}

void App::SlowerExecute()
{
}

bool App::HasTask()
{
    {
        if (!udp_work_node_) {
            udp_msg_mutex_->lock();
            if (!udp_work_node_ && udp_top_node_) {
                udp_work_node_ = udp_top_node_;
                udp_top_node_ = nullptr;
                udp_bot_node_ = nullptr;
            }
            udp_msg_mutex_->unlock();
        }
        if (udp_work_node_) {
            return true;
        }
    }
    return false;
}

void App::ProcessClientMsg(f8::MsgHdr* hdr)
{
    if (hdr->msgid == ss::_SS_CMLogin ||
        hdr->msgid == ss::_SS_CMReconnect ||
        hdr->msgid == ss::_SS_CMKcpHandshake) {
        auto down_wp = DownStreamMgr::Instance()->GetDownStream(hdr->socket_handle);
        if (down_wp.expired()) {
            switch (hdr->msgid) {
            case ss::_SS_CMLogin:
                {
                    ss::SS_CMLogin msg;
                    bool ok = msg.ParseFromArray(hdr->buf + hdr->offset, hdr->buflen - hdr->offset);
                    if (ok) {
                        MasterMgr::Instance()->RequestTargetServer
                            (hdr,
                             msg.team_uuid(),
                             msg.account_id(),
                             msg.session_id(),
                             "",
                             0,
                             msg.proto_version());
                    }
                }
                break;
            case ss::_SS_CMReconnect:
                {
                    ss::SS_CMReconnect msg;
                    bool ok = msg.ParseFromArray(hdr->buf + hdr->offset, hdr->buflen - hdr->offset);
                    if (ok) {
                        MasterMgr::Instance()->RequestTargetServer
                            (hdr,
                             msg.team_uuid(),
                             msg.account_id(),
                             msg.session_id(),
                             msg.server_info(),
                             1,
                             0);
                    }
                }
                break;
            case ss::_SS_CMKcpHandshake:
                {
                    ss::SS_CMKcpHandshake msg;
                    bool ok = msg.ParseFromArray(hdr->buf + hdr->offset, hdr->buflen - hdr->offset);
                    if (ok) {
                        LongSessionMgr::Instance()->_SS_CMKcpHandshake(hdr, msg);
                    }
                }
                break;
            default:
                {
                    abort();
                }
                break;
            }
        }
    } else {
        auto down_wp = DownStreamMgr::Instance()->GetDownStream(hdr->socket_handle);
        if (auto down = down_wp.lock(); !down_wp.expired()) {
            down->ProcCMMsg(hdr);
        }
    }
}

void App::ProcessMasterServerMsg(f8::MsgHdr* hdr)
{
    f8::NetMsgHandler* handler = f8::GetNetMsgHandler(&HandlerMgr::Instance()->msmsghandler,
                                                      hdr->msgid);
    if (handler) {
        switch (handler->handlerid) {
        case HID_MasterMgr:
            ProcessNetMsg(handler, MasterMgr::Instance(), hdr);
            break;
        }
    }
}

void App::ProcessTargetServerMsg(f8::MsgHdr* hdr)
{
    if (hdr->msgid == ss::_SS_ForceCloseSocket) {
        GCListener::Instance()->ForceCloseClient(hdr->socket_handle);
        return;
    }
    if (hdr->msgid == ss::_SS_CMLogin ||
        hdr->msgid == ss::_SS_CMReconnect ||
        hdr->msgid == ss::_SS_BindUpStreamSocket) {
        DownStreamMgr::Instance()->BindUpStream(hdr->socket_handle, hdr->ip_saddr);
        GCListener::Instance()->MarkClient(hdr->socket_handle, true);
    }
    if (hdr->msgid == ss::_SS_HttpTunnelResponse) {
        ss::SS_HttpTunnelResponse msg;
        bool ok = msg.ParseFromArray(hdr->buf + hdr->offset, hdr->buflen - hdr->offset);
        if (ok) {
            MasterMgr::Instance()->_SS_HttpTunnelResponse(hdr, msg);
        }
        return;
    }
    auto down_wp = DownStreamMgr::Instance()->GetDownStream(hdr->socket_handle);
    if (!down_wp.expired()) {
        down_wp.lock()->ForwardUpStreamMsg(hdr);
    }
}

void App::FreeUdpMsgQueue()
{
    udp_msg_mutex_->lock();
    if (!udp_work_node_) {
        udp_work_node_ = udp_top_node_;
        udp_top_node_ = nullptr;
        udp_bot_node_ = nullptr;
    }
    while (udp_work_node_) {
        UdpMsgNode* pdelnode = udp_work_node_;
        udp_work_node_ = udp_work_node_->next;
        {
            if (pdelnode->pkt->buf) {
                free((void*)pdelnode->pkt->buf);
            }
            delete pdelnode->pkt;
            free(pdelnode);
        }
        if (!udp_work_node_) {
            udp_work_node_ = udp_top_node_;
            udp_top_node_ = nullptr;
            udp_bot_node_ = nullptr;
        }
    }
    udp_msg_mutex_->unlock();
}

void App::AddUdpMsg(a8::UdpPacket* pkt)
{
    UdpMsgNode *p = (UdpMsgNode*) malloc(sizeof(UdpMsgNode));
    memset(p, 0, sizeof(UdpMsgNode));
    p->pkt = pkt;
    udp_msg_mutex_->lock();
    if (udp_bot_node_) {
        udp_bot_node_->next = p;
        udp_bot_node_ = p;
    } else {
        udp_top_node_ = p;
        udp_bot_node_ = p;
    }
    ++udp_msgnode_size_;
    udp_msg_mutex_->unlock();
    f8::App::Instance()->NotifyLoopCond();
}

void App::DispatchUdpMsg()
{
    long long starttick = a8::XGetTickCount();
    if (!udp_work_node_ && udp_top_node_) {
        udp_msg_mutex_->lock();
        udp_work_node_ = udp_top_node_;
        udp_top_node_ = nullptr;
        udp_bot_node_ = nullptr;
        udp_working_msgnode_size_ = udp_msgnode_size_;
        udp_msg_mutex_->unlock();
    }

    while (udp_work_node_) {
        UdpMsgNode *pdelnode = udp_work_node_;
        LongSessionMgr::Instance()->ProcUdpPacket(pdelnode->pkt);
        udp_work_node_ = pdelnode->next;
        {
            if (pdelnode->pkt->buf) {
                free((void*)pdelnode->pkt->buf);
            }
            delete pdelnode->pkt;
            free(pdelnode);
        }
        udp_working_msgnode_size_--;
        if (a8::XGetTickCount() - starttick > 200) {
            break;
        }
    }//end while

    if (!udp_work_node_) {
        udp_working_msgnode_size_ = 0;
    }
}

int App::GetGameId() const
{
#ifdef DEBUG
    std::string game_id = a8::XValue((f8::App::Instance()->GetArgv())[0]).GetString();
    std::vector<std::string> strings;
    a8::Split(game_id, strings, '/');
    game_id = strings.at(strings.size() - 1);
    a8::ReplaceString(game_id, "wsproxy", "");
    return a8::XValue(game_id);
#else
    return GAME_ID;
#endif
}

void App::DispatchSocketMsg(f8::MsgHdr* hdr)
{
    switch (hdr->sockfrom) {
    case SF_Client:
        {
            ProcessClientMsg(hdr);
        }
        break;
    case SF_TargetServer:
        {
            ProcessTargetServerMsg(hdr);
        }
        break;
    case SF_MasterServer:
        {
            ProcessMasterServerMsg(hdr);
        }
        break;
    default:
        {
            A8_ABORT();
        }
        break;
    }
}