Runtime timeouts (#1610)

jeffreylovitz · swilly22 · web-flow · commit 964b268e9efd · 2021-03-09T14:00:51.000+02:00
* Add run-time configuration for default query timeouts

* Timeout for write queries that haven't committed changes

* define TIMEOUT_NO_TIMEOUT

* Refactor timeout logic

* Address PR comments

* Do not use timeouts for write queries

Co-authored-by: swilly22 &lt;roi@redislabs.com&gt;
Co-authored-by: Roi Lipman &lt;swilly22@users.noreply.github.com&gt;
diff --git a/docs/configuration.md b/docs/configuration.md
@@ -82,15 +82,29 @@ If enabled, RedisGraph will maintain transposed copies of relationship matrices.
 $ redis-server --loadmodule ./redisgraph.so MAINTAIN_TRANSPOSED_MATRICES no
 ```
 
+---
+
+## TIMEOUT
+
+Timeout is a flag that specifies the maximum runtime for read queries in milliseconds. This configuration will not be respected by write queries, to avoid leaving the graph in an inconsistent state.
+
+### Default
+
+`TIMEOUT` is off by default (config value of `0`).
+
+### Example
+
+```
+$ redis-server --loadmodule ./redisgraph.so TIMEOUT 1000
+```
+
 # Query Configurations
 
 Some configurations may be set per query in the form of additional arguments after the query string. All per-query configurations are off by default unless using a language-specific client, which may establish its own defaults.
 
 ## Query Timeout
 
-The query flag `timeout` allows the user to specify the maximum runtime allowed for a query in milliseconds. This configuration can only be set for read queries to avoid leaving the graph in an inconsistent state.
-
-`timeout` may still return partial results followed by an error message indicating the timeout.
+The query flag `timeout` allows the user to specify a timeout as described in [TIMEOUT](#TIMEOUT) for a single query.
 
 ### Example
 
diff --git a/src/commands/cmd_dispatcher.c b/src/commands/cmd_dispatcher.c
@@ -6,6 +6,7 @@
 
 #include "RG.h"
 #include "commands.h"
+#include "../config.h"
 #include "cmd_context.h"
 #include "../util/thpool/pools.h"
 
@@ -22,9 +23,9 @@ static int _read_flags(RedisModuleString **argv, int argc, bool *compact,
 	ASSERT(timeout);
 
 	// set defaults
-	*timeout = 0;      // no timeout
 	*compact = false;  // verbose
 	*graph_version = GRAPH_VERSION_MISSING;
+	Config_Option_get(Config_TIMEOUT, timeout);
 
 	// GRAPH.QUERY <GRAPH_KEY> <QUERY>
 	// make sure we've got more than 3 arguments
diff --git a/src/commands/cmd_query.c b/src/commands/cmd_query.c
@@ -259,14 +259,8 @@ void Graph_Query(void *args) {
 
 	// set the query timeout if one was specified
 	if(command_ctx->timeout != 0) {
-		if(!readonly) {
-			// disallow timeouts on write operations to avoid leaving the graph in an inconsistent state
-			ErrorCtx_SetError("Query timeouts may only be specified on read-only queries");
-			ErrorCtx_EmitException();
-			goto cleanup;
-		}
-
-		Query_SetTimeOut(command_ctx->timeout, exec_ctx->plan);
+		// disallow timeouts on write operations to avoid leaving the graph in an inconsistent state
+		if(readonly) Query_SetTimeOut(command_ctx->timeout, exec_ctx->plan);
 	}
 
 	// populate the container struct for invoking _ExecuteQuery.
diff --git a/src/config.c b/src/config.c
@@ -16,13 +16,29 @@
 // Configuration parameters
 //-----------------------------------------------------------------------------
 
-#define CACHE_SIZE "CACHE_SIZE"  // Config param, the size of each thread cache size, per graph.
-#define ASYNC_DELETE "ASYNC_DELETE" // whether graphs should be deleted asynchronously
-#define THREAD_COUNT "THREAD_COUNT" // Config param, number of threads in thread pool
-#define RESULTSET_SIZE "RESULTSET_SIZE" // resultset size limit
-#define OMP_THREAD_COUNT "OMP_THREAD_COUNT" // Config param, max number of OpenMP threads
-#define VKEY_MAX_ENTITY_COUNT "VKEY_MAX_ENTITY_COUNT" // Config param, max number of entities in each virtual key
-#define MAINTAIN_TRANSPOSED_MATRICES "MAINTAIN_TRANSPOSED_MATRICES" // Whether the module should maintain transposed relationship matrices
+// config param, the timeout for each query in milliseconds
+#define TIMEOUT "TIMEOUT"
+
+// config param, the size of each thread cache size, per graph
+#define CACHE_SIZE "CACHE_SIZE"
+
+// whether graphs should be deleted asynchronously
+#define ASYNC_DELETE "ASYNC_DELETE"
+
+// config param, number of threads in thread pool
+#define THREAD_COUNT "THREAD_COUNT"
+
+// resultset size limit
+#define RESULTSET_SIZE "RESULTSET_SIZE"
+
+// config param, max number of OpenMP threads
+#define OMP_THREAD_COUNT "OMP_THREAD_COUNT"
+
+// config param, max number of entities in each virtual key
+#define VKEY_MAX_ENTITY_COUNT "VKEY_MAX_ENTITY_COUNT"
+
+// whether the module should maintain transposed relationship matrices
+#define MAINTAIN_TRANSPOSED_MATRICES "MAINTAIN_TRANSPOSED_MATRICES"
 
 //------------------------------------------------------------------------------
 // Configuration defaults
@@ -76,6 +92,18 @@ static inline bool _Config_ParseYesNo(RedisModuleString *rm_str, bool *value) {
 // Config access functions
 //==============================================================================
 
+//------------------------------------------------------------------------------
+// timeout
+//------------------------------------------------------------------------------
+
+void Config_timeout_set(uint64_t timeout) {
+	config.timeout = timeout;
+}
+
+uint Config_timeout_get(void) {
+	return config.timeout;
+}
+
 //------------------------------------------------------------------------------
 // thread count
 //------------------------------------------------------------------------------
@@ -168,6 +196,8 @@ bool Config_Contains_field(const char *field_str, Config_Option_Field *field) {
 
 	if(!strcasecmp(field_str, THREAD_COUNT)) {
 		f = Config_THREAD_POOL_SIZE;
+	} else if(!strcasecmp(field_str, TIMEOUT)) {
+		f = Config_TIMEOUT;
 	} else if(!strcasecmp(field_str, OMP_THREAD_COUNT)) {
 		f = Config_OPENMP_NTHREAD;
 	} else if(!strcasecmp(field_str, VKEY_MAX_ENTITY_COUNT)) {
@@ -190,6 +220,10 @@ const char *Config_Field_name(Config_Option_Field field) {
 	const char *name = NULL;
 	switch (field)
 	{
+		case Config_TIMEOUT:
+			name = TIMEOUT;
+			break;
+
 		case Config_CACHE_SIZE:
 			name = CACHE_SIZE;
 			break;
@@ -260,6 +294,9 @@ void _Config_SetToDefaults(RedisModuleCtx *ctx) {
 
 	// no limit on result-set size
 	config.resultset_size = RESULTSET_SIZE_UNLIMITED;
+
+	// no query timeout by default
+	config.timeout = CONFIG_TIMEOUT_NO_TIMEOUT;
 }
 
 int Config_Init(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
@@ -310,6 +347,18 @@ bool Config_Option_set(Config_Option_Field field, RedisModuleString *val) {
 
 	switch (field)
 	{
+		//----------------------------------------------------------------------
+		// timeout
+		//----------------------------------------------------------------------
+
+		case Config_TIMEOUT:
+			{
+				long long timeout;
+				if(!_Config_ParsePositiveInteger(val, &timeout)) return false;
+				Config_timeout_set(timeout);
+			}
+			break;
+
 		//----------------------------------------------------------------------
 		// cache size
 		//----------------------------------------------------------------------
@@ -421,6 +470,21 @@ bool Config_Option_get(Config_Option_Field field, ...) {
 
 	switch (field)
 	{
+		//----------------------------------------------------------------------
+		// timeout
+		//----------------------------------------------------------------------
+
+		case Config_TIMEOUT:
+			{
+				va_start(ap, field);
+				uint64_t *timeout = va_arg(ap, uint64_t*);
+				va_end(ap);
+
+				ASSERT(timeout != NULL);
+				(*timeout) = Config_timeout_get();
+			}
+			break;
+
 		//----------------------------------------------------------------------
 		// cache size
 		//----------------------------------------------------------------------
diff --git a/src/config.h b/src/config.h
@@ -9,22 +9,25 @@
 #include <stdbool.h>
 #include "redismodule.h"
 
-#define RESULTSET_SIZE_UNLIMITED UINT64_MAX
+#define RESULTSET_SIZE_UNLIMITED    UINT64_MAX
+#define CONFIG_TIMEOUT_NO_TIMEOUT   0
 #define VKEY_ENTITY_COUNT_UNLIMITED UINT64_MAX
 
 typedef enum {
-	Config_CACHE_SIZE               = 0,  // number of entries in cache
-	Config_ASYNC_DELETE             = 1,  // delete graph asynchronously
-	Config_OPENMP_NTHREAD           = 2,  // max number of OpenMP threads to use
-	Config_THREAD_POOL_SIZE         = 3,  // number of threads in thread pool
-	Config_RESULTSET_MAX_SIZE       = 4,  // max number of records in result-set
-	Config_MAINTAIN_TRANSPOSE       = 5,  // maintain transpose matrices
-	Config_VKEY_MAX_ENTITY_COUNT    = 6,  // max number of elements in vkey
-	Config_END_MARKER               = 7
+	Config_TIMEOUT                  = 0,  // timeout value for queries
+	Config_CACHE_SIZE               = 1,  // number of entries in cache
+	Config_ASYNC_DELETE             = 2,  // delete graph asynchronously
+	Config_OPENMP_NTHREAD           = 3,  // max number of OpenMP threads to use
+	Config_THREAD_POOL_SIZE         = 4,  // number of threads in thread pool
+	Config_RESULTSET_MAX_SIZE       = 5,  // max number of records in result-set
+	Config_MAINTAIN_TRANSPOSE       = 6,  // maintain transpose matrices
+	Config_VKEY_MAX_ENTITY_COUNT    = 7,  // max number of elements in vkey
+	Config_END_MARKER               = 8
 } Config_Option_Field;
 
 // configuration object
 typedef struct {
+	uint64_t timeout;                  // The timeout for each query in milliseconds.
 	bool async_delete;                 // If true, graph deletion is done asynchronously.
 	uint64_t cache_size;               // The cache size for each thread, per graph.
 	uint thread_pool_size;             // Thread count for thread pool.
@@ -35,8 +38,8 @@ typedef struct {
 } RG_Config;
 
 // Run-time configurable fields
-#define RUNTIME_CONFIG_COUNT 1
-static const Config_Option_Field RUNTIME_CONFIGS[] = { Config_RESULTSET_MAX_SIZE };
+#define RUNTIME_CONFIG_COUNT 2
+static const Config_Option_Field RUNTIME_CONFIGS[] = { Config_RESULTSET_MAX_SIZE, Config_TIMEOUT };
 
 // Set module-level configurations to defaults or to user arguments where provided.
 // returns REDISMODULE_OK on success, emits an error and returns REDISMODULE_ERR on failure.
diff --git a/src/execution_plan/execution_plan.c b/src/execution_plan/execution_plan.c
@@ -89,7 +89,7 @@ static ExecutionPlan *_ExecutionPlan_UnionPlans(AST *ast) {
 
 	// Introduce distinct only if `ALL` isn't specified.
 	const cypher_astnode_t *union_clause = AST_GetClause(ast, CYPHER_AST_UNION,
-			NULL);
+														 NULL);
 	if(!cypher_ast_union_has_all(union_clause)) {
 		OpBase *distinct_op = NewDistinctOp(plan);
 		ExecutionPlan_AddOp(results_op, distinct_op);
@@ -303,16 +303,6 @@ ExecutionPlan *NewExecutionPlan(void) {
 	return plan;
 }
 
-// Sets an AST segment in the execution plan.
-inline void ExecutionPlan_SetAST(ExecutionPlan *plan, AST *ast) {
-	plan->ast_segment = ast;
-}
-
-// Gets the AST segment from the execution plan.
-inline AST *ExecutionPlan_GetAST(const ExecutionPlan *plan) {
-	return plan->ast_segment;
-}
-
 void ExecutionPlan_PreparePlan(ExecutionPlan *plan) {
 	// Plan should be prepared only once.
 	ASSERT(!plan->prepared);
diff --git a/src/execution_plan/execution_plan.h b/src/execution_plan/execution_plan.h
@@ -28,12 +28,6 @@ struct ExecutionPlan {
 /* Creates a new execution plan from AST */
 ExecutionPlan *NewExecutionPlan(void);
 
-// Sets an AST segment in the execution plan.
-void ExecutionPlan_SetAST(ExecutionPlan *plan, AST *ast);
-
-// Gets the AST segment from the execution plan.
-AST *ExecutionPlan_GetAST(const ExecutionPlan *plan);
-
 /* Prepare an execution plan for execution: optimize, initialize result set schema. */
 void ExecutionPlan_PreparePlan(ExecutionPlan *plan);
 
diff --git a/tests/flow/test_timeout.py b/tests/flow/test_timeout.py
@@ -2,8 +2,10 @@
 from RLTest import Env
 from base import FlowTestsBase
 from redis import ResponseError
+from redisgraph import Graph
 
 redis_con = None
+redis_graph = None
 
 class testQueryTimeout(FlowTestsBase):
     def __init__(self):
@@ -13,29 +15,51 @@ def __init__(self):
 
         self.env = Env(decodeResponses=True)
         global redis_con
+        global redis_graph
         redis_con = self.env.getConnection()
+        redis_graph = Graph("timeout", redis_con)
 
-    def test_read_query_timeout(self):
+    def test01_read_query_timeout(self):
         query = "UNWIND range(0,100000) AS x WITH x AS x WHERE x = 10000 RETURN x"
         try:
             # The query is expected to time out
-            redis_con.execute_command("GRAPH.QUERY", "g", query, "timeout", 1)
+            redis_graph.query(query, timeout=1)
             assert(False)
         except ResponseError as error:
-            self.env.assertTrue(isinstance(error, ResponseError))
             self.env.assertContains("Query timed out", str(error))
 
         try:
             # The query is expected to succeed
-            redis_con.execute_command("GRAPH.QUERY", "g", query, "timeout", 100)
+            redis_graph.query(query, timeout=100)
         except:
             assert(False)
 
-    def test_write_query_timeout(self):
-        query = "create ()"
+    def test02_configured_timeout(self):
+        # Verify that the module-level timeout is set to the default of 0
+        response = redis_con.execute_command("GRAPH.CONFIG GET timeout")
+        self.env.assertEquals(response[1], 0)
+        # Set a default timeout of 1 millisecond
+        redis_con.execute_command("GRAPH.CONFIG SET timeout 1")
+        response = redis_con.execute_command("GRAPH.CONFIG GET timeout")
+        self.env.assertEquals(response[1], 1)
+
+        # Validate that a read query times out
+        query = "UNWIND range(0,100000) AS x WITH x AS x WHERE x = 10000 RETURN x"
         try:
-            redis_con.execute_command("GRAPH.QUERY", "g", query, "timeout", 1)
+            redis_graph.query(query)
+            assert(False)
+        except ResponseError as error:
+            self.env.assertContains("Query timed out", str(error))
+
+    def test03_write_query_ignore_timeout(self):
+        # Verify that the timeout argument is ignored by write queries
+        query = "CREATE (a:M) WITH a UNWIND range(1,10000) AS ctr SET a.v = ctr"
+        try:
+            # The query should complete successfully
+            actual_result = redis_graph.query(query, timeout=1)
+            # The query should have taken longer than the timeout value
+            self.env.assertGreater(actual_result.run_time_ms, 1)
+            # The query should have updated properties 10,000 times
+            self.env.assertEquals(actual_result.properties_set, 10000)
+        except ResponseError:
             assert(False)
-        except:
-            # Expecting an error.
-            pass
