MySQL执行

分手后的思念是犯贱 2022-09-28 01:45 9阅读 0赞

典型查询执行以一个简单查询为例：select \* from some\_table where field\_x = 200，some\_table为MyISAM存储引擎。我们可以看到MyISAM存储引擎怎么通过存储引擎抽象层完成实际的执行。

[![clip\_image002][clip_image002]][clip_image002_clip_image002]

我们从MySQL最开始的周期开始，从 / sql / mysqld.cc的main开始  
int main(int argc, char \* \* argv) \{  
 **init\_common\_variables(MYSQL\_CONFIG\_NAME, argc, argv, load\_default\_groups);**  
 init\_ssl();  
server\_init();  
**init\_server\_components();**  
 start\_signal\_handler(); // Creates pidfile  
**acl\_init((THD \* ) 0, opt\_noacl);**  
 init\_slave();  
create\_shutdown\_thread();  
create\_maintenance\_thread();  
**handle\_connections\_sockets(0);**  
 DBUG\_PRINT("quit", ("Exiting main thread"));  
exit(0);  
\}

这个是服务器进程执行的开始。高亮的部分是大家应该更感兴趣的部分。

init\_common\_variables()处理mysqld或者mysqld\_safe传进来的命令行参数以及MySQL的配置文件。

快速浏览一下init\_server\_components()和acl\_init()：init\_server\_components()确保MYSQL\_LOG对象在线并且工作，

acl\_init()让访问权限控制系统启动并运行，包括把权限缓存到内存中。MySQL会通过create\_maintenance\_thread()和create\_shutdown\_thread()分别创建一个独立的线程进行任务维护和停机时间处理。

handle\_connections\_sockets()函数是所有处理开始的地方，我们继续看：

handle\_connections\_sockets(arg attribute((unused))) \{  
if (ip\_sock != INVALID\_SOCKET) \{  
**FD\_SET(ip\_sock, & clientFDs);**  
 DBUG\_PRINT("general", ("Waiting for connections."));  
while (!abort\_loop) \{  
**new\_sock = accept(sock, my\_reinterpret\_cast(struct sockaddr \* )( & cAddr), & length);**  
 thd = new THD;  
if (sock == unix\_sock) thd - > host = (char \* ) my\_localhost;  
**create\_new\_thread(thd);**  
 \}  
\}  
\}

基本的原理就是：mysql.sock被绑定用于监听用户请求。如果接收到一个连接请求，一个新的THD的结构就会被创建，然后被传递给create\_new\_thread()函数。我们接着看create\_new\_thread：

static void create\_new\_thread(THD \* thd) \{  
DBUG\_ENTER("create\_new\_thread"); /\* don't allow too many connections \*/  
if (thread\_count - delayed\_insert\_threads >= max\_connections + 1 || abort\_loop) \{  
DBUG\_PRINT("error", ("Too many connections"));  
close\_connection(thd, ER\_CON\_COUNT\_ERROR, 1);  
delete thd;  
DBUG\_VOID\_RETURN;  
\}  
 **pthread\_mutex\_lock( & LOCK\_thread\_count);**  
 if (cached\_thread\_count > wake\_thread) \{  
 **start\_cached\_thread(thd);**  
 \} else \{  
 **thread\_count++;  
thread\_created++;**  
 if (thread\_count - delayed\_insert\_threads > max\_used\_connections) max\_used\_connections = thread\_count - delayed\_insert\_threads;  
DBUG\_PRINT("info", (("creating thread %d"), thd - > thread\_id));  
 **pthread\_create( & thd - > real\_id, & connection\_attrib, /  
handle\_one\_connection, (void \* ) thd))(void) pthread\_mutex\_unlock( & LOCK\_thread\_count);**  
\}  
DBUG\_PRINT("info", ("Thread created"));  
\}

我们接着看看这些高亮的函数。首先是使用pthread\_mutex\_lock()函数对LOCK\_thread\_count资源进行加锁。这是至关重要的，因为thread\_count和thread\_created变量在函数执行期间会被修改。thread\_count和thread\_created是被所有服务器进程中的线程共享。在create\_new\_thread执行时，pthread\_mutex\_lock通过加锁阻止其它线程修改它们的内存。这是MySQL资源管理子系统的典型工作方式。

start\_cached\_thread启动连接线程池机制。最后并且是最重要的是，pthread\_create()（线程库的一个函数）创建一个线程，并接收传递进来的THD -> real\_id成员变量和handle\_one\_connection()函数指针（用来处理单个连接的创建），handle\_one\_connection()的实现参见 /sql/sql\_parse.cc

