【负载均衡oj】(七)ojserver

一.目的

负责负载均衡，获取网站首页，通过网络罗调用编译并运行并提供结果给用户。根据用户的操作提供不同的功能。采用mvc结构。使用 ctemplate文字模板做数据渲染

m在这里指的是和数据的交互，可以用文件或者sql交互。在使用文件的情况下，就必须要有对应的文件目录存放每一道题。提供题目描述和题头还有测试用例。

二.实现model

负责将文件题库抽象成数据结构，并提供接口给ojcontrol调用。ojcontrol通过model模块获取全部的题目信息和测试用例，用来交给后端服务继续运行

题的数据结构

  typedef struct Question // 每一道题对应的基本信息
    {
        string number;    // 题目编号
        string title;     // 题目名称
        string diffculty; // 题目难度
        int cpu_limit;    // 运行时间限制
        int mem_limit;    // 内存时间限制
        string desc;      // 题目描述
        string header;    // 题目提供给用户的首部代码
        string tailer;    // 题目的测试用例，需要和header拼接，形成完整代码
    }Question;

获取题信息的数据结构，因为是从文件中读取，所以需要一个字符串格分割函数。

加载所有题库信息到数据结构。

 bool LoadQuestionList(const string question_path)   //根据文件题目列表获取到数据结构内
        {
            ifstream in(questionlist_path); // 打开题目列表的文件流

            if (!in.is_open()) // 打开文件流失败
            {
                LOG(FATAL) << "加载题库失败,请检查是否存在题库文件"
                           << "\n";
                return false;
            }

            string line;
            while (getline(in, line))
            {
                vector<string> tokens;                       // 题的五个信息
                StringUtil::SplitString(line, &tokens, " "); // 根据空格分隔出不同元素

                if (tokens.size() != 5) // 每道题有五种基本信息（编号，题目，,难度,mem,cpu）
                {
                    LOG(WARNING) << "加载部分题目失败, 请检查文件格式"
                                 << "\n";
                    continue; // 获取当前题目信息失败直接跳过
                }

                Question q; // 创建题目对象填充信息

                q.number = tokens[0];
                q.title = tokens[1];
                q.diffculty = tokens[2];
                q.cpu_limit = atoi(tokens[3].c_str());
                q.mem_limit = atoi(tokens[4].c_str());

                // 获取题目详细信息
                string path = question_path; // 需要拼出对应的题目路径
                path += q.number;
                path += "/";

                // 从文件内读出内容并填充进题目结构体里
                FileUtil::ReadFile(path + "desc.txt", &(q.desc), true);
                FileUtil::ReadFile(path + "header.cpp", &(q.header), true);
                FileUtil::ReadFile(path + "tail.cpp", &(q.tailer), true);

                // 形成哈希映射
                questions.insert({q.number, q});
            }
            LOG(INFO) << "题库加载成功"
                      << "\n";
            in.close();

            return true;
        }

获取一道题给客户

 bool GetOneQuestion(const string &number, Question *ret) // 通过题号获取对应题
        {
            const auto &iter = questions.find(number);
            if (iter == questions.end())
            {
                LOG(ERROR) << "用户获取题目失败, 题目编号: " << number << "\n";
                return false;
            }
            *ret = iter->second; // 输出行参数
            return true;
        }

获取所有题目给客户

bool GetAllQuestion(vector<Question> *out)
        {
            if (questions.size() == 0)
            {
                LOG(ERROR) << "用户获取题库失败"
                           << "\n";
                return false;
            }
            for (auto &q : questions)
            {
                out->push_back(q.second); // 遍历哈希映射把题目信息依次录入vector
            }
            return true;
        }

三.control模块

