#define SHUTDOWN_QUERY_TIMELIMIT (1 MINUTES) SUBSYSTEM_DEF(dbcore) name = "Database" flags = SS_TICKER|SS_NO_INIT wait = 10 // Not seconds because we're running on SS_TICKER runlevels = RUNLEVEL_LOBBY|RUNLEVELS_DEFAULT init_order = INIT_ORDER_DBCORE priority = FIRE_PRIORITY_DATABASE /// Number of failed connection attempts this try. Resets after the timeout or successful connection var/failed_connections = 0 /// Max number of consecutive failures before a timeout (here and not a define so it can be vv'ed mid round if needed) var/max_connection_failures = 5 /// world.time that connection attempts can resume var/failed_connection_timeout = 0 /// Total number of times connections have had to be timed out. var/failed_connection_timeout_count = 0 var/last_error /// The maximum number of queries that can be executed at the same time 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, mapped to complete arguments var/list/datum/db_query/queries_standby = list() /// We are in the process of shutting down and should not allow more DB connections var/shutting_down = FALSE /// Arbitrary handle returned from rust_g. var/connection /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) if(!length(queries_active) && !length(queries_standby) && !length(all_queries)) processing_queries = null return processing_queries = all_queries.Copy() // First handle the already running queries for (var/datum/db_query/query in queries_active) if(!process_query(query)) queries_active -= query // Now lets pull in standby queries if we have room. if (length(queries_standby) > 0 && length(queries_active) < max_concurrent_queries) var/list/queries_to_activate = queries_standby.Copy(1, min(length(queries_standby), max_concurrent_queries) + 1) for (var/datum/db_query/query in queries_to_activate) queries_standby.Remove(query) create_active_query(query) // And finally, let check queries for undeleted queries, check ticking if there is a lot of work to do. while(length(processing_queries)) var/datum/db_query/query = popleft(processing_queries) if(world.time - query.last_activity_time > (5 MINUTES)) stack_trace("Found undeleted query, check the sql.log for the undeleted query and add a delete call to the query datum.") log_subsystem_dbcore("Undeleted query: \"[query.sql]\" LA: [query.last_activity] LAT: [query.last_activity_time]") qdel(query) if(MC_TICK_CHECK) return /// Helper proc for handling activating queued queries /datum/controller/subsystem/dbcore/proc/create_active_query(datum/db_query/query) PRIVATE_PROC(TRUE) SHOULD_NOT_SLEEP(TRUE) 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(QDELETED(query)) return FALSE if(query.process((TICKS2DS(wait)) / 10)) queries_active -= query return FALSE return TRUE /datum/controller/subsystem/dbcore/proc/run_query_sync(datum/db_query/query) run_query(query) UNTIL(query.process()) return query /datum/controller/subsystem/dbcore/proc/run_query(datum/db_query/query) 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 (!length(queries_standby) && length(queries_active) < max_concurrent_queries) create_active_query(query) return queries_standby_num++ queries_standby |= query /datum/controller/subsystem/dbcore/Recover() connection = SSdbcore.connection /datum/controller/subsystem/dbcore/Shutdown() shutting_down = TRUE var/msg = "Clearing DB queries standby:[length(queries_standby)] active: [length(queries_active)] all: [length(all_queries)]" to_chat(world, SPAN_NOTICE(msg)) log_subsystem_dbcore(msg) //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 var/endtime = REALTIMEOFDAY + SHUTDOWN_QUERY_TIMELIMIT if(SSdbcore.Connect()) //Take over control of all active queries var/queries_to_check = queries_active.Copy() queries_active.Cut() //Start all waiting queries for(var/datum/db_query/query in queries_standby) run_query(query) queries_to_check += query queries_standby -= query //wait for them all to finish for(var/datum/db_query/query in queries_to_check) UNTIL(query.process() || REALTIMEOFDAY > endtime) //log shutdown to the db //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), // TRUE //) //query_round_shutdown.Execute(FALSE) //qdel(query_round_shutdown) msg = "Done clearing DB queries standby:[length(queries_standby)] active: [length(queries_active)] all: [length(all_queries)]" to_chat(world, SPAN_NOTICE(msg)) log_subsystem_dbcore(msg) if(IsConnected()) Disconnect() //nu /datum/controller/subsystem/dbcore/can_vv_get(var_name) 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_standby)) return FALSE if(var_name == NAMEOF(src, processing_queries)) 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_standby)) return FALSE if(var_name == NAMEOF(src, processing_queries)) return FALSE return ..() /// Loads the DB Configuration from the config/dbconfig.txt - To be replaced with a config subsystem at some point /datum/controller/subsystem/dbcore/proc/GetDBConfig() var/list/data = list( "address"="localhost", "port"="3306", "database"="aurorastation", "login", "password", "query_timeout_async"="10", "query_timeout_blocking"="5", "min_sql_connections"="1", "max_sql_connections"="25", "max_concurrent_queries"="25" ) var/list/Lines = file2list("config/dbconfig.txt") if (!Lines) // Return dummy object for safety. return new/DBConnection() for (var/t in Lines) if (!t) continue t = trim(t) if (length(t) == 0) continue else if (copytext(t, 1, 2) == "#") continue var/pos = findtext(t, " ") var/name = null var/value = null name = lowertext(copytext(t, 1, pos)) value = copytext(t, pos + 1) if (!name) continue if (name in data) data[name] = value else log_subsystem_dbcore("Unknown setting while setting up database connection. Filename: 'config/dbconfig.txt', value: '[value]'.") return data /// Establish a Connection with the Database if we are not already connected /datum/controller/subsystem/dbcore/proc/Connect() if(IsConnected()) return TRUE if(connection) Disconnect() //clear the current connection handle so isconnected() calls stop invoking rustg connection = null //make sure its cleared even if runtimes happened if(failed_connection_timeout <= world.time) //it's been long enough since we failed to connect, reset the counter failed_connections = 0 failed_connection_timeout = 0 if(failed_connection_timeout > 0) return FALSE if(!GLOB.config.sql_enabled) return FALSE //Just re-use the old db config file. //We might port the configuration subsystem from tg in the future, then that can be done better.. var/list/dbconfig = GetDBConfig() var/user = dbconfig["login"] var/pass = dbconfig["password"] var/db = dbconfig["database"] var/address = dbconfig["address"] var/port = text2num(dbconfig["port"]) var/timeout = max(text2num(dbconfig["query_timeout_async"]), text2num(dbconfig["query_timeout_blocking"])) var/min_sql_connections = text2num(dbconfig["min_sql_connections"]) var/max_sql_connections = text2num(dbconfig["max_sql_connections"]) var/result = json_decode(rustg_sql_connect_pool(json_encode(list( "host" = address, "port" = port, "user" = user, "pass" = pass, "db_name" = db, "read_timeout" = timeout, "write_timeout" = timeout, "min_threads" = min_sql_connections, "max_threads" = max_sql_connections, )))) . = (result["status"] == "ok") if (.) connection = result["handle"] else connection = null last_error = result["data"] log_subsystem_dbcore("Connect() failed | [last_error]") ++failed_connections //If it failed to establish a connection more than 5 times in a row, don't bother attempting to connect for a time. if(failed_connections > max_connection_failures) failed_connection_timeout_count++ //basic exponential backoff algorithm failed_connection_timeout = world.time + ((2 ** failed_connection_timeout_count) SECONDS) /// Disconnect from the Database /datum/controller/subsystem/dbcore/proc/Disconnect() failed_connections = 0 if (connection) rustg_sql_disconnect_pool(connection) connection = null /// Check if we have established a DB Connection /datum/controller/subsystem/dbcore/proc/IsConnected() if(!GLOB.config.sql_enabled) return FALSE if (!connection) return FALSE return json_decode(rustg_sql_connected(connection))["status"] == "online" /// Returns the last error message /datum/controller/subsystem/dbcore/proc/ErrorMsg() if(!GLOB.config.sql_enabled) return "Database disabled by configuration" return last_error /datum/controller/subsystem/dbcore/proc/ReportError(error) last_error = error /** * Creates a new /datum/db_query * * * sql_query - The Query string to be executed * * arguments - The Arguments for the quey * * allow_during_shutdown - If the query can be executed during shutdown * * Returns /datum/db_query */ /datum/controller/subsystem/dbcore/proc/NewQuery(sql_query, arguments, allow_during_shutdown=FALSE) //If the subsystem is shutting down, disallow new queries if(!allow_during_shutdown && shutting_down) CRASH("Attempting to create a new db query during the world shutdown") return new /datum/db_query(connection, sql_query, arguments) /** * Creates a new /datum/db_query_template * The returned template then needs to be Execute()´d to get a /datum/db_query * * * sql_query - The Query string to be executed * * Returns /datum/db_query_template */ /datum/controller/subsystem/dbcore/proc/NewQueryTemplate(sql_query) return new /datum/db_query_template(sql_query) /// The DB Query Template is used to prepare db queries that are re-used repeatedly /datum/db_query_template var/sql /** * Creates a new query template that can be reused multiple times * * * sql - The query to be Executed * */ /datum/db_query_template/New(sql) src.sql = sql /** * Executes a new instance of the query with the supplied arguments * * * arguments - The parameters of the query * * allow_during_shutdown - If the query is permited to be executed during shutdown * */ /datum/db_query_template/proc/Execute(arguments, allow_during_shutdown=FALSE) var/datum/db_query/query = SSdbcore.NewQuery(sql, arguments, allow_during_shutdown) return query.Execute() /datum/db_query // Inputs var/connection var/sql var/arguments /// The callback to invoke when the query is executed successfully var/datum/callback/success_callback /// The callback to invoke when the query was not executed successfully var/datum/callback/fail_callback // Status information /// 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 // Output var/list/list/rows var/next_row_to_take = 1 var/affected var/last_insert_id var/list/item //list of data values populated by NextRow() /datum/db_query/New(connection, sql, arguments) SSdbcore.all_queries += src SSdbcore.all_queries_num++ Activity("Created") item = list() src.connection = connection src.sql = sql src.arguments = arguments /datum/db_query/Destroy() Close() SSdbcore.all_queries -= src SSdbcore.queries_standby -= src SSdbcore.queries_active -= src return ..() /** * SetSuccessCallback * * Sets a callback to be executed if the query completes successfully. * The parameter for the callback is this query * * * success_callback - The callback to be executed * * Returns `TRUE` if the callback was set successfully * Returns `FALSE` if the callback could not be set, because another callback was already set */ /datum/db_query/proc/SetSuccessCallback(datum/callback/success_callback) if (src.success_callback != null) return FALSE src.success_callback = success_callback return TRUE /** * SetSuccessCallback * * Sets a callback to be executed if the query fails. * The parameter for the callback is this query * * * fail_callback - The callback to be executed * * Returns `TRUE` if the callback was set successfully * Returns `FALSE` if the callback could not be set, because another callback was already set */ /datum/db_query/proc/SetFailCallback(datum/callback/fail_callback) if (src.fail_callback != null) return FALSE src.fail_callback = fail_callback /// Stores the last activity and the time it happened on the query /datum/db_query/proc/Activity(activity) last_activity = activity last_activity_time = world.time /** * warn_execute * * Execute()'s the query and prints a warning to the chat of the `usr` if it fails * * * async - Default `TRUE` - See Execute() for a description * * Returns `TRUE` if the query was successfully executed */ /datum/db_query/proc/warn_execute(async = TRUE) . = Execute(async) if(!.) to_chat(usr, SPAN_DANGER("A SQL error occurred during this operation, check the server logs.")) /** * Execute * * Executes the query and waits until the query is successfully executed before return'ing * The query can be executed "sync" or "async". * If the query is executed async, the query is scheduled by the DBCore and not blocking other operations * If the query is executed sync, the query is executed immediately in a blocking manner * In case Success/Fail Callbacks are set, they are executed. * * * async - Default `TRUE` * * Returns `TRUE` if the query was successfully executed */ /datum/db_query/proc/Execute(async = TRUE) Activity("Execute") if(status == DB_QUERY_STARTED) CRASH("Attempted to start a new query while waiting on the old one") if(!SSdbcore.IsConnected()) last_error = "No connection!" return FALSE var/start_time if(!async) start_time = REALTIMEOFDAY Close() status = DB_QUERY_STARTED if(async) if(!MC_RUNNING(SSdbcore.init_stage)) 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_FINISHED) var/timed_out = !. && findtext(last_error, "Operation timed out") if(!async && timed_out) log_subsystem_dbcore("SLOW QUERY TIMEOUT. Query: [sql], start_time: [start_time], end_time: [REALTIMEOFDAY]") slow_query_check() /** * ExecuteNoSleep * * This executes a query without waiting for the results of the query. * This means, that you should not rely on being able to access the results of the query directly after the function call. * Instead you need to use the Callbacks, where you can then process the query result. * * - permit_before_dbcore_running `FALSE` - If set to TRUE allows the query to be queued before the DBCore Subsystem is running (will still only be executed after it is running) */ /datum/db_query/proc/ExecuteNoSleep(permit_before_dbcore_running=FALSE) Activity("Execute") if(status == DB_QUERY_STARTED) CRASH("Attempted to start a new query while waiting on the old one") if(!SSdbcore.IsConnected()) last_error = "No connection!" return FALSE Close() status = DB_QUERY_STARTED if(!MC_RUNNING(SSdbcore.init_stage) && !permit_before_dbcore_running) status = DB_QUERY_BROKEN last_error = "MC not running, Cant Execute NoSleep Query" log_subsystem_dbcore("Attemted to Execute NoSleep Query while MC was not running: [sql], Arguments: [json_encode(arguments)], error: [last_error]") return FALSE else SSdbcore.queue_query(src) return TRUE /// Sleeps until execution of the query has finished. /datum/db_query/proc/sync() while(status < DB_QUERY_FINISHED) stoplag() /** * process * * Invoked by the DBcore Subsystem * Checks if the query has been completed. * Stores the results of the query in the /datum/db_query * Calls invoke_callback() * And logs failed queries * * Returns `TRUE` if the query finished (successful or not) */ /datum/db_query/process(seconds_per_tick) if(status >= DB_QUERY_FINISHED) return TRUE // we are done processing after all status = DB_QUERY_STARTED var/job_result = rustg_sql_check_query(job_id) if(job_result == RUSTG_JOB_NO_RESULTS_YET) return FALSE //no results yet store_data(json_decode(job_result)) invoke_callback() if (status != DB_QUERY_FINISHED) log_subsystem_dbcore("SQL Query Failed. Query: [sql], Arguments: [json_encode(arguments)], error: [last_error]") return TRUE /// Stores the returned query data and updates the `status` of the /datum/db_query /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"] status = DB_QUERY_FINISHED return if("err") last_error = result["data"] status = DB_QUERY_BROKEN return if("offline") last_error = "CONNECTION OFFLINE" status = DB_QUERY_BROKEN return /// Invokes the appropriate query callback if set /datum/db_query/proc/invoke_callback() if (status == DB_QUERY_FINISHED) if (success_callback) success_callback.InvokeAsync(src) else if (fail_callback) fail_callback.InvokeAsync(src) /// Messages the admin on sync queries to ask if the server just hung /datum/db_query/proc/slow_query_check() message_admins("HEY! A database query timed out. Did the server just hang? \[YES\]|\[NO\]") /// Load the NextRow from the query data into the item variable of the /datum/db_query /datum/db_query/proc/NextRow(async = TRUE) Activity("NextRow") if (rows && next_row_to_take <= rows.len) item = rows[next_row_to_take] next_row_to_take++ return !!item else return FALSE /datum/db_query/proc/ErrorMsg() return last_error /// Deletes the Rows and Items from the Query /datum/db_query/proc/Close() rows = null item = null #undef SHUTDOWN_QUERY_TIMELIMIT