MDEV-39492 Parallel Query: Study how to create worker threads

Introduces parallel_worker_threads variable to control the number
of worker threads created by a parallel execution query.

2 new files, sql_parallel_workers.h sql_parallel_workers.cc which
contain structures for the creation, management and deletion of
parallel worker threads (pwt_ in the name).  Main management
class created in the stack in JOIN::exec, implemented for the
top level select.

Current parallel_worker_thread_func sleeps for 10 seconds, generates
a warning, signals the main thread, sleeps 10 seconds, signals the
main thread again, sets it's finished flag and cleans it's THD.

The main thread loops through worker threads, looking for finished
thread and cleans them up if they have finished.
It then waits for a signal, then processes it's message queue.

The thread management data is allocated on the stack in JOIN::exec.
Everything else is allocated using my_malloc() and my_free().

Threads are registed in server_threads, so are visible in
information_schema.processlist and the show processlist command.

We check that a kill query on a parallel worker is passed onto it's
manager and the query is properly aborted, and that a kill connection
is handled properly in parallel_worker.test.

Review input 1: cleanup earlier

Do cleanup before we've finished sending the result to the client.
This way, one can see the errors (and eventually warnings) marshalled
back to the main thread and returned to the user:

MariaDB [test]> set parallel_worker_threads=10;
Query OK, 0 rows affected (0.001 sec)

MariaDB [test]> select seq from seq_1_to_10;
ERROR 4103 (HY000): Argument to the worker_busted_function() function does not belong to the range [0,1]

Author: mariadb-RexJohnston

include/my_pthread.h

@@ -624,6 +624,18 @@ typedef uint64 my_thread_id;
 */
 #define MY_THREAD_ID_MAX UINT32_MAX
 
+#ifdef _WIN32
+#define MAX_THREAD_NAME 256
+#elif defined(__linux__)
+#define MAX_THREAD_NAME 16
+#elif defined(__FreeBSD__) || defined(__OpenBSD__)
+#define MAX_THREAD_NAME 19
+#include <pthread_np.h>
+#elif defined(__apple_build_version__)
+#include <sys/proc_info.h>
+#define MAX_THREAD_NAME MAXTHREADNAMESIZE
+#endif
+
 extern void my_threadattr_global_init(void);
 extern my_bool my_thread_global_init(void);
 extern void my_thread_set_name(const char *);

libmysqld/CMakeLists.txt

