1

f8/app.cc

@@ -0,0 +1,300 @@
+#include <signal.h>
+#include <unistd.h>
+#include <string.h>
+
+#include <mutex>
+#include <condition_variable>
+
+#include <a8/a8.h>
+#include <a8/uuid.h>
+#include <a8/perfmonitor.h>
+
+#include <f8/f8.h>
+#include <f8/udplog.h>
+#include <f8/app.h>
+#include <f8/msgqueue.h>
+#include <f8/timer.h>
+#include <f8/protoutils.h>
+#include <f8/tglog.h>
+#include <f8/httpclientpool.h>
+
+static const int MAX_ZONE_ID = 100;
+static const int MAX_NODE_ID = 8;
+static const int MAX_INSTANCE_ID = 500;
+
+static const int MAX_SYS_HTTP_NUM = 2;
+static const int MAX_USER_HTTP_NUM = 8;
+static const int MAX_ALL_HTTP_NUM = MAX_SYS_HTTP_NUM + MAX_USER_HTTP_NUM;
+
+static const char* const PROJ_ROOT_FMT = "/data/logs/%s";
+static const char* const PROJ_LOG_ROOT_FMT = "/data/logs/%s/logs";
+static const char* const PROJ_LOG_FILENAME_FMT = "log_$pid_%Y%m%d.log";
+
+namespace f8
+{
+
+    bool App::Init()
+    {
+        signal(SIGPIPE, SIG_IGN);
+        srand(time(nullptr));
+        nowtime_ = time(nullptr);
+
+        if (!ParseOpt()) {
+            exit_code_ = 1;
+            return false;
+        }
+
+        uuid_ = std::make_shared<a8::uuid::SnowFlake>();
+        loop_mutex_ = new std::mutex();
+        loop_cond_ = new std::condition_variable();
+
+        uuid_->SetMachineId((node_id_ - 1) * MAX_NODE_ID + instance_id_);
+        a8::PerfMonitor::Instance()->Init();
+        InitLog();
+        f8::MsgQueue::Instance()->Init();
+        f8::Timer::Instance()->Init();
+        f8::TGLog::Instance()->Init(user_app_->GetPkgName(), false, 0);
+        f8::HttpClientPool::Instance()->Init(MAX_ALL_HTTP_NUM, MAX_SYS_HTTP_NUM, MAX_USER_HTTP_NUM);
+        user_app_->Init();
+        return true;
+    }
+
+    void App::UnInit()
+    {
+        user_app_->UnInit();
+        {
+            queue_.Fetch();
+            list_head* work_list = queue_.GetWorkList();
+            while (!list_empty(work_list)){
+                MsgHdr* hdr = list_first_entry(work_list, MsgHdr, entry);
+                list_del_init(&hdr->entry);
+                MsgHdr::Destroy(hdr);
+            }
+        }
+        f8::TGLog::Instance()->UnInit();
+        f8::HttpClientPool::Instance()->UnInit();
+        f8::Timer::Instance()->UnInit();
+        f8::MsgQueue::Instance()->UnInit();
+        UnInitLog();
+        a8::PerfMonitor::Instance()->Init();
+
+        delete loop_cond_;
+        loop_cond_ = nullptr;
+        delete loop_mutex_;
+        loop_mutex_ = nullptr;
+    }
+
+    int App::Run(int argc, char* argv[], UserApp* user_app)
+    {
+        argc_ = argc;
+        argv_ = argv;
+        user_app_ = user_app;
+        int delta_time = 0;
+        if (Init()) {
+            a8::tick_t last_stat_tick = a8::XGetTickCount();
+            while (!Terminated()) {
+                a8::tick_t begin_tick = a8::XGetTickCount();
+                nowtime_ = time(nullptr);
+                f8::Timer::Instance()->Update();
+                f8::MsgQueue::Instance()->Update();
+                DispatchNetMsg();
+                user_app->Update(delta_time);
+                a8::tick_t end_tick = a8::XGetTickCount();
+                if (end_tick - begin_tick > max_run_delay_time_) {
+                    max_run_delay_time_ = end_tick - begin_tick;
+                }
+                Schedule();
+                end_tick = a8::XGetTickCount();
+                if (end_tick - last_stat_tick > 0) {
+                    delta_time = end_tick - last_stat_tick;
+                    last_stat_tick = end_tick;
+                } else {
+                    delta_time = 0;
+                }
+            }
+            UnInit();
+        }
+        return exit_code_;
+    }
+
+    bool App::ParseOpt()
+    {
+        int ch = 0;
+        while ((ch = getopt(argc_, argv_, "z:n:i:f:")) != -1) {
+            switch (ch) {
+            case 'z':
+                {
+                    zone_id_ = a8::XValue(optarg);
+                }
+                break;
+            case 'n':
+                {
+                    node_id_ = a8::XValue(optarg);
+                }
+                break;
+            case 'i':
+                {
+                    instance_id_ = a8::XValue(optarg);
+                }
+                break;
+            case 'f':
+                {
+                    std::vector<std::string> strings;
+                    a8::Split(optarg, strings, ',');
+                    for (auto& str : strings) {
+                        flags_.insert(a8::XValue(str).GetInt());
+                    }
+                }
+                break;
+            }
+        }
+        if (zone_id_ <= 0) {
+            a8::XPrintf("启动失败,缺少-z参数\n", {});
+            return false;
+        } else if (node_id_ > MAX_ZONE_ID) {
+            a8::XPrintf("启动失败,-z参数不能大于%d\n", {MAX_ZONE_ID});
+            return false;
+        } else if (node_id_ <= 0) {
+            a8::XPrintf("启动失败,缺少-n参数\n", {});
+            return false;
+        } else if (node_id_ > MAX_NODE_ID) {
+            a8::XPrintf("启动失败,-n参数不能大于%d\n", {MAX_NODE_ID});
+            return false;
+        } else if (instance_id_ <= 0) {
+            a8::XPrintf("启动失败,缺少-i参数\n", {});
+            return false;
+        } else if (instance_id_ > MAX_INSTANCE_ID) {
+            a8::XPrintf("启动失败,-i参数不能大于%d\n", {MAX_INSTANCE_ID});
+            return  false;
+        }
+        a8::XPrintf("starting zone_id:%d node_id:%d instance_id:%d pid:%d\n",
+                    {
+                     zone_id_,
+                     node_id_,
+                     instance_id_,
+                     GetPid()
+                    });
+        return true;
+    }
+
+    bool App::HasFlag(int flag)
+    {
+        return flags_.find(flag) != flags_.end();
+    }
+
+    void App::NotifyLoopCond()
+    {
+        std::unique_lock<std::mutex> lk(*loop_mutex_);
+        loop_cond_->notify_all();
+    }
+
+    long long App::NewNodeUuid()
+    {
+        return uuid_->Generate();
+    }
+
+    const std::string App::NewGlobalUuid()
+    {
+        std::string id = a8::Format("%d%d", {100 + zone_id_, uuid_->Generate()});
+        return id;
+    }
+
+    int App::GetPid()
+    {
+        return getpid();
+    }
+
+    void App::InitLog()
+    {
+        std::string filename_fmt = PROJ_LOG_FILENAME_FMT;
+        a8::ReplaceString(filename_fmt, "$pid",
+                          a8::XValue(f8::App::Instance()->GetPid()));
+
+        std::string proj_root_dir = a8::Format
+            (PROJ_ROOT_FMT, {user_app_->GetPkgName()});
+        std::string proj_log_root_dir = a8::Format
+            (PROJ_LOG_ROOT_FMT, {user_app_->GetPkgName()});
+        std::string log_file_name = a8::Format
+            (PROJ_LOG_ROOT_FMT, {user_app_->GetPkgName()}) + "/" + filename_fmt;
+
+        a8::MkDir(proj_root_dir);
+        a8::MkDir(proj_log_root_dir);
+        a8::XPrintf("log_file_name:%s\n", {log_file_name});
+        f8::UdpLog::Instance()->SetLogFileName(log_file_name);
+        f8::UdpLog::Instance()->Init();
+        f8::UdpLog::Instance()->Info("proj_root_dir:%s", {proj_root_dir});
+        f8::UdpLog::Instance()->Info("proj_log_root_dir:%s", {proj_log_root_dir});
+        f8::UdpLog::Instance()->Info("log_file_name:%s", {log_file_name});
+    }
+
+    void App::UnInitLog()
+    {
+        f8::UdpLog::Instance()->UnInit();
+    }
+
+    void App::Schedule()
+    {
+        std::unique_lock<std::mutex> lk(*loop_mutex_);
+        bool has_task = false;
+        {
+            queue_.Fetch();
+            if (!list_empty(queue_.GetWorkList())) {
+                has_task = true;
+            }
+        }
+        if (!has_task) {
+            has_task = user_app_->HasTask();
+        }
+        if (!has_task) {
+            int sleep_time = f8::Timer::Instance()->GetIdleTime();
+            loop_cond_->wait_for(lk, std::chrono::milliseconds(sleep_time));
+        } else {
+            int sleep_time = 1;
+            loop_cond_->wait_for(lk, std::chrono::milliseconds(sleep_time));
+        }
+    }
+
+    void App::DispatchNetMsg()
+    {
+        queue_.Fetch();
+        list_head* work_list = queue_.GetWorkList();
+        while (!list_empty(work_list)){
+            MsgHdr* hdr = list_first_entry(work_list, MsgHdr, entry);
+            list_del_init(&hdr->entry);
+            user_app_->DispatchSocketMsg(hdr);
+            --msgnode_size_;
+            MsgHdr::Destroy(hdr);
+        }
+    }
+
+    void App::AddSocketMsg(int sockfrom,
+                           int sockhandle,
+                           long ip_saddr,
+                           unsigned short msgid,
+                           unsigned int seqid,
+                           const char *msgbody,
+                           int bodylen,
+                           int tag)
+    {
+        char *p = (char*)malloc(sizeof(MsgHdr) + bodylen);
+        MsgHdr* hdr = (MsgHdr*)p;
+        hdr->sockfrom = sockfrom;
+        hdr->seqid = seqid;
+        hdr->msgid = msgid;
+        hdr->socket_handle = sockhandle;
+        hdr->ip_saddr = ip_saddr;
+        hdr->buf = p + sizeof(MsgHdr);
+        hdr->buflen = bodylen;
+        hdr->offset = 0;
+        hdr->hum = nullptr;
+        hdr->user_data = nullptr;
+        hdr->tag = tag;
+        if (bodylen > 0) {
+            memmove((void*)hdr->buf, msgbody, bodylen);
+        }
+        ++msgnode_size_;
+        queue_.Push(&hdr->entry);
+        NotifyLoopCond();
+    }
+
+}

