diff --git a/src/component/http.cpp b/src/component/http.cpp index c2e4615..304d8b3 100644 --- a/src/component/http.cpp +++ b/src/component/http.cpp @@ -9,78 +9,140 @@ #include "scripting.hpp" #include +#include #include namespace http { - std::unordered_map active_requests{}; - uint64_t request_id{}; + namespace + { + constexpr const auto max_result_size = 0x5000u; + + struct http_request_params_t + { + std::string url; + std::string fields; + std::unordered_map headers; + }; + + struct http_request_t + { + http_request_params_t params; + scripting::object handle; + std::optional result; + std::atomic_bool completed; + }; + + std::vector> requests; + + void notify_request_result(std::shared_ptr& request) + { + const auto handle_id = request->handle.get_entity_id(); + + if (!request->result.has_value()) + { + scripting::notify(handle_id, "done", {{}, false, "unknown error"}); + return; + } + + auto& result = request->result.value(); + const auto error = curl_easy_strerror(result.code); + + if (result.code != CURLE_OK) + { + scripting::notify(handle_id, "done", {{}, false, error}); + return; + } + + if (result.buffer.size() >= max_result_size) + { + printf("^3WARNING: http result size bigger than %i bytes (%i), truncating!", max_result_size, + static_cast(result.buffer.size())); + result.buffer.resize(max_result_size); + } + + scripting::notify(handle_id, "done", {result.buffer, true}); + } + + void check_requests() + { + for (auto i = requests.begin(); i != requests.end(); ) + { + auto& request = *i; + if (!request->completed) + { + ++i; + continue; + } + else + { + notify_request_result(request); + i = requests.erase(i); + } + } + } + + scripting::object create_request(const http_request_params_t& params) + { + const auto request = std::make_shared(); + requests.emplace_back(request); + + request->params = params; + scheduler::thread_pool.push([request] + { + request->result = utils::http::get_data( + request->params.url, request->params.fields, request->params.headers); + request->completed = true; + }); + + return request->handle; + } + + template + void push_request(T&& r) + { + const auto request = std::make_shared(std::forward(r)); + requests.emplace_back(request); + } + + void wait_and_clear_requests() + { + for (auto& task : requests) + { + while (!task->completed) + { + std::this_thread::sleep_for(1ms); + } + } + + requests.clear(); + } + } class component final : public component_interface { public: + void on_shutdown([[maybe_unused]] plugin::plugin* plugin) override + { + scripting::on_shutdown(wait_and_clear_requests); + } + void on_startup([[maybe_unused]] plugin::plugin* plugin) override { - scripting::on_shutdown([]() - { - active_requests.clear(); - }); + scheduler::loop(check_requests, scheduler::server); + scripting::on_shutdown(wait_and_clear_requests); gsc::function::add_multiple([](const std::string& url) { - const auto id = request_id++; - active_requests[id] = true; - - const auto object = scripting::object{}; - const auto object_id = object.get_entity_id(); - - scheduler::once([id, object_id, url]() - { - const auto data = utils::http::get_data(url); - scheduler::once([id, object_id, data]() - { - if (active_requests.find(id) == active_requests.end()) - { - return; - } - - if (!data.has_value()) - { - scripting::notify(object_id, "done", {{}, false, "Unknown error"}); - return; - } - - const auto& result = data.value(); - const auto error = curl_easy_strerror(result.code); - - if (result.code != CURLE_OK) - { - scripting::notify(object_id, "done", {{}, false, error}); - return; - } - - if (result.buffer.size() >= 0x5000) - { - printf("^3WARNING: http result size bigger than 20480 bytes (%i), truncating!", static_cast(result.buffer.size())); - } - - scripting::notify(object_id, "done", {result.buffer.substr(0, 0x5000), true}); - }, scheduler::pipeline::server); - }, scheduler::pipeline::async); - - return object; + http_request_params_t params{}; + params.url = url; + return create_request(params); }, "http::get", "httpget", "curl"); gsc::function::add("http::request", [](const std::string& url, const scripting::variadic_args& va) { - const auto id = request_id++; - active_requests[id] = true; - - const auto object = scripting::object{}; - const auto object_id = object.get_entity_id(); - - std::string fields_string{}; - std::unordered_map headers_map{}; + http_request_params_t params{}; + params.url = url; if (va.size() > 0) { @@ -104,20 +166,19 @@ namespace http const auto key_ = key.as(); const auto value = fields_[key].to_string(); - fields_string += key_ + "=" + value + "&"; + params.fields += key_ + "=" + value + "&"; } } - - if (body.is()) + else if (body.is()) { - fields_string = body.as(); + params.fields = body.as(); } if (headers.is()) { - const auto headers_ = headers.as(); - const auto keys = headers_.get_keys(); + const auto headers_arr = headers.as(); + const auto keys = headers_arr.get_keys(); for (const auto& key : keys) { @@ -126,49 +187,15 @@ namespace http continue; } - const auto key_ = key.as(); - const auto value = headers_[key].to_string(); + const auto key_str = key.as(); + const auto value = headers_arr[key].to_string(); - headers_map[key_] = value; + params.headers[key_str] = value; } } } - scheduler::once([id, object_id, url, fields_string, headers_map]() - { - const auto data = utils::http::get_data(url, fields_string, headers_map); - scheduler::once([data, object_id, id] - { - if (active_requests.find(id) == active_requests.end()) - { - return; - } - - if (!data.has_value()) - { - scripting::notify(object_id, "done", {{}, false, "Unknown error"}); - return; - } - - const auto& result = data.value(); - const auto error = curl_easy_strerror(result.code); - - if (result.code != CURLE_OK) - { - scripting::notify(object_id, "done", {{}, false, error}); - return; - } - - if (result.buffer.size() >= 0x5000) - { - printf("^3WARNING: http result size bigger than 20480 bytes (%i), truncating!", static_cast(result.buffer.size())); - } - - scripting::notify(object_id, "done", {result.buffer.substr(0, 0x5000), true}); - }, scheduler::pipeline::server); - }, scheduler::pipeline::async); - - return object; + return create_request(params); }); } }; diff --git a/src/component/mysql.cpp b/src/component/mysql.cpp index 94727fb..55ed39a 100644 --- a/src/component/mysql.cpp +++ b/src/component/mysql.cpp @@ -22,15 +22,13 @@ namespace mysql struct task_t { - std::thread thread; - bool done; - bool canceled; - std::unique_ptr handle; - mysql_result_t result; + scripting::object handle; + mysql_result_t result{}; + std::atomic_bool completed; }; - uint64_t task_index{}; - std::unordered_map tasks; + std::vector> tasks; + std::array connection_pool; scripting::script_value field_to_value(const MYSQL_FIELD* field, const std::string& row) { @@ -210,35 +208,136 @@ namespace mysql template scripting::object create_mysql_query(F&& cb) { - auto task = &tasks[task_index++]; + auto task = std::make_shared(); + tasks.emplace_back(task); - task->done = false; - task->canceled = false; - task->handle = std::make_unique(); - - task->thread = std::thread([=]() + scheduler::thread_pool.push([=] { try { mysql::access([&](database_t& db) { task->result = cb(db); - task->done = true; + task->completed = true; }); } catch (const std::exception& e) { printf("%s\n", e.what()); - task->done = true; + task->completed = true; } }); - return *task->handle.get(); + return task->handle; + } + + void wait_and_clear_tasks() + { + for (auto& task : tasks) + { + while (!task->completed) + { + std::this_thread::sleep_for(1ms); + } + } + + tasks.clear(); + } + + void notify_task_result(std::shared_ptr& task) + { + const auto result = task->result.result + ? generate_result(task->result.result) + : generate_result(task->result.stmt); + + const auto rows = static_cast(task->result.affected_rows); + scripting::notify(task->handle.get_entity_id(), "done", {result, rows, task->result.error}); + + if (task->result.result != nullptr) + { + mysql_free_result(task->result.result); + task->result.result = nullptr; + } + + if (task->result.stmt != nullptr) + { + mysql_stmt_close(task->result.stmt); + task->result.stmt = nullptr; + } + } + + void check_tasks() + { + for (auto i = tasks.begin(); i != tasks.end(); ) + { + auto& task = *i; + if (!task->completed) + { + ++i; + continue; + } + else + { + notify_task_result(task); + i = tasks.erase(i); + } + } + } + + template + void bind_statement_args(MYSQL_BIND* binds, std::size_t& bind_count, const T& args) + { + bind_count = args.size(); + binds = utils::memory::allocate_array(bind_count); + + for (auto i = 0u; i < args.size(); i++) + { + const auto& arg = args[i]; + const auto& raw_value = arg.get_raw(); + + switch (raw_value.type) + { + case game::SCRIPT_FLOAT: + { + binds[i].buffer = utils::memory::allocate(); + binds[i].buffer_type = MYSQL_TYPE_FLOAT; + *reinterpret_cast(binds[i].buffer) = raw_value.u.floatValue; + break; + } + case game::SCRIPT_INTEGER: + { + binds[i].buffer = utils::memory::allocate(); + binds[i].buffer_type = MYSQL_TYPE_LONG; + *reinterpret_cast(binds[i].buffer) = raw_value.u.intValue; + break; + } + case game::SCRIPT_STRING: + { + const auto str = arg.as(); + const auto str_copy = utils::memory::duplicate_string(str); + binds[i].buffer = str_copy; + binds[i].buffer_length = str.size(); + binds[i].buffer_type = MYSQL_TYPE_STRING; + break; + } + default: + { + binds[i].buffer_type = MYSQL_TYPE_NULL; + break; + } + } + } + } + + void cleanup_connections() + { + for (auto& connection : connection_pool) + { + connection.cleanup(); + } } } - std::array connection_pool; - utils::concurrency::container& get_config() { static utils::concurrency::container config; @@ -260,23 +359,70 @@ namespace mysql return config; } - void cleanup_connections() + void connection::check() { - for (auto& connection : connection_pool) + const auto now = std::chrono::high_resolution_clock::now(); + const auto diff = now - this->start_; + + if (this->db.get() == nullptr || !this->db->ping_server() || diff >= connection_timeout) { - std::unique_lock lock(connection.mutex, std::try_to_lock); + get_config().access([&](sql::connection_config& cfg) + { + this->db = std::make_unique(cfg); + this->start_ = now; + }); + } + + this->last_access_ = now; + } + + void connection::cleanup() + { + std::unique_lock lock(this->mutex, std::try_to_lock); + if (!lock.owns_lock()) + { + return; + } + + const auto now = std::chrono::high_resolution_clock::now(); + const auto diff = now - this->last_access_; + if (diff >= connection_timeout) + { + this->db.reset(); + } + } + + connection* get_connection(std::unique_lock& lock) + { + static thread_local connection* last_connection{}; + if (last_connection != nullptr) + { + lock = std::unique_lock(last_connection->mutex, std::try_to_lock); + if (lock.owns_lock()) + { + return last_connection; + } + } + + for (auto i = 0u; i < connection_pool.size(); i++) + { + auto connection = &connection_pool[i]; + if (connection == last_connection) + { + continue; + } + + lock = std::unique_lock(connection->mutex, std::try_to_lock); if (!lock.owns_lock()) { continue; } - const auto now = std::chrono::high_resolution_clock::now(); - const auto diff = now - connection.last_access; - if (diff >= connection_timeout) - { - connection.db.reset(); - } + last_connection = connection; + return connection; } + + return nullptr; } class component final : public component_interface @@ -284,74 +430,14 @@ namespace mysql public: void on_shutdown([[maybe_unused]] plugin::plugin* plugin) override { - for (auto i = tasks.begin(); i != tasks.end(); ++i) - { - i->second.canceled = true; - - if (i->second.thread.joinable()) - { - i->second.thread.join(); - } - } + wait_and_clear_tasks(); } void on_startup([[maybe_unused]] plugin::plugin* plugin) override { - scripting::on_shutdown([]() - { - for (auto i = tasks.begin(); i != tasks.end(); ++i) - { - i->second.canceled = true; - i->second.handle.reset(); - } - }); - - scheduler::loop([] - { - cleanup_connections(); - }, scheduler::async, 1s); - - scheduler::loop([] - { - for (auto i = tasks.begin(); i != tasks.end(); ) - { - if (!i->second.done) - { - ++i; - continue; - } - - if (i->second.thread.joinable()) - { - i->second.thread.join(); - } - - if (!i->second.canceled) - { - const auto result = i->second.result.result - ? generate_result(i->second.result.result) - : generate_result(i->second.result.stmt); - - const auto rows = static_cast(i->second.result.affected_rows); - - scripting::notify(i->second.handle->get_entity_id(), "done", {result, rows, i->second.result.error}); - - if (i->second.result.result) - { - mysql_free_result(i->second.result.result); - i->second.result.result = nullptr; - } - - if (i->second.result.stmt) - { - mysql_stmt_close(i->second.result.stmt); - i->second.result.stmt = nullptr; - } - } - - i = tasks.erase(i); - } - }, scheduler::server); + scripting::on_shutdown(wait_and_clear_tasks); + scheduler::loop(cleanup_connections, scheduler::async, 60s); + scheduler::loop(check_tasks, scheduler::server); gsc::function::add("mysql::set_config", [](const scripting::object& config) { @@ -407,57 +493,13 @@ namespace mysql try { - const auto bind_args = [&](const T& args) - { - bind_count = args.size(); - binds = utils::memory::allocate_array(bind_count); - - for (auto i = 0u; i < args.size(); i++) - { - const auto& arg = args[i]; - const auto& raw_value = arg.get_raw(); - - switch (raw_value.type) - { - case game::SCRIPT_FLOAT: - { - binds[i].buffer = utils::memory::allocate(); - binds[i].buffer_type = MYSQL_TYPE_FLOAT; - *reinterpret_cast(binds[i].buffer) = raw_value.u.floatValue; - break; - } - case game::SCRIPT_INTEGER: - { - binds[i].buffer = utils::memory::allocate(); - binds[i].buffer_type = MYSQL_TYPE_LONG; - *reinterpret_cast(binds[i].buffer) = raw_value.u.intValue; - break; - } - case game::SCRIPT_STRING: - { - const auto str = arg.as(); - const auto str_copy = utils::memory::duplicate_string(str); - binds[i].buffer = str_copy; - binds[i].buffer_length = str.size(); - binds[i].buffer_type = MYSQL_TYPE_STRING; - break; - } - default: - { - binds[i].buffer_type = MYSQL_TYPE_NULL; - break; - } - } - } - }; - if (values.size() > 0 && values[0].is()) { - bind_args(values[0].as()); + bind_statement_args(binds, bind_count, values[0].as()); } else { - bind_args(values); + bind_statement_args(binds, bind_count, values); } } catch (const std::exception& e) diff --git a/src/component/mysql.hpp b/src/component/mysql.hpp index 49724eb..bd88662 100644 --- a/src/component/mysql.hpp +++ b/src/component/mysql.hpp @@ -16,55 +16,43 @@ namespace sql = sqlpp::mysql; namespace mysql { - constexpr auto max_connections = 256; + constexpr auto max_connections = 100; constexpr auto connection_timeout = 200s; using database_mutex_t = std::recursive_mutex; using database_t = std::unique_ptr; - struct connection_t + utils::concurrency::container& get_config(); + + class connection { + public: + connection() = default; + void check(); + void cleanup(); + database_t db; database_mutex_t mutex; - std::chrono::high_resolution_clock::time_point start; - std::chrono::high_resolution_clock::time_point last_access; + + private: + std::chrono::high_resolution_clock::time_point start_; + std::chrono::high_resolution_clock::time_point last_access_; + }; - extern std::array connection_pool; - - utils::concurrency::container& get_config(); + connection* get_connection(std::unique_lock& lock); template T access(F&& accessor) { - for (auto& connection : connection_pool) + std::unique_lock lock; + auto conn = get_connection(lock); + if (conn == nullptr) { - std::unique_lock lock(connection.mutex, std::try_to_lock); - if (!lock.owns_lock()) - { - continue; - } - - const auto now = std::chrono::high_resolution_clock::now(); - const auto diff = now - connection.start; - - if (!connection.db.get() || !connection.db->ping_server() || diff >= 1h) - { - get_config().access([&](sql::connection_config& cfg) - { - connection.db = std::make_unique(cfg); - connection.start = now; - }); - } - - connection.last_access = now; - return accessor(connection.db); + throw std::runtime_error("out of connections"); } - throw std::runtime_error("out of connections"); + conn->check(); + return accessor(conn->db); } - - void cleanup_connections(); - - void run_tasks(); } diff --git a/src/component/scheduler.cpp b/src/component/scheduler.cpp index 4b68b73..2d85400 100644 --- a/src/component/scheduler.cpp +++ b/src/component/scheduler.cpp @@ -9,6 +9,8 @@ namespace scheduler { + utils::thread_pool thread_pool; + namespace { utils::hook::detour server_frame_hook; @@ -142,6 +144,9 @@ namespace scheduler public: void on_startup([[maybe_unused]] plugin::plugin* plugin) override { + thread_pool.initialize(8); + thread_pool.start(); + thread = std::thread([]() { while (!kill_thread) @@ -156,6 +161,7 @@ namespace scheduler void on_shutdown([[maybe_unused]] plugin::plugin* plugin) override { + thread_pool.stop(); kill_thread = true; if (thread.joinable()) diff --git a/src/component/scheduler.hpp b/src/component/scheduler.hpp index cabb37e..46ccef1 100644 --- a/src/component/scheduler.hpp +++ b/src/component/scheduler.hpp @@ -1,5 +1,7 @@ #pragma once +#include + namespace scheduler { enum pipeline @@ -18,4 +20,6 @@ namespace scheduler std::chrono::milliseconds delay = 0ms); void once(const std::function& callback, pipeline type = pipeline::server, std::chrono::milliseconds delay = 0ms); + + extern utils::thread_pool thread_pool; } diff --git a/src/utils/thread_pool.cpp b/src/utils/thread_pool.cpp new file mode 100644 index 0000000..092c89b --- /dev/null +++ b/src/utils/thread_pool.cpp @@ -0,0 +1,96 @@ +#include + +#include "thread_pool.hpp" + +namespace utils +{ + thread_pool::worker::worker() + { + } + + void thread_pool::worker::start(thread_pool& pool) + { + this->thread_ = std::thread([&] + { + this->loop(pool); + }); + } + + void thread_pool::worker::stop() + { + if (this->thread_.joinable()) + { + this->thread_.join(); + } + } + + void thread_pool::worker::loop(thread_pool& pool) + { + while (!pool.stopped_) + { + pool.run_job(); + } + } + + thread_pool::job_ptr thread_pool::pop_job() + { + thread_pool::job_ptr job = std::move(this->queue_.front()); + this->queue_.pop_front(); + return job; + } + + void thread_pool::run_job() + { + std::unique_lock lock(this->mutex_); + + this->event_.wait(lock, [&]() + { + return !this->queue_.empty() || this->stopped_; + }); + + if (this->stopped_ || this->queue_.empty()) + { + return; + } + + auto job = this->pop_job(); + lock.unlock(); + job->operator()(); + } + + thread_pool::thread_pool(const std::size_t num_workers) + { + this->initialize(num_workers); + } + + void thread_pool::initialize(const std::size_t num_workers) + { + for (auto i = 0u; i < num_workers; i++) + { + this->workers_.emplace_back(std::make_unique()); + } + } + + void thread_pool::start() + { + for (auto& worker : this->workers_) + { + worker->start(*this); + } + } + + void thread_pool::update() + { + + } + + void thread_pool::stop() + { + this->stopped_ = true; + this->event_.notify_all(); + for (auto& worker : this->workers_) + { + worker->stop(); + } + } +} diff --git a/src/utils/thread_pool.hpp b/src/utils/thread_pool.hpp new file mode 100644 index 0000000..9b9bcb2 --- /dev/null +++ b/src/utils/thread_pool.hpp @@ -0,0 +1,60 @@ +#pragma once + +namespace utils +{ + class thread_pool + { + public: + using job = std::function; + using job_ptr = std::unique_ptr; + + friend class worker; + class worker + { + public: + worker(); + + friend class thread_pool; + + void start(thread_pool& pool); + void stop(); + void loop(thread_pool& pool); + + private: + std::thread thread_; + + }; + + thread_pool() = default; + thread_pool(const std::size_t num_workers); + void initialize(const std::size_t num_workers); + + void start(); + void update(); + void stop(); + + template + void push(F&& job) + { + if (this->stopped_) + { + return; + } + + std::lock_guard lock(this->mutex_); + this->queue_.emplace_back(std::make_unique(std::forward(job))); + this->event_.notify_one(); + } + + private: + thread_pool::job_ptr pop_job(); + void run_job(); + + std::mutex mutex_; + std::atomic_bool stopped_; + std::deque queue_; + std::condition_variable event_; + std::deque> workers_; + + }; +}