/sql/sql\_parse.cc handle\_one\_connection()

handle\_one\_connection(THD \* thd)

\{  
while (!net-> error && net-> vio != 0 && !(thd-> killed == THD::KILL\_CONNECTION)) \{

**if (do\_command(thd)) break;**  
 \}  
\}

为了简单期间，我们省略了很多函数里面的代码。保留的函数聚焦在THD结构的初始化。需要net-> error被校验为真，意味着THD的net成员变量在loop循环中需要使用，换句话说do\_command()一定在发送和接受数据包。

bool do\_command(THD \* thd)

**char \* packet;**

ulong packet\_length;

NET \* net;

**enum enum\_server\_command command;**

packet = 0;

**net\_new\_transaction(net);**

packet\_length = my\_net\_read(net);

packet = (char \* ) net - > read\_pos;

**command = (enum enum\_server\_command)(uchar) packet\[0\];**

DBUG\_RETURN(dispatch\_command(command, thd, packet + 1, (uint) packet\_length));  
\}

现在开始我们已经开始逐渐深入了。首先需要记住数据包的发送是通过MySQL网络子系统的通讯协议完成的。net\_new\_transaction()开始启动时会初始化服务端和客户端的第一个数据包通信。客户端使用传递过来的netstruct，并将net buffers进行填充后回送给服务端。服务端通过调用my\_net\_read，获得客户端包的长度，并且将数据包的内容填充进入net - > read\_posbuffer。网络子系统就完成了他的光荣使命了。

然后我们来看command变量。它和THD指针，数据包变量（查询的SQL语句就在里面），数据包长度一并被传递给了dispatch\_command。

/sql/sql\_parse.cc dispatch\_command()

bool dispatch\_command(enum enum\_server\_command command, THD \* thd, char \* packet, uint packet\_length) \{  
switch (command) \{

// ... omitted  
case COM\_TABLE\_DUMP:

case COM\_CHANGE\_USER:

// ... omitted  
 **case COM\_QUERY:**

if (**alloc\_query(thd, packet, packet\_length)**)

break; // fatal error is set  
 **mysql\_log.write(thd, command, "%s", thd - > query);**

**mysql\_parse(thd, thd - > query, thd - > query\_length);**

// ... omitted  
\}

就象函数名字解释的那样，它所作的就是把query分配给合适的处理句柄。我们进入COM\_QUERY部分，alloc\_query()简单的将数据包中的内容传递给THD-> querymember成员变量，并且为线程分配需要的内存。然后用log模块记录了这个请求。最后我们看一下mysql\_parse()的调用，其实这个名称是不准确的，mysql\_parse()实际上也会执行查询工作。

/sql/sql\_parse.cc mysql\_parse()

void mysql\_parse(THD \* thd, char \* inBuf, uint length)

if (**query\_cache\_send\_result\_to\_client(thd, inBuf, length) <= 0**)

LEX \* lex = thd - > lex;

**yyparse((void \* ) thd);**

**mysql\_execute\_command(thd);**

**query\_cache\_end\_of\_result(thd);**

DBUG\_VOID\_RETURN;  
\}

首先服务器检查查询缓存是否保存了一个和当前查询请求结果相同的查询请求。如果没有命中，THD就会被传递给yyparse()（Bison生成的分析器）进行解析。yyparse返回给THD-> Lexstruct经过优化的执行路径。上述执行完毕后，mysql\_execute\_command()就开始执行。需要注意一点，在查询执行完毕后，query\_cache\_end\_of\_result()函数紧接着被调用。此函数功能就是让查询缓存知道thd完成了处理。接下来我们看一下查询缓存如何保存返回的结果集。

/sql/sql\_parse.cc mysql\_execute\_command()

bool mysql\_execute\_command(THD \* thd)

all\_tables = lex - > query\_tables;

**statistic\_increment(thd - > status\_var.com\_stat\[lex - > sql\_command\],**

**& LOCK\_status);**

switch (lex - > sql\_command) \{

case SQLCOM\_SELECT:  
\{

select\_result \* result = lex - > result;

**check\_table\_access(thd, lex - > exchange ? SELECT\_ACL | FILE\_ACL : SELECT\_ACL, all\_tables, 0);**

**open\_and\_lock\_tables(thd, all\_tables);  
query\_cache\_store\_query(thd, all\_tables);  
res = handle\_select(thd, lex, result);**  
 break;

\}  
case SQLCOM\_PREPARE:  
case SQLCOM\_EXECUTE:

// ...  
default:  
/\* Impossible \*/  
send\_ok(thd);  
break;

\}  
\}

