之前的一篇内核月报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. }

流程图

flow-aio.png

源码分析

  • 物理数据页操作入口函数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请求的执行流程进行了梳理，对重点数据结构以及函数进行了分析，希望对读者日后进行源码阅读及修改有所帮助。