mysql的thread_running数量分析

2024-04-02 19:04:59 132人浏览薄情痞子

摘要

本篇内容主要讲解“Mysql的thread_running数量分析”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“mysql的thread_running数量分析

本篇内容主要讲解“Mysql的thread_running数量分析”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“mysql的thread_running数量分析”吧!

thread pool的原理：

已在server层完成解析；

层创建多组常驻线程，用于接收客户端连接发送的query并代为执行，而不是为每个连接单独创建一个线程。

层进行running thread数量判断，如果达到阈值则直接报错或sleep。

状态变量记录了当前并发执行stmt/command的数量，执行前加1执行后减1；

突然飙高的诱因：

客户端连接暴增；

系统性能瓶颈，如CPU,io或者mem swap；

异常sql；

会表现出hang住的假象。

执行，为此引入两个阈值low_watermark和high_watermark，以及变量threads_running_ctl_mode(selects或者all )；

前，检查thread_running，

若其已达high_watermark阈值则直接拒绝执行并返回错误：mysql server is too busy

若其位于low和high之间，则sleep 5ms，然后继续尝试，累计等待100ms后则执行

对于已经开启事务和super用户，不做限制

控制query类型：SELECTS/ALL，默认为SELECTS，表示只影响SELECT语句

源码见注1

优化为基于FIFO的cond-wait/signal(实现8个FIFO)；

高水位限流(这点保持不变)；

低水位优化；其他解决方案：mariadb开发thread pool，percona在其上实现了优先队列；

优势：思路与thread pool一致，但代码更简洁(不到1000行)；而且增加了特定query的过滤；

代码见注2

新增thread_active记录并发线程数，位于mysql_execute_command(sql解析之后)，高水位则在query解析之前判断；

只统计select/DML，而commit/rollback则放过。

采用FIFO，当thread_active >= thread_running_low_watermark时进程进入FIFO等待，其他线程执行完sql后唤醒FIFO；

内，同时引入threads_running_wait_timeout控制线程在FIFO最大等待时间，超时则直接报错返回。

引入8个FIFO，降低了进出FIFO的锁竞争，线程采用RR分配到不同fifo，每个队列限制并发运行线程为threads_running_low_watermark/8。

，开始执行query，[解析后进行低水位判断，若通过则执行]，执行当前sql完毕后，thread可能发起新query，则重复[]过程。

：进入FIFO排队最长时间，等待超时后sql被拒，默认100，单位为毫秒ms。

当前并发SELECT/INSERT/UPDATE/DELETE执行的线程数目；

：当前进入到FIFO中等待的线程数目；

未打补丁版本，设置innodb_thread_concurrency=0

未打补丁版本，innodb_thread_concurrency=32

低水位限流补丁版本（活跃线程数不超过64）

注1

Http://www.gpfeng.com/wp-content/uploads/2013/09/threads_running_control.txt

+static my_bool thread_running_control(THD *thd, ulong tr)
+{
+ int slept_cnt= 0;
+ ulong tr_low, tr_high;
+ DBUG_ENTER("thread_running_control");
+
+
+ if ( thd->security_ctx->master_access & SUPER_ACL|| --对于super权限的用户和已经开启的事务不做限制
+ thd->in_active_multi_stmt_transaction() ||
+ thd->slave_thread)
+ DBUG_RETURN(FALSE);
+
+
+ do
+ {
+ tr_low= thread_running_low_watermark;
+ tr_high= thread_running_high_watermark;
+
+ } while (tr_low > tr_high);
+
+check_buzy:
+
+
+ if ((tr_low == 0 && tr < tr_high) || (tr_low != 0 && tr < tr_low))
+ DBUG_RETURN(FALSE);
+
+ if (tr >= tr_high)
+ {
+ int can_reject= 1;
+
+
+ if (thread_running_ctl_mode == 0)
+ {
+ int query_is_select= 0;
+ if (thd->query_length() >= 8)
+ {
+ char *p= thd->query(); --读取query text的前6个字符，以判断是否为select
+ if (my_toupper(system_charset_info, p[0]) == 'S' &&
+ my_toupper(system_charset_info, p[1]) == 'E' &&
+ my_toupper(system_charset_info, p[2]) == 'L' &&
+ my_toupper(system_charset_info, p[3]) == 'E' &&
+ my_toupper(system_charset_info, p[4]) == 'C' &&
+ my_toupper(system_charset_info, p[5]) == 'T')
+
+ query_is_select= 1;
+ }
+
+ if (!query_is_select)
+ can_reject= 0;
+ }
+
+ if (can_reject)
+ {
+ inc_thread_rejected();
+ DBUG_RETURN(TRUE);
+ }
+ else
+ DBUG_RETURN(FALSE);
+ }
+
+ if (tr_low != 0 && tr >= tr_low)
+ {
+
+ if (slept_cnt >= 20)
+ DBUG_RETURN(FALSE);
+
+ dec_thread_running()
+
+
+ my_sleep(5000UL);
+
+ slept_cnt++;
+ tr= inc_thread_running() - 1;
+
+ Goto check_buzy;
+ }
+
+ DBUG_RETURN(FALSE);
+}
+
+
+ int32 tr= inc_thread_running() - 1;
if (!(server_command_flags[command] & CF_SKIP_QUESTIONS))
statistic_increment(thd->status_var.questions, &LOCK_status);