@@ -72,6 +72,7 @@ SET(SQL_EMBEDDED_SOURCES emb_qcache.cc libmysqld.c lib_sql.cc
            ../sql/mf_iocache.cc ../sql/my_decimal.cc 
            ../sql/net_serv.cc ../sql/opt_range.cc
            ../sql/opt_group_by_cardinality.cc
+           ../sql/sql_parallel_workers.cc
            ../sql/opt_rewrite_date_cmp.cc
            ../sql/opt_rewrite_remove_casefold.cc
            ../sql/opt_sargable_left.cc

mysql-test/main/mysqld--help.result

@@ -1012,6 +1012,9 @@ The following specify which files/extra groups are read (specified before remain
  Cost of checking the row against the WHERE clause.
  Increasing this will have the optimizer to prefer plans
  with less row combinations
+ --parallel-worker-threads=# 
+ Number of worker threads available for parallel query
+ execution. 0 means parallel execution is disabled
  --path=name         Comma-separated list of schema names that defines the
  search order for stored routines
  --performance-schema 
@@ -2004,6 +2007,7 @@ optimizer-trace
 optimizer-trace-max-mem-size 1048576
 optimizer-use-condition-selectivity 4
 optimizer-where-cost 0.032
+parallel-worker-threads 0
 path CURRENT_SCHEMA
 performance-schema FALSE
 performance-schema-accounts-size -1

mysql-test/main/parallel_query.result

@@ -0,0 +1,25 @@
+#
+# MDEV-39492 Parallel Query: Study how to create worker threads
+#
+connect  killee, localhost, root, , ;
+# check that kill query on a parallel worker is passed to the manager
+connection killee;
+set session parallel_worker_threads=3;
+select seq from seq_1_to_2;;
+connection default;
+kill query ID;
+connection killee;
+ERROR 70100: Query execution was interrupted
+# reset connection, discard any generated errors
+disconnect killee;
+connect  killee, localhost, root, , ;
+# check that kill on a parallel worker is passed to the manager
+# doing this last so we don't need to restart default connection
+set session parallel_worker_threads=3;
+select seq from seq_1_to_2;;
+connection default;
+kill ID;
+connection killee;
+Got one of the listed errors
+# generated error die with connection
+connection default;

mysql-test/main/parallel_query.test

@@ -0,0 +1,58 @@
+#
+# Test KILL and KILL QUERY statements.
+#
+
+-- source include/count_sessions.inc
+-- source include/not_embedded.inc
+# this will be used in the future  -- source include/have_innodb.inc
+-- source include/have_sequence.inc
+
+--disable_service_connection
+
+--echo #
+--echo # MDEV-39492 Parallel Query: Study how to create worker threads
+--echo #
+
+connect (killee, localhost, root, , );
+
+--echo # check that kill query on a parallel worker is passed to the manager
+
+connection killee;
+let $id= `select connection_id()`;
+set session parallel_worker_threads=3;
+--send select seq from seq_1_to_2;
+connection default;
+let $name= "%parallel worker $id";
+let $wait_condition= SELECT @kid:=ID from information_schema.processlist where info like $name limit 1;
+source include/wait_condition.inc;
+let $killID= `select @kid`;
+--replace_result $killID ID
+eval kill query $killID;
+connection killee;
+--error 1317
+--reap
+
+--echo # reset connection, discard any generated errors
+disconnect killee;
+connect (killee, localhost, root, , );
+
+--echo # check that kill on a parallel worker is passed to the manager
+--echo # doing this last so we don't need to restart default connection
+
+let $id= `select connection_id()`;
+set session parallel_worker_threads=3;
+--send select seq from seq_1_to_2;
+connection default;
+let $name= "%parallel worker $id";
+let $wait_condition= SELECT @kid:=ID from information_schema.processlist where info like $name limit 1;
+source include/wait_condition.inc;
+let $killID= `select @kid`;
+--replace_result $killID ID
+eval kill $killID;
+connection killee;
+--error 1317,2013
+--reap
+--echo # generated error die with connection
+
+connection default;
+--source include/wait_until_count_sessions.inc

mysql-test/suite/perfschema/r/dml_setup_instruments.result

@@ -49,8 +49,8 @@ wait/synch/cond/sql/COND_gtid_ignore_duplicates	YES	YES
 wait/synch/cond/sql/COND_manager	YES	YES
 wait/synch/cond/sql/COND_parallel_entry	YES	YES
 wait/synch/cond/sql/COND_prepare_ordered	YES	YES
+wait/synch/cond/sql/COND_pwt_new_message	YES	YES
 wait/synch/cond/sql/COND_queue_state	YES	YES
-wait/synch/cond/sql/COND_rpl_thread	YES	YES
 select * from performance_schema.setup_instruments
 where name='Wait';
 select * from performance_schema.setup_instruments

mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result

@@ -2992,6 +2992,16 @@ NUMERIC_BLOCK_SIZE	NULL
 ENUM_VALUE_LIST	NULL
 READ_ONLY	NO
 COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	PARALLEL_WORKER_THREADS
+VARIABLE_SCOPE	SESSION
+VARIABLE_TYPE	BIGINT UNSIGNED
+VARIABLE_COMMENT	Number of worker threads available for parallel query execution. 0 means parallel execution is disabled
+NUMERIC_MIN_VALUE	0
+NUMERIC_MAX_VALUE	100
+NUMERIC_BLOCK_SIZE	1
+ENUM_VALUE_LIST	NULL
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	REQUIRED
 VARIABLE_NAME	PATH
 VARIABLE_SCOPE	SESSION
 VARIABLE_TYPE	VARCHAR

mysys/my_thread_name.cc

@@ -20,12 +20,7 @@
 #include <stdio.h>
 
 #ifdef _WIN32
-#define MAX_THREAD_NAME 256
 typedef HRESULT (*func_SetThreadDescription)(HANDLE,PCWSTR);
-#elif defined(__linux__)
-#define MAX_THREAD_NAME 16
-#elif defined(__FreeBSD__) || defined(__OpenBSD__)
-#include <pthread_np.h>
 #endif
 
 #if defined(HAVE_PSI_THREAD_INTERFACE) && !defined DBUG_OFF

sql/CMakeLists.txt

@@ -115,6 +115,7 @@ SET (SQL_SOURCE
                ../sql-common/client_plugin.c
                opt_range.cc vector_mhnsw.cc
                opt_group_by_cardinality.cc
+               sql_parallel_workers.cc
                opt_rewrite_date_cmp.cc
                opt_rewrite_remove_casefold.cc
                opt_sargable_left.cc

sql/mysqld.cc

@@ -9911,6 +9911,7 @@ PSI_stage_info stage_starting= { 0, "starting", 0};
 PSI_stage_info stage_waiting_for_flush= { 0, "Waiting for non trans tables to be flushed", 0};
 PSI_stage_info stage_waiting_for_ddl= { 0, "Waiting for DDLs", 0};
 PSI_stage_info stage_waiting_for_reset_master= { 0, "Waiting for a running RESET MASTER to complete", 0};
+PSI_stage_info stage_reading_data_from_parallel_worker= { 0, "Reading data from parallel worker", 0};
 
 #ifdef WITH_WSREP
 // Additional Galera thread states
@@ -10003,6 +10004,10 @@ PSI_memory_key key_memory_user_var_entry_value;
 PSI_memory_key key_memory_String_value;
 PSI_memory_key key_memory_WSREP;
 PSI_memory_key key_memory_trace_ddl_info;
+PSI_memory_key key_memory_pwt_queued_event;
+PSI_memory_key key_memory_pwt_error_message;
+PSI_memory_key key_memory_pwt_workers;
+PSI_memory_key key_memory_pwt_db;
 
 #ifdef HAVE_PSI_INTERFACE
 
@@ -10256,6 +10261,9 @@ static PSI_memory_info all_server_memory[]=
   { &key_memory_trace_ddl_info, "TRACE_DDL_INFO", 0}
 };
 
+
+extern void pwt_init_psi_keys(void);
+
 /**
   Initialise all the performance schema instrumentation points
   used by the server.
@@ -10342,6 +10350,7 @@ void init_server_psi_keys(void)
   stmt_info_rpl.m_flags= PSI_FLAG_MUTABLE;
   mysql_statement_register(category, &stmt_info_rpl, 1);
 #endif
+  pwt_init_psi_keys();
 }
 
 #endif /* HAVE_PSI_INTERFACE */