Future/t5-gsc-utils
1
0
Fork 0
mirror of https://github.com/alicealys/t5-gsc-utils.git synced 2026-07-28 10:40:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

mysql & http refactor

Browse Source
This commit is contained in:
alice
2026-07-27 06:06:23 +02:00
parent 05be91f6bc
commit 4969d6f28f
7 changed files with 492 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/component/http.cpp
+126 -99
View File
@@ -9,78 +9,140 @@
#include "scripting.hpp"
#include <utils/http.hpp>
#include <utils/concurrency.hpp>
#include <curl/curl.h>
namespace http
{
std::unordered_map<uint64_t, bool> 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<std::string, std::string> headers;
};
struct http_request_t
{
http_request_params_t params;
scripting::object handle;
std::optional<utils::http::result> result;
std::atomic_bool completed;
};
std::vector<std::shared_ptr<http_request_t>> requests;
void notify_request_result(std::shared_ptr<http_request_t>& 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<int>(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<http_request_t>();
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 <typename T>
void push_request(T&& r)
{
const auto request = std::make_shared<http_request_t>(std::forward<T>(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<int>(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<std::string, std::string> 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<std::string>();
const auto value = fields_[key].to_string();
fields_string += key_ + "=" + value + "&";
params.fields += key_ + "=" + value + "&";
}
}
if (body.is<std::string>())
else if (body.is<std::string>())
{
fields_string = body.as<std::string>();
params.fields = body.as<std::string>();
}
if (headers.is<scripting::array>())
{
const auto headers_ = headers.as<scripting::array>();
const auto keys = headers_.get_keys();
const auto headers_arr = headers.as<scripting::array>();
const auto keys = headers_arr.get_keys();
for (const auto& key : keys)
{
@@ -126,49 +187,15 @@ namespace http
continue;
}
const auto key_ = key.as<std::string>();
const auto value = headers_[key].to_string();
const auto key_str = key.as<std::string>();
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<int>(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);
});
}
};
src/component/mysql.cpp
+179 -137
View File
@@ -22,15 +22,13 @@ namespace mysql
struct task_t
{
std::thread thread;
bool done;
bool canceled;
std::unique_ptr<scripting::object> handle;
mysql_result_t result;
scripting::object handle;
mysql_result_t result{};
std::atomic_bool completed;
};
uint64_t task_index{};
std::unordered_map<uint64_t, task_t> tasks;
std::vector<std::shared_ptr<task_t>> tasks;
std::array<connection, max_connections> connection_pool;
scripting::script_value field_to_value(const MYSQL_FIELD* field, const std::string& row)
{
@@ -210,35 +208,136 @@ namespace mysql
template <typename F>
scripting::object create_mysql_query(F&& cb)
{
auto task = &tasks[task_index++];
auto task = std::make_shared<task_t>();
tasks.emplace_back(task);
task->done = false;
task->canceled = false;
task->handle = std::make_unique<scripting::object>();
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_t>& task)
{
const auto result = task->result.result
? generate_result(task->result.result)
: generate_result(task->result.stmt);
const auto rows = static_cast<std::size_t>(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 <typename T>
void bind_statement_args(MYSQL_BIND* binds, std::size_t& bind_count, const T& args)
{
bind_count = args.size();
binds = utils::memory::allocate_array<MYSQL_BIND>(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<float>();
binds[i].buffer_type = MYSQL_TYPE_FLOAT;
*reinterpret_cast<float*>(binds[i].buffer) = raw_value.u.floatValue;
break;
}
case game::SCRIPT_INTEGER:
{
binds[i].buffer = utils::memory::allocate<int>();
binds[i].buffer_type = MYSQL_TYPE_LONG;
*reinterpret_cast<int*>(binds[i].buffer) = raw_value.u.intValue;
break;
}
case game::SCRIPT_STRING:
{
const auto str = arg.as<std::string>();
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_t, max_connections> connection_pool;
utils::concurrency::container<sql::connection_config>& get_config()
{
static utils::concurrency::container<sql::connection_config> 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<database_mutex_t> lock(connection.mutex, std::try_to_lock);
get_config().access([&](sql::connection_config& cfg)
{
this->db = std::make_unique<sql::connection>(cfg);
this->start_ = now;
});
}
this->last_access_ = now;
}
void connection::cleanup()
{
std::unique_lock<database_mutex_t> 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<database_mutex_t>& 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<size_t>(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 = [&]<typename T>(const T& args)
{
bind_count = args.size();
binds = utils::memory::allocate_array<MYSQL_BIND>(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<float>();
binds[i].buffer_type = MYSQL_TYPE_FLOAT;
*reinterpret_cast<float*>(binds[i].buffer) = raw_value.u.floatValue;
break;
}
case game::SCRIPT_INTEGER:
{
binds[i].buffer = utils::memory::allocate<int>();
binds[i].buffer_type = MYSQL_TYPE_LONG;
*reinterpret_cast<int*>(binds[i].buffer) = raw_value.u.intValue;
break;
}
case game::SCRIPT_STRING:
{
const auto str = arg.as<std::string>();
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<scripting::array>())
{
bind_args(values[0].as<scripting::array>());
bind_statement_args(binds, bind_count, values[0].as<scripting::array>());
}
else
{
bind_args(values);
bind_statement_args(binds, bind_count, values);
}
}
catch (const std::exception& e)
src/component/mysql.hpp
+21 -33
View File
@@ -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<sql::connection>;
struct connection_t
utils::concurrency::container<sql::connection_config>& 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_t, max_connections> connection_pool;
utils::concurrency::container<sql::connection_config>& get_config();
connection* get_connection(std::unique_lock<database_mutex_t>& lock);
template <typename T = void, typename F>
T access(F&& accessor)
{
for (auto& connection : connection_pool)
std::unique_lock<database_mutex_t> lock;
auto conn = get_connection(lock);
if (conn == nullptr)
{
std::unique_lock<database_mutex_t> 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<sql::connection>(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();
}
src/component/scheduler.cpp
+6
View File
@@ -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())
src/component/scheduler.hpp
+4
View File
@@ -1,5 +1,7 @@
#pragma once
#include <utils/thread_pool.hpp>
namespace scheduler
{
enum pipeline
@@ -18,4 +20,6 @@ namespace scheduler
std::chrono::milliseconds delay = 0ms);
void once(const std::function<void()>& callback, pipeline type = pipeline::server,
std::chrono::milliseconds delay = 0ms);
extern utils::thread_pool thread_pool;
}
src/utils/thread_pool.cpp
+96
View File
@@ -0,0 +1,96 @@
#include <stdinc.hpp>
#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<std::mutex> 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<thread_pool::worker>());
}
}
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();
}
}
}
src/utils/thread_pool.hpp
+60
View File
@@ -0,0 +1,60 @@
#pragma once
namespace utils
{
class thread_pool
{
public:
using job = std::function<void()>;
using job_ptr = std::unique_ptr<job>;
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 <typename F>
void push(F&& job)
{
if (this->stopped_)
{
return;
}
std::lock_guard lock(this->mutex_);
this->queue_.emplace_back(std::make_unique<thread_pool::job>(std::forward<F>(job)));
this->event_.notify_one();
}
private:
thread_pool::job_ptr pop_job();
void run_job();
std::mutex mutex_;
std::atomic_bool stopped_;
std::deque<job_ptr> queue_;
std::condition_variable event_;
std::deque<std::unique_ptr<worker>> workers_;
};
}