主要实现逻辑控制，从网页上拿来各种信息，提取有用信息后，结合本地数据向后端提交，如判题功能，或者根据要求获取对应本地信息，并通过前段模块返回给用户

 // 核心业务逻辑
    class Control // 控制ojserver的基础功能，包括修改题库和前端界面交互，整合数据控制和前端交互
    {
    private:
        ns_model::Model _model; // 提供后台数据
        View _view;             // 提供html渲染功能
        LoadBalance _load;      // 提供负载均衡模块
    public:
        void RecoveryMachine()
        {
            _load.OnlineMachine();
        }
        // 获取所有题目并生成html文件
        bool AllQuestions(string *html) // 输出型参数
        {
            vector<Question> all;
            if (_model.GetAllQuestion(&all)) // 获取所有题目到vector内
            {
                // 获取信息成功构建html
                sort(all.begin(),all.end(),[](const Question& q1,const Question& q2){ 
                    return atoi(q1.number.c_str())<atoi(q2.number.c_str());
                });
                _view.AllExpandHtml(all, html);
            }
            else
            {
                *html = "获取题目失败, 形成题目列表失败";
                return false;
            }
            return true;
        }

        // 获取一道题
        bool OneQuestion(const string &number, string *html)
        {
            Question q;

            if (_model.GetOneQuestion(number, &q))
            {
                _view.OneExepandHtml(q, html);
                return true;
            }
            else
            {
                *html = "指定题目: " + number + " 不存在!";
                return false;
            }
        }

        // 判断用户代码是否正确
        void Judge(const string &in_json, string *outj_son, const string &number)
        {
            // 需要读取上来用户的代码并和测试用例拼接成一份完整代码最后
            // 0. 根据题目编号，直接拿到对应的题目细节
            Question q;
            _model.GetOneQuestion(number, &q);

            // 1. in_json进行反序列化，得到题目的id，得到用户提交源代码，input
            Json::Reader reader;
            Json::Value usr_val;
            reader.parse(in_json, usr_val);
            string code = usr_val["code"].asString(); // 用户写的代码

            // 2. 重新拼接用户代码+测试用例代码，形成新的代码
            Json::Value compile_val;
            compile_val["code"] = code + '\n' + q.tailer;
            compile_val["input"] = usr_val["input"].asString();
            compile_val["cpu_limit"] = q.cpu_limit;
            compile_val["mem_limit"] = q.mem_limit;

            // 序列化准备传输
            Json::FastWriter writer;
            std::string compile_string = writer.write(compile_val);
            // 3. 选择负载最低的主机(差错处理)
            // 规则: 一直选择，直到主机可用，否则，就是全部挂掉
            while (true)
            {
                int id;
                Machine *m;
                // 跟据负载选择服务器
                if (!_load.SmartChoice(&m, &id))
                {
                    break; // 选择主机失败
                }
                // 4. 选择完主机后发起http请求链接到主机
                Client cli(m->_ip, m->_port);
                m->IncLoad();
                LOG(INFO) << " 选择主机成功, 主机id: " << id << " 详情: " << m->_ip << ":" << m->_port << " 当前主机的负载是: " << m->Load() << "\n";

                // 5.将数据提交给服务主机并进行编译运行
                if (auto res = cli.Post("/Compile_and_run", compile_string, "application/json;charset=utf-8")) // 这个请求完成编译运行并返回结果
                {
                    // 需要判断返回结果，有返回结果不一定成功运行
                    if (res->status == 200) // 200表示成功运行
                    {
                        // 将结果返回给用户
                        *outj_son = res->body;
                        m->DecLoad();
                        LOG(INFO) << "主机执行任务成功"
                                  << "\n";
                        break;
                    }
                    // 若提交失败也需要将负载复原并重新执行该流程
                    m->DecLoad();
                }
                else
                {
                    // 请求失败(当前主机不存在)
                    LOG(ERROR) << " 当前请求的主机id: " << id << " 详情: " << m->_ip << ":" << m->_port << " 可能已经离线"
                               << "\n";
                    _load.OfflineMachine(id);

                    _load.ShowMachines(); // 仅仅是为了用来调试
                }
            }
        }

四.view模块

在拿到model的题目信息后，结合本地html进行渲染，给用户提供前端展示页面。采用ctemplate渲染。

通过互取到的所有题目信息，使用ctemplate渲染展示给用户。

// 获取所有题目并生成html文件
        bool AllQuestions(string *html) // 输出型参数
        {
            vector<Question> all;
            if (_model.GetAllQuestion(&all)) // 获取所有题目到vector内
            {
                // 获取信息成功构建html
                sort(all.begin(),all.end(),[](const Question& q1,const Question& q2){ 
                    return atoi(q1.number.c_str())<atoi(q2.number.c_str());
                });
                _view.AllExpandHtml(all, html);
            }
            else
            {
                *html = "获取题目失败, 形成题目列表失败";
                return false;
            }
            return true;
        }

一道题

 bool OneQuestion(const string &number, string *html)
        {
            Question q;

            if (_model.GetOneQuestion(number, &q))
            {
                _view.OneExepandHtml(q, html);
                return true;
            }
            else
            {
                *html = "指定题目: " + number + " 不存在!";
                return false;
            }
        }

五.负载均衡

我们需要以一个文件存放所有可用的主机和端口号。作为主机的配置文件。同时主机需要保存自身的负载情况，所以需要加锁。

class Machine // 标识提供服务的主机
    {
    public:
        std::string _ip; // 本机的ip端口号和负载程度
        int _port;
        uint64_t _load;
        std::mutex *_mtx; // 重点必须用指针不能用实例，容器内有拷贝，c++里的mutex是禁止拷贝的，只能用指针取地址绕过

        Machine()
            : _ip(""),
              _port(0),
              _load(0),
              _mtx(nullptr)
        {
        }

        // 提升负载(有可能多个主机同时运行有竞争问题)
        void IncLoad()
        {
            if (_mtx)
            {
                _mtx->lock();
            }
            _load++;
            if (_mtx)
            {
                _mtx->unlock();
            }
        }

        // 降低负载
        void DecLoad()
        {
            if (_mtx)
            {
                _mtx->lock();
            }
            _load--;
            if (_mtx)
            {
                _mtx->unlock();
            }
        }

        // 获取主机负载,没有太大的意义，只是为了统一接口
        uint64_t Load()
        {
            uint64_t load = 0;
            if (_mtx)
                _mtx->lock();
            load = _load;
            if (_mtx)
                _mtx->unlock();

            return load;
        }

        void ResetLoad()
        {
            if (_mtx)
            {
                _mtx->lock();
            }
            _load = 0;
            if (_mtx)
            {
                _mtx->unlock();
            }
        }
    };

负载均衡模块负责所有主机的情况，包括是否上线。并提供负载最小的主机，也需要加锁，因为涉及到对所有主机属性的更改。算临界区

class LoadBalance // 负载均衡模块
    {
        // 可以给我们提供编译服务的所有的主机
        // 每一台主机都有自己的下标，充当当前主机的id
        vector<Machine> machines;
        // 所有在线的主机id
        std::vector<int> online;
        // 所有离线的主机id
        std::vector<int> offline;
        std::mutex mtx; //需要保证负载均衡的数据安全

    public:
        LoadBalance()
        {
            assert(LoadConf(service_machine));
            LOG(INFO) << "加载 " << service_machine << " 成功"
                      << "\n";
        }

        bool LoadConf(const string &machine_conf) // 从conf文件里读取上来所有主机信息
        {
            ifstream in(machine_conf); // 打开文件流
            if (!in.is_open())
            {
                LOG(FATAL) << " 加载: " << machine_conf << " 失败"
                           << "\n";
                return false;
            }
            // 读取数据到line
            string line;
            while (getline(in, line))
            {
                vector<string> tokens; // 每一行切分到这里，只有两个元素，ip和端口号
                StringUtil::SplitString(line, &tokens, ":");

                if (tokens.size() != 2)
                {
                    LOG(WARNING) << " 切分 " << line << " 失败"
                                 << "\n";
                    continue;
                }
                // 读取数据完毕构建对象
                Machine m;
                m._ip = tokens[0]; // 填充ip和端口号
                m._port = atoi(tokens[1].c_str());

                m._load = 0;
                m._mtx = new std::mutex();

                online.push_back(machines.size()); // 先让所有机器数作为下标，在把主机放入主机列表
                machines.push_back(m);             // 抽象的哈希映射
            }
            in.close();
            return true;
        }

        bool SmartChoice(Machine **m, int *id) // 两个输出出行参数，返回选择的主机，或者看情况离线主机
        {                                      // 因为传参的时候不想通过下标访问，所以通过地址修改
            // 1. 选择主机(更新该主机的负载)
            // 2. 我们需要可能离线该主机

            // 选择主机有安全问题
            mtx.lock();
            int online_num = online.size(); // 检查活跃主机数
            if (online_num == 0)
            {
                mtx.unlock();
                LOG(FATAL) << " 所有的后端编译主机已经离线, 请尽快查看"
                           << "\n";
                return false;
            }
            // 通过遍历的方式，找到所有负载最小的机器
            *id = online[0];
            *m = &machines[online[0]];
            uint64_t min_load = machines[online[0]].Load();

            for (int i = 0; i < online_num; i++) // 寻找最小负载
            {
                uint64_t cur_load = machines[online[i]].Load();
                if (cur_load < min_load)
                {
                    min_load = cur_load;
                    *id = online[i];
                    *m = &machines[online[i]];
                }
            }
            mtx.unlock();
            return true;
        }

        void OnlineMachine() // 上线服务器(一次直接全部上线)
        {
            mtx.lock();
            online.insert(online.end(), offline.begin(), offline.end());
            offline.erase(offline.begin(), offline.end());
            LOG(INFO) << "重新登陆主机成功"
                      << "\n";
            mtx.unlock();
        }

        void ShowMachines() // 查看所有服务器状态
        {
            mtx.lock();
            std::cout << "当前在线主机列表: "
                      << "\n";
            for (auto &id : online)
            {
                std::cout << id << " ";
            }
            std::cout << std::endl;

            std::cout << "当前离线主机列表: "
                      << "\n";
            for (auto &id : offline)
            {
                std::cout << id << " ";
            }
            mtx.unlock();
        }

        void OfflineMachine(const int &mid) // 下线服务器
        {
            mtx.lock();
            for (auto iter = online.begin(); iter != online.end(); iter++) // 用迭代器遍历好一些，利于删除
            {
                if (*iter == mid)
                {
                    machines[mid].ResetLoad(); // 下线前清空负载
                    online.erase(iter);
                    offline.push_back(mid);

                    break; // 因为break所以不用考虑迭代器失效，此时循环已经终止了
                }
            }
            mtx.unlock();
        }
    };