f8/app.h

@@ -0,0 +1,99 @@
+#pragma once
+
+#include <atomic>
+
+#include <a8/singleton.h>
+#include <a8/queue.h>
+
+namespace a8
+{
+    namespace uuid
+    {
+        class SnowFlake;
+    }
+}
+
+namespace f8
+{
+
+    class UserApp
+    {
+    public:
+        virtual const std::string GetPkgName() = 0;
+        virtual void Init() = 0;
+        virtual void UnInit() = 0;
+        virtual void Update(int delta_time) = 0;
+        virtual bool HasTask() = 0;
+        virtual void DispatchSocketMsg(MsgHdr* hdr) = 0;
+    };
+
+    class App : public a8::Singleton<App>
+    {
+    private:
+        App() {};
+        friend class a8::Singleton<App>;
+
+    public:
+
+        int Run(int argc, char* argv[], UserApp* user_app);
+
+        const std::string GetPkgName();
+        void NotifyLoopCond();
+        bool HasFlag(int flag);
+        long long NewNodeUuid();
+        const std::string NewGlobalUuid();
+        int GetZoneId() { return zone_id_; }
+        int GetNodeId() { return node_id_; }
+        int GetInstanceId() { return instance_id_; }
+        int GetNowTime() { return nowtime_; };
+        int GetPid();
+        auto Terminated() { return terminated_; };
+        void Terminate() { terminated_ = true; };
+        void AddSocketMsg(int sockfrom,
+                          int sockhandle,
+                          long ip_saddr,
+                          unsigned short msgid,
+                          unsigned int seqid,
+                          const char *msgbody,
+                          int bodylen,
+                          int tag);
+        char** GetArgv() { return argv_; }
+        int GetArgc() { return argc_; }
+        long long GetMsgNodeSize() { return msgnode_size_; }
+        long long GetWorkingMsgNodeSize() { return working_msgnode_size_; }
+        long long GetMaxRunDelayTime() { return max_run_delay_time_; }
+        void ResetMaxRunDelayTime() { max_run_delay_time_ = 0; }
+
+    private:
+        bool Init();
+        void UnInit();
+        bool ParseOpt();
+        void InitLog();
+        void UnInitLog();
+        void Schedule();
+        void DispatchNetMsg();
+
+    private:
+        UserApp* user_app_ = nullptr;
+        int argc_ = 0;
+        char** argv_ = nullptr;
+        int exit_code_ = 0;
+        volatile bool terminated_ = false;
+        int nowtime_ = 0;
+        long long max_run_delay_time_ = 0;
+
+        int zone_id_ = 0;
+        int node_id_ = 0;
+        int instance_id_ = 0;
+        std::set<int> flags_;
+
+        a8::Queue queue_;
+        std::atomic<long long> msgnode_size_ = {0};
+        std::atomic<long long> working_msgnode_size_ = {0};
+
+        std::shared_ptr<a8::uuid::SnowFlake> uuid_;
+        std::mutex *loop_mutex_ = nullptr;
+        std::condition_variable *loop_cond_ = nullptr;
+    };
+
+}

