之前的一篇内核月报InnoDB IO子系统 中介绍了InnoDB IO子系统中包含的同步IO以及异步IO。本篇文章将从源码层面剖析一下InnoDB IO子系统中，数据页的同步IO以及异步IO请求的具体实现过程。

在MySQL5.6中，InnoDB的异步IO主要是用来处理预读以及对数据文件的写请求的。而对于正常的页面数据读取则是通过同步IO进行的。到底二者在代码层面上的实现过程有什么样的区别？ 接下来我们将以Linux native io的执行过程为主线，对IO请求的执行过程进行梳理。

重点数据结构

• os_aio_array_t

1. /** 用来记录某一类（ibuf,log,read,write）异步IO（aio）请求的数组类型。每一个异步IO请求都会在类型对应的数组中注册一个innodb
2. aio对象。*/

4. os_aio_array_t  {

6. os_ib_mutex_t mutex;  // 主要用来控制异步read/write线程的并发操作。对于ibuf,log类型，由于只有一个线程，所以不存在并发操作问题
7. os_event_t  not_full; // 一个条件变量event，用来通知等待获取slot的线程是否os_aio_array_t数组有空闲的slot供aio请求

9. os_event_t  is_empty; // 条件变量event，用来通知IO线程os_aio_array_t数组是否有pening的IO请求。

11. ulint   n_slots; // 数组容纳的IO请求数。= 线程数 * 每个segment允许pending的请求数（256）

13. ulint   n_segments; // 允许独立wait的segment数，即某种类型的IO的允许最大线程数

15. ulint   cur_seg; /* IO请求会按照round robin的方式分配到不同的segment中，该变量指示下一个IO请求可以分配的segment */
16. ulint   n_reserved; // 已经Pending的IO请求数

18. os_aio_slot_t*  slots; // 用来记录具体的每个IO请求对象的数组，也即n_segments 个线程共用n_slots个槽位来存放pending io请求

20. \#ifdef __WIN__

22. HANDLE*   handles;
23. /*!< Pointer to an array of OS native
24. event handles where we copied the
25. handles from slots, in the same
26. order. This can be used in
27. WaitForMultipleObjects; used only in
28. Windows */

30. \#endif __WIN__

32. \#if defined(LINUX_NATIVE_AIO)

34. io_context_t*   aio_ctx; // aio上下文的数组，每个segment拥有独立的一个aio上下文数组，用来记录以及完成的IO请求上下文

36. struct io_event*  aio_events; // 该数组用来记录已经完成的IO请求事件。异步IO通过设置事件通知IO线程处理完成的IO请求

38. struct iocb**  pending; // 用来记录pending的aio请求

40. ulint*         count; // 该数组记录了每个segment对应的pending aio请求数量

42. \#endif /* LINUX_NATIV_AIO */

44. }

• os_aio_slot_t

1. // os_aio_array_t数组中用来记录一个异步IO(aio)请求的对象
2. os_aio_slot_t {

4. ibool   is_read;  /*!< TRUE if a read operation */

6. ulint   pos;    // os_aio_array_t数组中所在的位置

8. ibool   reserved; // TRUE表示该Slot已经被别的IO请求占用了

10. time_t    reservation_time; // 占用的时间

12. ulint   len;    // io请求的长度

14. byte*   buf;    // 数据读取或者需要写入的buffer，通常指向buffer pool的一个页面，压缩页面有特殊处理

16. ulint   type;   /* 请求类型，即读还是写IO请求 */

18. os_offset_t offset;   /*!< file offset in bytes */

20. os_file_t file;   /*!< file where to read or write */

22. const char* name;   /*!< 需要读取的文件及路径信息 */

24. ibool   io_already_done; /* TRUE表示IO已经完成了

26. fil_node_t* message1; /* 该aio操作的innodb文件描述符（f_node_t）*/

28. void*   message2; /* 用来记录完成IO请求所对应的具体buffer pool bpage页 */

30. \#ifdef WIN_ASYNC_IO

32. HANDLE    handle;   /*!< handle object we need in the
33. OVERLAPPED struct */

35. OVERLAPPED  control;  /*!< Windows control block for the
36. aio request */

38. \#elif defined(LINUX_NATIVE_AIO)

40. struct iocb control;  /* 该slot使用的aio请求控制块iocb */

42. int   n_bytes;  /* 读写bytes */

44. int   ret;    /* AIO return code */

46. \#endif /* WIN_ASYNC_IO */

48. }