进入mysql\_execute\_command()，我们可以看到一系列的有趣事情。首先，我们来看一下statistic\_increment()是如何更新统计信息的。先统计SELECT语句的com\_stat变量。然后，执行子系统通过check\_table\_access()和权限控制子系统进行交互。它将检查THD中执行的查询对于查询所需要的表是否具有权限。最有趣的是open\_and\_lock\_tables这个进程。我们先不深入进去，仅仅描述一下它的功能：对用户的连接线程建立表缓存，并对需要的所有表进行加锁。接下来我们看一下query\_cache\_store\_query()。查询缓存使用内部的HASH算法对请求的查询进行存储。最后，调用handle\_select()处理查询, 它就是存储引擎抽象层的第一个主要标识。

/sql /sql\_select.cc handle\_select()

bool handle\_select(THD \* thd, LEX \* lex, select\_result \* result) \{

res = **mysql\_select**(thd, & select\_lex - > ref\_pointer\_array, (TABLE\_LIST \* ) select\_lex - > table\_list.first, select\_lex - > with\_wild, select\_lex - > item\_list, select\_lex - > where, select\_lex - > order\_list.elements + select\_lex - > group\_list.elements, (ORDER \* ) select\_lex - > order\_list.first, (ORDER \* ) select\_lex - > group\_list.first,

select\_lex - > having, (ORDER \* ) lex - > proc\_list.first, select\_lex - > options | thd - > options, result, unit, select\_lex);

DBUG\_RETURN(res);  
\}

handle\_select除了对mysql\_select()函数进行简单包裹之外，其它什么也没有做。

/sql/sql\_select.cc mysql\_select()

bool mysql\_select(THD \* thd, Item \* \* \* rref\_pointer\_array, TABLE\_LIST \* tables, uint wild\_num, List < Item > & fields, COND \* conds, uint og\_num, ORDER \* order, ORDER \* group, Item \* having, ORDER \* proc\_param, ulong select\_options, select\_result \* result, SELECT\_LEX\_UNIT \* unit, SELECT\_LEX \* select\_lex)

JOIN \* join;  
join = new JOIN(thd, fields, select\_options, result);  
 **join - > prepare**(rref\_pointer\_array, tables, wild\_num,

conds, og\_num, order, group, having, proc\_param,

select\_lex, unit));  
 **join - > optimize();  
join - > exec();**

mysql\_select()也简单的讲执行的功能交给JOIN对象。执行之前，join先执行optimize()。我们接下来看一下join::exec的实现

/sql/sql\_select.cc JOIN: exec()

void JOIN::exec()

\{  
error = do\_select(curr\_join, curr\_fields\_list, NULL, procedure);  
thd - > limit\_found\_rows = curr\_join - > send\_records;  
thd - > examined\_row\_count = curr\_join - > examined\_rows;

JOIN::exec()看起来又象另外一个包裹函数。JOIN::exec()简单的调用do\_select()函数来处理大部分工作。但是，我们需要知道一旦do\_select()返回，我们将结果计数的信息赋值给THD的成员变量。接下来我们进入do\_select():

/sql/sql\_select.cc do\_select()

static int do\_select(JOIN \* join, List < Item > \* fields, TABLE/ \* table, Procedure \* procedure)

\{  
JOIN\_TAB \* join\_tab;  
 **sub\_select(join, join\_tab, 0);  
join - > result - > send\_eof())**

我们貌似看到希望了。JOIN对象的结果成员变量在调用sub\_select()后发送了一个EOF标识，所以我们要深入进去。通过这个行为，它看起来sub\_select()函数应该使用结果对join对象的结果成员变量进行赋值。好的，我们看看这个猜想是否正确：

/ sql / sql\_select.cc sub\_select()

static int sub\_select(JOIN \* join, JOIN\_TAB \* join\_tab, bool end\_of\_records)

\{  
join\_init\_read\_record(join\_tab);  
READ\_RECORD \* info = & join\_tab - > read\_record;

join - > thd - > row\_count = 0;  
 **do \{**

**join - > examined\_rows++;  
join - > thd - > row\_count++;**

**\} while (info - > read\_record(info)));**  
\}  
return 0;

sub\_select()函数的关键是do…while循环，此循环直到READ\_RECORD结构体变量停止调用它的read\_record()成员函数后结束。我们接着看：

READ\_RECORD结构体在 /sql /structs.h中进行定义。它定义了MySQL内部格式的记录。

