From eb384bd2d72a5b23dd9785cc06815049d507d3d5 Mon Sep 17 00:00:00 2001 From: Useroth <37159550+Useroth@users.noreply.github.com> Date: Wed, 19 Jan 2022 01:21:07 +0100 Subject: [PATCH] Telemetry 'n shit (#10810) * Refactors dbcore and limits the maximum amount of concurrent async queries to a variable amount (#59676) Refactors dbcore to work off a subsystem if executed async and limits the maximum amount of concurrent async queries to 25. This has been tested locally on a mysql docker image and there were no crashes (as long as you didn't run it with debug extools) + data was getting recorded fine. Why It's Good For The Game May or may not resolve terry crashes, however, each query creates a new thread which takes up 2mb, preventing the game from using that 2mb. This can lead to ooms if they stack up, e.g. due to poor connectivity. This solves that issue. maintainer note: this did not actually resolve the crashes, but has value anyway. Crashes were sidestepped fixed by finding out Large Address Awareness works cl refactor: Refactors dbcore.dm to possibly resolve the crashes that happen on Terry. /cl * Fixes an oversight in database code and cleans up telemetry (#64177) As it is right now, we never actually clear the temporary list processing_queries So if the subsystem is for some reason unable to complete a run, we will just whip right back around to it again If it's been long enough, this could even cause horrific log spam. There was just now a manuel round with roughly 30k undeleted query errors. not good. But what was actually not deleting you may ask? Well When you create a db request, a 5 minute timer starts. after those 5 minutes are up, the request is qdeleted by the db subsystem This is to prevent the creation of unused requests, and to handle requests that are never cleaned up Telemetry code was creating all of its db requests inside a for loop that could check tick, and then later attempting to call them in series Since requests by default sleep, this almost always lead to undeleted queries, which harddel'd given long enough periods I've fixed this by moving the data gathering away from the query creation Why is it good for the game I was working on atmos code, happy, safe in my delusion, when suddenly I got a ping from tattle freaking out over 200 undeleted queries a second This resolves that issue, so I can once again live in peace Changelog cl admin: Telemetry code will spam you with undeleted query logs much less often now! server: Improved how the db subsystem handles undeleted queries, should never have an incident like that again /cl * Fixes an error in telemetry queries (#64205) * Hardsynced time_track.dm with upstream Co-authored-by: Watermelon914 <37270891+Watermelon914@users.noreply.github.com> Co-authored-by: LemonInTheDark <58055496+LemonInTheDark@users.noreply.github.com> --- code/__DEFINES/database.dm | 6 + code/__DEFINES/subsystems.dm | 1 + .../configuration/entries/dbconfig.dm | 2 +- code/controllers/subsystem/dbcore.dm | 237 +++++++++++++++--- code/controllers/subsystem/time_track.dm | 8 +- code/modules/admin/admin_verbs.dm | 1 + code/modules/admin/verbs/debug.dm | 18 ++ code/modules/tgui_panel/telemetry.dm | 52 ++-- config/dbconfig.txt | 3 + tgstation.dme | 1 + 10 files changed, 263 insertions(+), 66 deletions(-) create mode 100644 code/__DEFINES/database.dm diff --git a/code/__DEFINES/database.dm b/code/__DEFINES/database.dm new file mode 100644 index 00000000000..3d20b3b9a3c --- /dev/null +++ b/code/__DEFINES/database.dm @@ -0,0 +1,6 @@ +/// When a query has been queued up for execution/is being executed +#define DB_QUERY_STARTED 0 +/// When a query is finished executing +#define DB_QUERY_FINISHED 1 +/// When there was a problem with the execution of a query. +#define DB_QUERY_BROKEN 2 diff --git a/code/__DEFINES/subsystems.dm b/code/__DEFINES/subsystems.dm index eae71e7140c..a3d5ecdcd82 100644 --- a/code/__DEFINES/subsystems.dm +++ b/code/__DEFINES/subsystems.dm @@ -174,6 +174,7 @@ #define FIRE_PRIORITY_VIS 10 #define FIRE_PRIORITY_AMBIENCE 10 #define FIRE_PRIORITY_GARBAGE 15 +#define FIRE_PRIORITY_DATABASE 16 #define FIRE_PRIORITY_WET_FLOORS 20 #define FIRE_PRIORITY_AIR 20 #define FIRE_PRIORITY_NPC 20 diff --git a/code/controllers/configuration/entries/dbconfig.dm b/code/controllers/configuration/entries/dbconfig.dm index 842e4f88058..2597a123a68 100644 --- a/code/controllers/configuration/entries/dbconfig.dm +++ b/code/controllers/configuration/entries/dbconfig.dm @@ -49,5 +49,5 @@ min_val = 1 /datum/config_entry/number/max_concurrent_queries - default = 25 + config_entry_value = 25 min_val = 1 diff --git a/code/controllers/subsystem/dbcore.dm b/code/controllers/subsystem/dbcore.dm index 2b73b992592..56d10d44150 100644 --- a/code/controllers/subsystem/dbcore.dm +++ b/code/controllers/subsystem/dbcore.dm @@ -1,8 +1,11 @@ SUBSYSTEM_DEF(dbcore) name = "Database" - flags = SS_BACKGROUND - wait = 1 MINUTES + flags = SS_TICKER + wait = 10 // Not seconds because we're running on SS_TICKER + runlevels = RUNLEVEL_INIT|RUNLEVEL_LOBBY|RUNLEVELS_DEFAULT init_order = INIT_ORDER_DBCORE + priority = FIRE_PRIORITY_DATABASE + var/failed_connection_timeout = 0 var/schema_mismatch = 0 @@ -11,7 +14,29 @@ SUBSYSTEM_DEF(dbcore) var/failed_connections = 0 var/last_error - var/list/active_queries = list() + + var/max_concurrent_queries = 25 + + /// Number of all queries, reset to 0 when logged in SStime_track. Used by SStime_track + var/all_queries_num = 0 + /// Number of active queries, reset to 0 when logged in SStime_track. Used by SStime_track + var/queries_active_num = 0 + /// Number of standby queries, reset to 0 when logged in SStime_track. Used by SStime_track + var/queries_standby_num = 0 + + /// All the current queries that exist. + var/list/all_queries = list() + /// Queries being checked for timeouts. + var/list/processing_queries + + /// Queries currently being handled by database driver + var/list/datum/db_query/queries_active = list() + /// Queries pending execution that will be handled this controller firing + var/list/datum/db_query/queries_new + /// Queries pending execution, mapped to complete arguments + var/list/datum/db_query/queries_standby = list() + /// Queries left to handle during controller firing + var/list/datum/db_query/queries_current var/connection // Arbitrary handle returned from rust_g. @@ -26,22 +51,109 @@ SUBSYSTEM_DEF(dbcore) return ..() -/datum/controller/subsystem/dbcore/fire() - for(var/I in active_queries) - var/datum/db_query/Q = I - if(world.time - Q.last_activity_time > (5 MINUTES)) +/datum/controller/subsystem/dbcore/stat_entry(msg) + msg = "P:[length(all_queries)]|Active:[length(queries_active)]|Standby:[length(queries_standby)]" + return ..() + +/// Resets the tracking numbers on the subsystem. Used by SStime_track. +/datum/controller/subsystem/dbcore/proc/reset_tracking() + all_queries_num = 0 + queries_active_num = 0 + queries_standby_num = 0 + +/datum/controller/subsystem/dbcore/fire(resumed = FALSE) + if(!IsConnected()) + return + + if(!resumed) + queries_new = null + if(!length(queries_active) && !length(queries_standby) && !length(all_queries)) + processing_queries = null + queries_current = null + return + queries_current = queries_active.Copy() + processing_queries = all_queries.Copy() + + while(length(processing_queries)) + var/datum/db_query/query = popleft(processing_queries) + if(world.time - query.last_activity_time > (5 MINUTES)) message_admins("Found undeleted query, please check the server logs and notify coders.") - log_sql("Undeleted query: \"[Q.sql]\" LA: [Q.last_activity] LAT: [Q.last_activity_time]") - qdel(Q) + log_sql("Undeleted query: \"[query.sql]\" LA: [query.last_activity] LAT: [query.last_activity_time]") + qdel(query) if(MC_TICK_CHECK) return + // First handle the already running queries + while(length(queries_current)) + var/datum/db_query/query = popleft(queries_current) + if(!process_query(query)) + queries_active -= query + if(MC_TICK_CHECK) + return + + // Then strap on extra new queries as possible + if(isnull(queries_new)) + if(!length(queries_standby)) + return + queries_new = queries_standby.Copy(1, min(length(queries_standby), max_concurrent_queries) + 1) + + while(length(queries_new) && length(queries_active) < max_concurrent_queries) + var/datum/db_query/query = popleft(queries_new) + queries_standby.Remove(query) + create_active_query(query) + if(MC_TICK_CHECK) + return + +/// Helper proc for handling queued new queries +/datum/controller/subsystem/dbcore/proc/create_active_query(datum/db_query/query) + PRIVATE_PROC(TRUE) + SHOULD_NOT_SLEEP(TRUE) + if(IsAdminAdvancedProcCall()) + return FALSE + run_query(query) + queries_active_num++ + queries_active += query + return query + +/datum/controller/subsystem/dbcore/proc/process_query(datum/db_query/query) + PRIVATE_PROC(TRUE) + SHOULD_NOT_SLEEP(TRUE) + if(IsAdminAdvancedProcCall()) + return FALSE + if(QDELETED(query)) + return FALSE + if(query.process(wait)) + queries_active -= query + return FALSE + return TRUE + +/datum/controller/subsystem/dbcore/proc/run_query_sync(datum/db_query/query) + if(IsAdminAdvancedProcCall()) + return + run_query(query) + UNTIL(query.process()) + return query + +/datum/controller/subsystem/dbcore/proc/run_query(datum/db_query/query) + if(IsAdminAdvancedProcCall()) + return + query.job_id = rustg_sql_query_async(connection, query.sql, json_encode(query.arguments)) + +/datum/controller/subsystem/dbcore/proc/queue_query(datum/db_query/query) + if(IsAdminAdvancedProcCall()) + return + queries_standby_num++ + queries_standby |= query + /datum/controller/subsystem/dbcore/Recover() connection = SSdbcore.connection /datum/controller/subsystem/dbcore/Shutdown() //This is as close as we can get to the true round end before Disconnect() without changing where it's called, defeating the reason this is a subsystem if(SSdbcore.Connect()) + for(var/datum/db_query/query in queries_current) + run_query(query) + var/datum/db_query/query_round_shutdown = SSdbcore.NewQuery( "UPDATE [format_table_name("round")] SET shutdown_datetime = Now(), end_state = :end_state WHERE id = :round_id", list("end_state" = SSticker.end_state, "round_id" = GLOB.round_id) @@ -53,11 +165,34 @@ SUBSYSTEM_DEF(dbcore) //nu /datum/controller/subsystem/dbcore/can_vv_get(var_name) - return var_name != NAMEOF(src, connection) && var_name != NAMEOF(src, active_queries) && ..() + if(var_name == NAMEOF(src, connection)) + return FALSE + if(var_name == NAMEOF(src, all_queries)) + return FALSE + if(var_name == NAMEOF(src, queries_active)) + return FALSE + if(var_name == NAMEOF(src, queries_new)) + return FALSE + if(var_name == NAMEOF(src, queries_standby)) + return FALSE + if(var_name == NAMEOF(src, queries_active)) + return FALSE + + return ..() /datum/controller/subsystem/dbcore/vv_edit_var(var_name, var_value) if(var_name == NAMEOF(src, connection)) return FALSE + if(var_name == NAMEOF(src, all_queries)) + return FALSE + if(var_name == NAMEOF(src, queries_active)) + return FALSE + if(var_name == NAMEOF(src, queries_new)) + return FALSE + if(var_name == NAMEOF(src, queries_standby)) + return FALSE + if(var_name == NAMEOF(src, queries_active)) + return FALSE return ..() /datum/controller/subsystem/dbcore/proc/Connect() @@ -82,6 +217,8 @@ SUBSYSTEM_DEF(dbcore) var/timeout = max(CONFIG_GET(number/async_query_timeout), CONFIG_GET(number/blocking_query_timeout)) var/thread_limit = CONFIG_GET(number/bsql_thread_limit) + max_concurrent_queries = CONFIG_GET(number/max_concurrent_queries) + var/result = json_decode(rustg_sql_connect_pool(json_encode(list( "host" = address, "port" = port, @@ -196,7 +333,7 @@ SUBSYSTEM_DEF(dbcore) for (var/thing in querys) var/datum/db_query/query = thing - UNTIL(!query.in_progress) + query.sync() if (qdel) qdel(query) @@ -280,8 +417,14 @@ Delayed insert mode was removed in mysql 7 and only works with MyISAM type table var/sql var/arguments + var/datum/callback/success_callback + var/datum/callback/fail_callback + // Status information - var/in_progress + /// Current status of the query. + var/status + /// Job ID of the query passed by rustg. + var/job_id var/last_error var/last_activity var/last_activity_time @@ -295,7 +438,8 @@ Delayed insert mode was removed in mysql 7 and only works with MyISAM type table var/list/item //list of data values populated by NextRow() /datum/db_query/New(connection, sql, arguments) - SSdbcore.active_queries[src] = TRUE + SSdbcore.all_queries += src + SSdbcore.all_queries_num++ Activity("Created") item = list() @@ -305,7 +449,9 @@ Delayed insert mode was removed in mysql 7 and only works with MyISAM type table /datum/db_query/Destroy() Close() - SSdbcore.active_queries -= src + SSdbcore.all_queries -= src + SSdbcore.queries_standby -= src + SSdbcore.queries_active -= src return ..() /datum/db_query/CanProcCall(proc_name) @@ -323,7 +469,7 @@ Delayed insert mode was removed in mysql 7 and only works with MyISAM type table /datum/db_query/proc/Execute(async = TRUE, log_error = TRUE) Activity("Execute") - if(in_progress) + if(status == DB_QUERY_STARTED) CRASH("Attempted to start a new query while waiting on the old one") if(!SSdbcore.IsConnected()) @@ -334,7 +480,18 @@ Delayed insert mode was removed in mysql 7 and only works with MyISAM type table if(!async) start_time = REALTIMEOFDAY Close() - . = run_query(async) + status = DB_QUERY_STARTED + if(async) + if(!Master.current_runlevel || Master.processing == 0) + SSdbcore.run_query_sync(src) + else + SSdbcore.queue_query(src) + sync() + else + var/job_result_str = rustg_sql_query_blocking(connection, sql, json_encode(arguments)) + store_data(json_decode(job_result_str)) + + . = (status != DB_QUERY_BROKEN) var/timed_out = !. && findtext(last_error, "Operation timed out") if(!. && log_error) log_sql("[last_error] | Query used: [sql] | Arguments: [json_encode(arguments)]") @@ -345,34 +502,40 @@ Delayed insert mode was removed in mysql 7 and only works with MyISAM type table log_query_debug("Query used: [sql]") slow_query_check() -/datum/db_query/proc/run_query(async) - var/job_result_str +/// Sleeps until execution of the query has finished. +/datum/db_query/proc/sync() + while(status < DB_QUERY_FINISHED) + stoplag() - if (async) - var/job_id = rustg_sql_query_async(connection, sql, json_encode(arguments)) - in_progress = TRUE - UNTIL((job_result_str = rustg_sql_check_query(job_id)) != RUSTG_JOB_NO_RESULTS_YET) - in_progress = FALSE +/datum/db_query/process(delta_time) + if(status >= DB_QUERY_FINISHED) + return - if (job_result_str == RUSTG_JOB_ERROR) - last_error = job_result_str - return FALSE - else - job_result_str = rustg_sql_query_blocking(connection, sql, json_encode(arguments)) + status = DB_QUERY_STARTED + var/job_result = rustg_sql_check_query(job_id) + if(job_result == RUSTG_JOB_NO_RESULTS_YET) + return - var/result = json_decode(job_result_str) - switch (result["status"]) - if ("ok") + store_data(json_decode(job_result)) + return TRUE + +/datum/db_query/proc/store_data(result) + switch(result["status"]) + if("ok") rows = result["rows"] affected = result["affected"] last_insert_id = result["last_insert_id"] - return TRUE - if ("err") + status = DB_QUERY_FINISHED + return + if("err") last_error = result["data"] - return FALSE - if ("offline") - last_error = "offline" - return FALSE + status = DB_QUERY_BROKEN + return + if("offline") + last_error = "CONNECTION OFFLINE" + status = DB_QUERY_BROKEN + return + /datum/db_query/proc/slow_query_check() message_admins("HEY! A database query timed out. Did the server just hang? \[YES\]|\[NO\]") diff --git a/code/controllers/subsystem/time_track.dm b/code/controllers/subsystem/time_track.dm index 9e3250a6aba..c2a0ff0f195 100644 --- a/code/controllers/subsystem/time_track.dm +++ b/code/controllers/subsystem/time_track.dm @@ -142,10 +142,10 @@ SUBSYSTEM_DEF(time_track) length(SSair.networks), length(SSair.high_pressure_delta), length(SSair.active_super_conductivity), - //SSdbcore.all_queries_num, - //SSdbcore.queries_active_num, - //SSdbcore.queries_standby_num + SSdbcore.all_queries_num, + SSdbcore.queries_active_num, + SSdbcore.queries_standby_num ) + send_maps_values ) - //SSdbcore.reset_tracking() + SSdbcore.reset_tracking() diff --git a/code/modules/admin/admin_verbs.dm b/code/modules/admin/admin_verbs.dm index a68a0149f3d..c2776b46581 100644 --- a/code/modules/admin/admin_verbs.dm +++ b/code/modules/admin/admin_verbs.dm @@ -182,6 +182,7 @@ GLOBAL_PROTECT(admin_verbs_debug) /client/proc/map_template_upload, /client/proc/jump_to_ruin, /client/proc/clear_dynamic_transit, + /client/proc/run_empty_query, /client/proc/toggle_medal_disable, /client/proc/view_runtimes, /client/proc/pump_random_event, diff --git a/code/modules/admin/verbs/debug.dm b/code/modules/admin/verbs/debug.dm index 55875137d81..8dd56c898ea 100644 --- a/code/modules/admin/verbs/debug.dm +++ b/code/modules/admin/verbs/debug.dm @@ -695,6 +695,24 @@ But you can call procs that are of type /mob/living/carbon/human/proc/ for that else to_chat(src, span_warning("Failed to place [template.name]."), confidential = TRUE) +/client/proc/run_empty_query(val as num) + set category = "Debug" + set name = "Run empty query" + set desc = "Amount of queries to run" + + var/list/queries = list() + for(var/i in 1 to val) + var/datum/db_query/query = SSdbcore.NewQuery("NULL") + INVOKE_ASYNC(query, /datum/db_query.proc/Execute) + queries += query + + for(var/datum/db_query/query as anything in queries) + query.sync() + qdel(query) + queries.Cut() + + message_admins("[key_name_admin(src)] ran [val] empty queries.") + /client/proc/clear_dynamic_transit() set category = "Debug" set name = "Clear Dynamic Turf Reservations" diff --git a/code/modules/tgui_panel/telemetry.dm b/code/modules/tgui_panel/telemetry.dm index 63d175b06ce..790b9fe10db 100644 --- a/code/modules/tgui_panel/telemetry.dm +++ b/code/modules/tgui_panel/telemetry.dm @@ -76,7 +76,7 @@ var/list/found - var/list/insert_queries = list() + var/list/query_data = list() for(var/i in 1 to len) if(QDELETED(client)) @@ -89,28 +89,10 @@ return if (!isnull(GLOB.round_id)) - insert_queries += SSdbcore.NewQuery({" - INSERT INTO [format_table_name("telemetry_connections")] ( - ckey, - telemetry_ckey, - address, - computer_id, - first_round_id, - latest_round_id - ) VALUES( - :ckey, - :telemetry_ckey, - INET_ATON(:address), - :computer_id, - :round_id, - :round_id - ) ON DUPLICATE KEY UPDATE latest_round_id = :round_id - "}, list( - "ckey" = ckey, + query_data += list(list( "telemetry_ckey" = row["ckey"], "address" = row["address"], "computer_id" = row["computer_id"], - "round_id" = GLOB.round_id, )) if (row["ckey"] in our_known_alts) @@ -130,7 +112,29 @@ log_suspicious_login(msg, access_log_mirror = FALSE) // Only log them all at the end, since it's not as important as reporting an evader - for (var/datum/db_query/insert_query as anything in insert_queries) - insert_query.Execute() - - QDEL_LIST(insert_queries) + for (var/list/one_query as anything in query_data) + var/datum/db_query/query = SSdbcore.NewQuery({" + INSERT INTO [format_table_name("telemetry_connections")] ( + ckey, + telemetry_ckey, + address, + computer_id, + first_round_id, + latest_round_id + ) VALUES( + :ckey, + :telemetry_ckey, + INET_ATON(:address), + :computer_id, + :round_id, + :round_id + ) ON DUPLICATE KEY UPDATE latest_round_id = :round_id + "}, list( + "ckey" = ckey, + "telemetry_ckey" = one_query["telemetry_ckey"], + "address" = one_query["address"], + "computer_id" = one_query["computer_id"], + "round_id" = GLOB.round_id, + )) + query.Execute() + qdel(query) diff --git a/config/dbconfig.txt b/config/dbconfig.txt index a6d0f1c29af..849daa5dbab 100644 --- a/config/dbconfig.txt +++ b/config/dbconfig.txt @@ -40,3 +40,6 @@ BLOCKING_QUERY_TIMEOUT 5 ## The maximum number of additional threads BSQL is allowed to run at once BSQL_THREAD_LIMIT 50 + +## The maximum number of concurrent asynchronous queries that can run at one time, handled by DM. +MAX_CONCURRENT_QUERIES 25 diff --git a/tgstation.dme b/tgstation.dme index c91a63580dd..9eeb62798db 100644 --- a/tgstation.dme +++ b/tgstation.dme @@ -54,6 +54,7 @@ #include "code\__DEFINES\construction.dm" #include "code\__DEFINES\cooldowns.dm" #include "code\__DEFINES\cult.dm" +#include "code\__DEFINES\database.dm" #include "code\__DEFINES\devices.dm" #include "code\__DEFINES\directional.dm" #include "code\__DEFINES\diseases.dm"