f8/jsonhttprequest.cc

@@ -9,24 +9,38 @@
 namespace f8
 {
 
-    JsonHttpRequest::JsonHttpRequest()
+    JsonHttpRequest::JsonHttpRequest(unsigned long saddr,
+                                     const std::string url,
+                                     const std::string& query_str,
+                                     a8::CommonCbProc cb)
     {
-        params = std::make_shared<a8::XObject>();
-        resp_xobj = a8::MutableXObject::CreateObject();
+        saddr_ = saddr;
+        url_ = url;
+        query_str_ = query_str;
+        cb_ = cb;
+        params_ = std::make_shared<a8::XObject>();
+        resp_xobj_ = a8::MutableXObject::CreateObject();
+
+        params_->ReadFromUrlQueryString(query_str);
+        GetResp()->SetVal("errcode", 0);
+        GetResp()->SetVal("errmsg", "");
     }
 
     JsonHttpRequest::~JsonHttpRequest()
     {
-        if (context && free_context)  {
-            free_context(context);
+        Response();
+    }
+
+    void JsonHttpRequest::Response()
+    {
+        if (!resped_) {
+            if (cb_) {
+                std::string response;
+                resp_xobj_->ToJsonStr(response);
+                cb_(a8::Args({a8::HttpResponse(response)}));
+            }
+            resped_ = true;
         }
     }
 
-    std::string JsonHttpRequest::Response()
-    {
-        std::string response;
-        resp_xobj->ToJsonStr(response);
-        return a8::HttpResponse(response);
-    }
-
 }

f8/jsonhttprequest.h

@@ -9,24 +9,28 @@ namespace a8
 namespace f8
 {
 
-    struct JsonHttpRequest
+    class JsonHttpRequest
     {
-        bool pending = false;
-        unsigned long saddr = 0;
-        int socket_handle = 0;
-        time_t create_time = 0;
-        time_t handle_time = 0;
-        std::string query_str;
-        std::shared_ptr<a8::XObject> params;
-        std::shared_ptr<a8::MutableXObject> resp_xobj;
+    private:
+        unsigned long saddr_ = 0;
+        std::string url_;
+        std::string query_str_;
+        std::shared_ptr<a8::XObject> params_;
+        std::shared_ptr<a8::MutableXObject> resp_xobj_;
+        a8::CommonCbProc cb_;
+        bool resped_ = false;
 
-        int async_pending_count = 0;
-        void* context = nullptr;
-        void (*free_context)(void*) = nullptr;
-
-        JsonHttpRequest();
+    public:
+        JsonHttpRequest(unsigned long saddr,
+                        const std::string url,
+                        const std::string& query_str,
+                        a8::CommonCbProc cb);
         ~JsonHttpRequest();
-        std::string Response();
+
+        const std::string GetUrl() { return this->url_; }
+        std::shared_ptr<a8::XObject> GetParams() { return this->params_; }
+        std::shared_ptr<a8::MutableXObject> GetResp() { return this->resp_xobj_; }
+        void Response();
     };
 
 }

f8/netmsghandler.h

@@ -15,7 +15,7 @@ namespace f8
     const unsigned short MAX_MSG_ID = 2000;
 
     struct MsgHdr;
-    typedef bool (*CUSTOM_PARSER)(MsgHdr&, google::protobuf::Message*);
+    typedef bool (*CUSTOM_PARSER)(MsgHdr*, google::protobuf::Message*);
 
     struct NetMsgHandler
     {
@@ -45,21 +45,21 @@ namespace f8
     };
 
     template<typename InstanceType, typename MsgType>
-    static void NetMsgHandlerWrapper(InstanceType* instance, MsgHdr& hdr, NetMsgHandler* handler)
+    static void NetMsgHandlerWrapper(InstanceType* instance, MsgHdr* hdr, NetMsgHandler* handler)
     {
         MsgType msg;
         bool ok = false;
         if (handler->custom_parser) {
             ok = handler->custom_parser(hdr, &msg);
         } else {
-            ok = msg.ParseFromArray(hdr.buf + hdr.offset, hdr.buflen - hdr.offset);
+            ok = msg.ParseFromArray(hdr->buf + hdr->offset, hdr->buflen - hdr->offset);
         }
         assert(ok);
         if (ok) {
             struct Invoker: public NetMsgHandler
             {
-                void (*wrapper)(InstanceType*, MsgHdr&, NetMsgHandler*);
-                void (InstanceType::*func)(MsgHdr&, const MsgType&);
+                void (*wrapper)(InstanceType*, MsgHdr*, NetMsgHandler*);
+                void (InstanceType::*func)(MsgHdr*, const MsgType&);
             };
             Invoker* invoker = (Invoker*)handler;
             auto ptr = invoker->func;
@@ -69,7 +69,7 @@ namespace f8
 
     template<typename InstanceType, typename MsgType>
     static void RegisterNetMsgHandler(NetMsgHandlerObject* msghandler,
-                                      void (InstanceType::*ptr)(MsgHdr&, const MsgType&),
+                                      void (InstanceType::*ptr)(MsgHdr*, const MsgType&),
                                       CUSTOM_PARSER custom_parser = nullptr
                                       )
     {
@@ -81,8 +81,8 @@ namespace f8
         }
         struct Invoker: public NetMsgHandler
         {
-            void (*wrapper)(InstanceType*, MsgHdr&, NetMsgHandler*);
-            void (InstanceType::*func)(MsgHdr&, const MsgType&);
+            void (*wrapper)(InstanceType*, MsgHdr*, NetMsgHandler*);
+            void (InstanceType::*func)(MsgHdr*, const MsgType&);
         };
         Invoker *p = new Invoker();
         p->wrapper = &NetMsgHandlerWrapper<InstanceType, MsgType>;
@@ -96,12 +96,12 @@ namespace f8
     template<typename InstanceType>
     static bool ProcessNetMsg(NetMsgHandler* handler,
                               InstanceType* instance,
-                              MsgHdr& hdr)
+                              MsgHdr* hdr)
     {
         if(handler){
             struct Invoker: public NetMsgHandler
             {
-                void (*wrapper)(InstanceType*, MsgHdr&, NetMsgHandler*);
+                void (*wrapper)(InstanceType*, MsgHdr*, NetMsgHandler*);
                 void* func;
             };
             Invoker *p = (Invoker*)handler;
@@ -115,6 +115,6 @@ namespace f8
     NetMsgHandler* GetNetMsgHandler(NetMsgHandlerObject* handlers,
                                     unsigned short msgid);
 
-    void DumpMsgToLog(f8::MsgHdr& hdr, f8::NetMsgHandler* handler, const char* prefix);
+    void DumpMsgToLog(f8::MsgHdr* hdr, f8::NetMsgHandler* handler, const char* prefix);
     void DumpMsgToLog(const ::google::protobuf::Message& msg, const char* prefix);
 }

f8/protoutils.cc

@@ -306,20 +306,14 @@ namespace f8
 
     MsgHdr* MsgHdr::Clone()
     {
-        MsgHdr* hdr = (MsgHdr*)malloc(sizeof(MsgHdr));
-        *hdr = *this;
-        if (hdr->buflen > 0) {
-            hdr->buf = (char*)malloc(hdr->buflen);
-            memmove((void*)hdr->buf, (void*)buf, buflen);
-        }
+        MsgHdr* hdr = (MsgHdr*)malloc(sizeof(MsgHdr) + buflen);
+        memmove((void*)hdr, (void*)this, sizeof(MsgHdr) + buflen);
+        hdr->buf = ((char*)hdr) + sizeof(MsgHdr);
         return hdr;
     }
 
     void MsgHdr::Destroy(MsgHdr* hdr)
     {
-        if (hdr->buf) {
-            free((void*)hdr->buf);
-        }
         free((void*)hdr);
     }
 

f8/protoutils.h

@@ -13,6 +13,7 @@ namespace f8
 
     struct MsgHdr
     {
+        int sockfrom;
         unsigned int seqid;
         unsigned short msgid;
         int socket_handle;
@@ -22,6 +23,8 @@ namespace f8
         int offset;
         Player *hum = nullptr;
         const void* user_data = nullptr;
+        int tag;
+        list_head entry;
 
         MsgHdr* Clone();
         static void Destroy(MsgHdr* hdr);

f8/timer.cc

@@ -115,4 +115,9 @@ namespace f8
         return xtimer_.IsRunning();
     }
 
+    long long Timer::GetIdleTime()
+    {
+        return xtimer_.GetIdleTime();
+    }
+
 }

f8/timer.h

@@ -58,6 +58,7 @@ namespace f8
         long long GetRemainTime(TimerWp& timer_wp);
         void DeleteCurrentTimer();
         bool IsRunning();
+        long long GetIdleTime();
 
     private:
         bool initialized_ = false;

f8/udplog.cc

@@ -40,7 +40,7 @@ namespace f8
         impl_->save_cond = new std::condition_variable();
 
         impl_->log_level = 0;
-        #ifdef DEBUG
+        #ifdef MYDEBUG
         impl_->debuging = true;
         #endif
         impl_->top_node = nullptr;