首先join\_init\_read\_record()函数是连接存储引擎抽象层的纽带。它初始化JOIN\_TAB结构体中的记录然后将READ\_RECORD对象赋值给read\_record成员变量。

/sql/sql\_select.cc join\_init\_read\_record()

static int join\_init\_read\_record(JOIN\_TAB \* tab) \{  
**init\_read\_record**( & tab - > read\_record, tab - > join - > thd, tab - > table, tab - > select, 1, 1);  
return ( \* tab - > read\_record.read\_record)( & tab - > read\_record);  
\}

它直接调用init\_read\_record()函数，然后返回从表中获取的记录数目。这就是它所做的全部，那么存储引擎和记录缓存在哪里进行交互呢？看看init\_read\_record()就清楚了。

/sql/records.cc init\_read\_record()

void init\_read\_record(READ\_RECORD \* info, THD \* thd, TABLE \* table,

SQL\_SELECT \* select,

int **use\_record\_cache**, bool print\_error) \{

**info - > read\_record = rr\_sequential;**

**table - > file - > ha\_rnd\_init(1);**  
\}

这里完成了两件重要的事情。首先，READ\_RECORD类型指针的read\_record成员变量被赋值为rr\_sequential。意味着，后面对于info - > read\_record的调用将会转化成为rr\_sequential(READ\_RECORD \* info)，它会使用**记录缓存来获取数据**。现在需要记住，所有loop循环对read\_record()的调用将会从现在开始命中记录缓存。我们现在来关注对ha\_rnd\_init()函数的调用，无论何时你看见ha\_开头的函数，你立即应该知道你在使用表（存储引擎函数）的句柄函数。第一个猜测是这个函数用来从存储引擎的磁盘中扫描记录段。所以，让我们来分析ha\_rnd\_init().为什么只有头文件？因为句柄类仅仅是存储引擎子类实现的的接口。通过类的定义我们看到一个单例方法被定义了。

/sql/handler.h handler::ha\_rnd\_init()

int ha\_rnd\_init(bool scan)

DBUG\_ENTER("ha\_rnd\_init");

DBUG\_ASSERT(inited == NONE || (inited == RND && scan));

inited = RND;

DBUG\_RETURN(rnd\_init(scan));

既然我们在一张MyISAM表上进行查询，我们来看一下ha\_myisam句柄类的rnd\_init()虚函数的声明。

/sql/ha\_myisam.cc ha\_myisam::rnd\_init()

int ha\_myisam::rnd\_init(bool scan) \{

if (scan)

return mi\_scan\_init(file);  
// …  
\}

现在我们已经可以肯定了，rnd\_init方法包括了对表记录的扫描。具体的实现在mi\_scan\_init()函数中。

/myisam/mi\_scan.c mi\_scan\_init()

int mi\_scan\_init(register MI\_INFO \* info) \{

**info - > nextpos = info - > s - > pack.header\_length; /\* Read first record \*/**  
 // …  
\}

不可思议的是所有的工作仅仅是把一条记录读入READ\_RECORD结构体中！幸运的是，我们已经基本上完结了。

/sql/records.cc rr\_sequential()

static int rr\_sequential(READ\_RECORD \* info)

while ((tmp = info - > file - > rnd\_next(info - > record)))

if (tmp == HA\_ERR\_END\_OF\_FILE)

tmp = -1;

return tmp;  
\}

当sub\_select()调用read\_record()成员函数的时候，这个函数都会被调用。它依次调用另外一个MyISAM句柄方法，rnd\_next()，简单的把当前记录指针赋值到需要的READ\_RECORD结构体中。这是因为rnd\_next已经被简单的映射到mi\_scan函数实现上。

/myisam/mi\_scan.c mi\_scan()

int mi\_scan(MI\_INFO \* info, byte \* buf) \{  
// …  
info - > update &= (HA\_STATE\_CHANGED | HA\_STATE\_ROW\_CHANGED);

DBUG\_RETURN(( \* info - > s - > read\_rnd)(info, buf, info - > nextpos, 1));  
\}

这样，记录缓存的行为更多的像是一个句柄的包裹库而不是一个缓存。其实不然我们省略掉了对共享IO\_CACHE对象的介绍。

这基本上就是一条基本查询的整个执行过程。

[clip_image002]: http://hi.csdn.net/attachment/201012/29/0_129358578669cR.gif
[clip_image002_clip_image002]: /images/20220708/3b8a63c03c4c4f4884961aa3051f1c76.png