流程图

源码分析

• 物理数据页操作入口函数os_aio_func

1. ibool
2. os_aio_func(
3. /*========*/
4. ulint   type, /* IO类型，READ还是WRITE IO */
5. ulint   mode, /* 这里表示是否使用SIMULATED aio执行异步IO请求 */
6. const char* name, /* IO需要打开的tablespace路径+名称 */
7. os_file_t file, /* IO操作的文件 */
8. void*   buf,  // 数据读取或者需要写入的buffer，通常指向buffer pool的一个页面，压缩页面有特殊处理
9. os_offset_t offset, /*!< in: file offset where to read or write */
10. ulint   n,  /* 读取或写入字节数 */
11. fil_node_t* message1, /* 该aio操作的innodb文件描述符（f_node_t），只对异步IO起作用 */
12. void*   message2, /* 用来记录完成IO请求所对应的具体buffer pool bpage页，只对异步IO起作用 */
13. ibool   should_buffer, // 是否需要缓存aio请求，该变量主要对预读起作用
14. ibool   page_encrypt,
15. /*!< in: Whether to encrypt */
16. ulint   page_size)
17. /*!< in: Page size */
18. {
19. ...

21. wake_later = mode & OS_AIO_SIMULATED_WAKE_LATER;
22. mode = mode & (~OS_AIO_SIMULATED_WAKE_LATER);

24. if (mode == OS_AIO_SYNC
25. #ifdef WIN_ASYNC_IO
26. && !srv_use_native_aio
27. #endif /* WIN_ASYNC_IO */
28. ) {
29. /* This is actually an ordinary synchronous read or write:
30. no need to use an i/o-handler thread. NOTE that if we use
31. Windows async i/o, Windows does not allow us to use
32. ordinary synchronous os_file_read etc. on the same file,
33. therefore we have built a special mechanism for synchronous
34. wait in the Windows case.
35. Also note that the Performance Schema instrumentation has
36. been performed by current os_aio_func()'s wrapper function
37. pfs_os_aio_func(). So we would no longer need to call
38. Performance Schema instrumented os_file_read() and
39. os_file_write(). Instead, we should use os_file_read_func()
40. and os_file_write_func() */

42. /* 这里如果是同步IO，并且native io没有开启的情况下，直接使用os_file_read/write函数进行读取，
43. 不需要经过IO线程进行处理 */

45. if (type == OS_FILE_READ) {
46. if (page_encrypt) {
47. return(os_file_read_decrypt_page(file, buf, offset, n, page_size));
48. } else {
49. return(os_file_read_func(file, buf, offset, n));
50. }
51. }
52. ut_ad(!srv_read_only_mode);
53. ut_a(type == OS_FILE_WRITE);
54. if (page_encrypt) {
55. return(os_file_write_encrypt_page(name, file, buf, offset, n, page_size));
56. } else {
57. return(os_file_write_func(name, file, buf, offset, n));
58. }
59. }
60. try_again:
61. switch (mode) {
62. // 根据访问类型，定位IO请求数组
63. case OS_AIO_NORMAL:
64. if (type == OS_FILE_READ) {
65. array = os_aio_read_array;
66. } else {
67. ut_ad(!srv_read_only_mode);
68. array = os_aio_write_array;
69. }
70. break;
71. case OS_AIO_IBUF:
72. ut_ad(type == OS_FILE_READ);
73. /* Reduce probability of deadlock bugs in connection with ibuf:
74. do not let the ibuf i/o handler sleep */

76. wake_later = FALSE;

78. if (srv_read_only_mode) {
79. array = os_aio_read_array;
80. }
81. break;
82. case OS_AIO_LOG:
83. if (srv_read_only_mode) {
84. array = os_aio_read_array;
85. } else {
86. array = os_aio_log_array;
87. }
88. break;
89. case OS_AIO_SYNC:
90. array = os_aio_sync_array;
91. #if defined(LINUX_NATIVE_AIO)
92. /* In Linux native AIO we don't use sync IO array. */
93. ut_a(!srv_use_native_aio);
94. #endif /* LINUX_NATIVE_AIO */
95. break;
96. default:
97. ut_error;
98. array = NULL; /* Eliminate compiler warning */
99. }
100. // 阻塞为当前IO请求申请一个用来执行异步IO的slot
101. slot = os_aio_array_reserve_slot(type, array, message1, message2, file,
102. name, buf, offset, n, page_encrypt, page_size);

104. DBUG_EXECUTE_IF("simulate_slow_aio",
105. {
106. os_thread_sleep(1000000);
107. }
108. );
109. if (type == OS_FILE_READ) {
110. if (srv_use_native_aio) {
111. os_n_file_reads++;
112. os_bytes_read_since_printout += n;
113. #ifdef WIN_ASYNC_IO
114. // 这里是Windows用来处理异步IO读请求
115. ret = ReadFile(file, buf, (DWORD) n, &len,
116. &(slot->control));

118. #elif defined(LINUX_NATIVE_AIO)
119. // 这里是Linux来处理native io
120. if (!os_aio_linux_dispatch(array, slot, should_buffer)) {
121. goto err_exit;
122. #endif /* WIN_ASYNC_IO */
123. } else {
124. if (!wake_later) {
125. // 唤醒simulated aio处理线程
126. os_aio_simulated_wake_handler_thread(
127. os_aio_get_segment_no_from_slot(
128. array, slot));
129. }
130. }
131. } else if (type == OS_FILE_WRITE) {
132. ut_ad(!srv_read_only_mode);
133. if (srv_use_native_aio) {
134. os_n_file_writes++;
135. #ifdef WIN_ASYNC_IO
136. // 这里是Windows用来处理异步IO写请求
137. ret = WriteFile(file, buf, (DWORD) n, &len,
138. &(slot->control));

140. #elif defined(LINUX_NATIVE_AIO)
141. // 这里是Linux来处理native io
142. if (!os_aio_linux_dispatch(array, slot, false)) {
143. goto err_exit;
144. }
145. #endif /* WIN_ASYNC_IO */
146. } else {
147. if (!wake_later) {
148. // 唤醒simulated aio处理线程
149. os_aio_simulated_wake_handler_thread(
150. os_aio_get_segment_no_from_slot(
151. array, slot));
152. }
153. }
154. } else {
155. ut_error;
156. }

158. ...
159. }

• 负责通知Linux内核执行native IO请求的函数os_aio_linux_dispatch

1. static
2. ibool
3. os_aio_linux_dispatch(
4. /*==================*/
5. os_aio_array_t* array,  /* IO请求函数 */
6. os_aio_slot_t*  slot, /* 申请好的slot */
7. ibool           should_buffer)  // 是否需要缓存aio 请求，该变量主要对预读起作用
8. {
9. ...

11. /* Find out what we are going to work with.
12. The iocb struct is directly in the slot.
13. The io_context is one per segment. */

15. // 每个segment包含的slot个数，Linux下每个segment包含256个slot
16. slots_per_segment = array->n_slots / array->n_segments;
17. iocb = &slot->control;
18. io_ctx_index = slot->pos / slots_per_segment;
19. if (should_buffer) {
20. /* 这里也可以看到aio请求缓存只对读请求起作用 */
21. ut_ad(array == os_aio_read_array);

23. ulint n;
24. ulint count;
25. os_mutex_enter(array->mutex);
26. /* There are array->n_slots elements in array->pending, which is divided into
27. * array->n_segments area of equal size.  The iocb of each segment are
28. * buffered in its corresponding area in the pending array consecutively as
29. * they come.  array->count[i] records the number of buffered aio requests in
30. * the ith segment.*/
31. n = io_ctx_index * slots_per_segment
32. + array->count[io_ctx_index];
33. array->pending[n] = iocb;
34. array->count[io_ctx_index] ++;
35. count = array->count[io_ctx_index];
36. os_mutex_exit(array->mutex);
37. // 如果当前segment的slot都已经被占用了，就需要提交一次异步aio请求
38. if (count == slots_per_segment) {
39. os_aio_linux_dispatch_read_array_submit(); //no cover line
40. }
41. // 否则就直接返回
42. return (TRUE);
43. }
44. // 直接提交IO请求到内核
45. ret = io_submit(array->aio_ctx[io_ctx_index], 1, &iocb);
46. ...
47. }

• IO线程负责监控aio请求的主函数fil_aio_wait

1. void
2. fil_aio_wait(
3. /*=========*/
4. ulint segment)  /*!< in: the number of the segment in the aio
5. array to wait for */
6. {
7. ibool   ret;
8. fil_node_t* fil_node;
9. void*   message;
10. ulint   type;

12. ut_ad(fil_validate_skip());

14. if (srv_use_native_aio) { // 使用native io
15. srv_set_io_thread_op_info(segment, "native aio handle");
16. #ifdef WIN_ASYNC_IO
17. ret = os_aio_windows_handle( // Window监控入口
18. segment, 0, &fil_node, &message, &type);
19. #elif defined(LINUX_NATIVE_AIO)
20. ret = os_aio_linux_handle( // Linux native io监控入口
21. segment, &fil_node, &message, &type);
22. #else
23. ut_error;
24. ret = 0; /* Eliminate compiler warning */
25. #endif /* WIN_ASYNC_IO */
26. } else {
27. srv_set_io_thread_op_info(segment, "simulated aio handle");

29. ret = os_aio_simulated_handle( // Simulated aio监控入口
30. segment, &fil_node, &message, &type);
31. }

33. ut_a(ret);
34. if (fil_node == NULL) {
35. ut_ad(srv_shutdown_state == SRV_SHUTDOWN_EXIT_THREADS);
36. return;
37. }
38. srv_set_io_thread_op_info(segment, "complete io for fil node");
39. mutex_enter(&fil_system->mutex);

41. // 到这里表示至少有一个IO请求已经完成，该函数设置状态信息
42. fil_node_complete_io(fil_node, fil_system, type);

44. mutex_exit(&fil_system->mutex);

46. ut_ad(fil_validate_skip());

48. /* Do the i/o handling */
49. /* IMPORTANT: since i/o handling for reads will read also the insert
50. buffer in tablespace 0, you have to be very careful not to introduce
51. deadlocks in the i/o system. We keep tablespace 0 data files always
52. open, and use a special i/o thread to serve insert buffer requests. */

54. if (fil_node->space->purpose == FIL_TABLESPACE) { // 数据文件读写IO
55. srv_set_io_thread_op_info(segment, "complete io for buf page");
56. // IO请求完成后，这里处理buffer pool对应的bpage相关的一些状态信息并根据checksum验证数据的正确性
57. buf_page_io_complete(static_cast<buf_page_t*>(message));
58. } else { // 日志文件的读写IO
59. srv_set_io_thread_op_info(segment, "complete io for log");
60. log_io_complete(static_cast<log_group_t*>(message));
61. }
62. }
63. #endif /* UNIV_HOTBACKUP */

• IO线程负责处理native IO请求的函数os_aio_linux_handle

1. ibool
2. os_aio_linux_handle(ulint    global_seg, // 属于哪个segment
3. fil_node_t**message1, /* 该aio操作的innodb文件描述符（f_node_t）*/
4. void**    message2, /* 用来记录完成IO请求所对应的具体buffer pool bpage页 */
5. ulint*    type){ // 读or写IO
6. // 根据global_seg获得该aio 的os_aio_array_t数组，并返回对应的segment
7. segment = os_aio_get_array_and_local_segment(&array, global_seg);
8. n = array->n_slots / array->n_segments; //获得一个线程可监控的io event数
9. /* Loop until we have found a completed request. */
10. for (;;) {
11. ibool    any_reserved = FALSE;
12. os_mutex_enter(array->mutex);
13. for (i = 0; i < n; ++i) {  // 遍历该线程所发起的所有aio请求
14. slot = os_aio_array_get_nth_slot(
15. array, i + segment * n);
16. if (!slot->reserved) {  // 该slot是否被占用
17. continue;
18. } else if (slot->io_already_done) {  // IO请求已经完成，可以通知主线程返回数据了
19. /* Something for us to work on. */
20. goto found;
21. } else {
22. any_reserved = TRUE;
23. }
24. }
25. os_mutex_exit(array->mutex);
26. // 到这里说明没有找到一个完成的io，则再去collect
27. os_aio_linux_collect(array, segment, n);
28. found:   // 找到一个完成的io，将内容返回
29. *message1 = slot->message1;
30. *message2 = slot->message2; // 返回完成IO所对应的bpage页
31. *type = slot->type;
32. if (slot->ret == 0 && slot->n_bytes == (long) slot->len) {
33. if (slot->page_encrypt
34. && slot->type == OS_FILE_READ) {
35. os_decrypt_page(slot->buf, slot->len, slot->page_size, FALSE);
36. }

38. ret = TRUE;
39. } else {
40. errno = -slot->ret;
41. /* os_file_handle_error does tell us if we should retry
42. this IO. As it stands now, we don't do this retry when
43. reaping requests from a different context than
44. the dispatcher. This non-retry logic is the same for
45. windows and linux native AIO.
46. We should probably look into this to transparently
47. re-submit the IO. */
48. os_file_handle_error(slot->name, "Linux aio");

50. ret = FALSE;
51. }

53. os_mutex_exit(array->mutex);

55. os_aio_array_free_slot(array, slot);
56. return(ret);
57. }

• 等待native IO请求完成os_aio_linux_collect

1. os_aio_linux_collect(os_aio_array_t* array,
2. ulint segment,
3. ulint seg_size){
4. events = &array->aio_events[segment * seg_size]; // 定位segment所对应的io event的数组位置
5. /* 获得该线程的aio上下文数组 */
6. io_ctx = array->aio_ctx[segment];
7. /* Starting point of the segment we will be working on. */
8. start_pos = segment * seg_size;
9. /* End point. */
10. end_pos = start_pos + seg_size;

13. retry:
14. /* Initialize the events. The timeout value is arbitrary.
15. We probably need to experiment with it a little. */
16. memset(events, 0, sizeof(*events) * seg_size);
17. timeout.tv_sec = 0;
18. timeout.tv_nsec = OS_AIO_REAP_TIMEOUT;

20. ret = io_getevents(io_ctx, 1, seg_size, events, &timeout); // 阻塞等待该IO线程所监控的任一IO请求完成

22. if (ret > 0) { // 有IO请求完成
23. for (i = 0; i < ret; i++) {
24. // 记录完成IO的请求信息到对应的os_aio_slot_t 对象
25. os_aio_slot_t*    slot;
26. struct iocb*    control;
27. control = (struct iocb*) events[i].obj; // 获得完成的aio的iocb，即提交这个aio请求的iocb
28. ut_a(control != NULL);
29. slot = (os_aio_slot_t*) control->data; // 通过data获得这个aio iocb所对应的os_aio_slot_t
30. /* Some sanity checks. */
31. ut_a(slot != NULL);
32. ut_a(slot->reserved);
33. os_mutex_enter(array->mutex);
34. slot->n_bytes = events[i].res; // 将该io执行的结果保存到slot里
35. slot->ret = events[i].res2;
36. slot->io_already_done = TRUE; // 标志该io已经完成了，这个标志也是外层判断的条件
37. os_mutex_exit(array->mutex);
38. }
39. return;
40. }
41. …
42. }

综上重点对InnoDB navtive IO读写数据文件从源码角度进行了分析，有兴趣的读者也可以继续了解InnoDB自带的simulated IO的实现过程，原理雷同native IO，只是在实现方式上自己进行了处理。本篇文章对InnoDB IO请求的执行流程进行了梳理，对重点数据结构以及函数进行了分析，希望对读者日后进行源码阅读及修改有所帮助。

原文：http://mysql.taobao.org/monthly/2017/07/10/