@@ -1129,6 +1240,13 @@
{
if (alloc_query(thd, packet, packet_length))
break; // fatal error is set
+
+ if (thread_running_control(thd, (ulong)tr))
+ {
+ my_error(ER_SERVER_THREAD_RUNNING_TOO_HIGH, MYF(0));
+ break;
+ }
+
MYSQL_QUERY_START(thd->query(), thd->thread_id, (char *) (thd->db ? thd->db : ""), &thd->security_ctx->priv_user[0])

注2
http://www.gpfeng.com/wp-content/uploads/2014/01/tr-control.diff_.txt
+
+ if ((!tr_high || tr <= tr_high) ||
+ thd->transaction.is_active() ||
+ thd->get_command() != COM_QUERY ||
+ thd->security_ctx->master_access & SUPER_ACL ||
+ thd->slave_thread)
+ DBUG_RETURN(FALSE);
+
+ const char *query= thd->query();
+ uint32 len= thd->query_length();
+
+ if ((!has_prefix(query, len, "SELECT", 6) && thread_running_ctl_mode == 0) || --不再是逐个字符判断
+ has_prefix(query, len, "COMMIT", 6) ||
+ has_prefix(query, len, "ROLLBACK", 8))
+ DBUG_RETURN(FALSE);
+
+
+ if (tr > tr_high && get_thread_running() > tr_high)
+ {
+ __sync_add_and_fetch(&thread_rejected, 1);
+ DBUG_RETURN(TRUE);
+ }
+
+ DBUG_RETURN(FALSE);
+}
+

+static my_bool thread_running_control_low_enter(THD *thd)
+{
+ int res= 0;
+ int32 tr_low;
+ my_bool ret= FALSE;
+ my_bool slept= FALSE;
+ struct timespec timeout;
+ Thread_conc_queue *queue;
+ DBUG_ENTER("thread_running_control_low_enter");
+
+
+ __sync_add_and_fetch(&thread_active, 1);
+
+ tr_low= (int32)queue_tr_low_watermark;
+ queue= thread_conc_queues + thd->query_id % N_THREAD_CONC_QUEUE;
+
+ queue->lock();--问1：在进行低水位判断前，先锁定FIFO，避免低水位验证失败时无法获取FIFO锁进而不能放入FIFO；
+
+retry:
+
+ if ((!tr_low || queue->thread_active < tr_low) ||
+ (thd->lex->sql_command != SQLCOM_SELECT && thread_running_ctl_mode == 0) ||
+ (!slept && (thd->transaction.is_active() ||
+ thd->security_ctx->master_access & SUPER_ACL || thd->slave_thread)))
+ {
+ queue->thread_active++; --判断是否满足进入FIFO条件，如不满足则立即更新thread_active++，解锁queue并退出；
+ queue->unlock();
+ DBUG_RETURN(ret);
+ }
+
+ if (!slept)
+ {
+ queue->unlock();
+
+
+ my_sleep(500);
+ slept= TRUE;
+ queue->lock();
+
+ goto retry;
+ }
+
+
+ Thread_wait_slot *slot= queue->pop_free();
+
+
+ if (!slot)-- 当FIFO都满了，即无法把当前线程放入，则必须放行让该sql正常执行
+ {
+ queue->thread_active++;
+ queue->unlock();
+ DBUG_RETURN(ret);
+ }
+
+ slot->signaled= false;
+ slot->wait_ended= false;
+
+
+ queue->push_back_wait(slot);
+ queue->thread_wait++;
+
+ queue->unlock();
+
+
+ thd_proc_info(thd, "waiting in server fifo");
+ __sync_sub_and_fetch(&thread_active, 1);
+ __sync_add_and_fetch(&thread_wait, 1);
+
+
+ set_timespec_nsec(timeout, thread_running_wait_timeout_ns);
+
+ mysql_mutex_lock(&slot->mutex);
+ while (!slot->signaled)
+ {
+ res= mysql_cond_timedwait(&slot->cond, &slot->mutex, &timeout);
+
+ slot->signaled= true;
+ }
+ mysql_mutex_unlock(&slot->mutex);
+
+ queue->lock();
+ queue->thread_wait--;
+ queue->thread_active++;
+
+
+ queue->remove_wait(slot);
+
+ queue->push_back_free(slot);
+
+ queue->unlock();
+
+
+ __sync_sub_and_fetch(&thread_wait, 1);
+ __sync_add_and_fetch(&thread_active, 1);
+ thd_proc_info(thd, 0);
+
+ if (res == ETIMEDOUT || res == ETIME)
+ {
+ ret= TRUE; // indicate that query is rejected.
+ __sync_add_and_fetch(&thread_rejected, 1);
+ }
+
+ DBUG_RETURN(ret);
+}