redo log write和flush

2022-11-11 20:16:50

http://bbs.chinaunix.net/thread-1753130-1-1.html

在事务提交时innobase会调用ha_innodb.cc 中的innobase_commit，而innobase_commit通过调用trx_commit_complete_for_mysql（trx0trx.c)来调用log_write_up_to（log0log.c),也就是当innobase提交事务的时候就会调用log_write_up_to来写redo log
innobase_commit中

  1. if (all              # 如果是事务提交
  2. || (!thd_test_options(thd, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))) {

复制代码

通过下面的代码实现事务的commit串行化

  1. if (innobase_commit_concurrency > 0) {
  2. pthread_mutex_lock(&commit_cond_m);
  3. commit_threads++;
  4. if (commit_threads > innobase_commit_concurrency) {
  5. commit_threads--;
  6. pthread_cond_wait(&commit_cond,
  7. &commit_cond_m);
  8. pthread_mutex_unlock(&commit_cond_m);
  9. goto retry;
  10. }
  11. else {
  12. pthread_mutex_unlock(&commit_cond_m);
  13. }
  14. }

复制代码

  1. trx->flush_log_later = TRUE;   # 在做提交操作时禁止flush binlog 到磁盘
  2. innobase_commit_low(trx);
  3. trx->flush_log_later = FALSE;

复制代码

先略过innobase_commit_low调用 ,下面开始调用trx_commit_complete_for_mysql做write日志操作

  1. trx_commit_complete_for_mysql(trx);  #开始flush  log
  2. trx->active_trans = 0;

复制代码

在trx_commit_complete_for_mysql中，主要做的是对系统参数srv_flush_log_at_trx_commit值做判断来调用
log_write_up_to，或者write redo log file或者write&&flush to disk

  1. if (!trx->must_flush_log_later) {
  2. /* Do nothing */
  3. } else if (srv_flush_log_at_trx_commit == 0) {  #flush_log_at_trx_commit=0，事务提交不写redo log
  4. /* Do nothing */
  5. } else if (srv_flush_log_at_trx_commit == 1) { #flush_log_at_trx_commit=1,事务提交写log并flush磁盘,如果flush方式不是SRV_UNIX_NOSYNC （这个不是很熟悉）
  6. if (srv_unix_file_flush_method == SRV_UNIX_NOSYNC) {
  7. /* Write the log but do not flush it to disk */
  8. log_write_up_to(lsn, LOG_WAIT_ONE_GROUP, FALSE);
  9. } else {
  10. /* Write the log to the log files AND flush them to
  11. disk */
  12. log_write_up_to(lsn, LOG_WAIT_ONE_GROUP, TRUE);
  13. }
  14. } else if (srv_flush_log_at_trx_commit == 2) {   #如果是2，则只write到redo log
  15. /* Write the log but do not flush it to disk */
  16. log_write_up_to(lsn, LOG_WAIT_ONE_GROUP, FALSE);
  17. } else {
  18. ut_error;
  19. }

复制代码

那么下面看log_write_up_to

  1. if (flush_to_disk              #如果flush到磁盘，则比较当前commit的lsn是否大于已经flush到磁盘的lsn
  2. && ut_dulint_cmp(log_sys->flushed_to_disk_lsn, lsn) >= 0) {
  3. mutex_exit(&(log_sys->mutex));
  4. return;
  5. }
  6. if (!flush_to_disk       #如果不flush磁盘则比较当前commit的lsn是否大于已经写到所有redo log file的lsn,或者在只等一个group完成条件下是否大于已经写到某个redo file的lsn
  7. && (ut_dulint_cmp(log_sys->written_to_all_lsn, lsn) >= 0
  8. || (ut_dulint_cmp(log_sys->written_to_some_lsn, lsn)
  9. >= 0
  10. && wait != LOG_WAIT_ALL_GROUPS))) {
  11. mutex_exit(&(log_sys->mutex));
  12. return;
  13. }
  14. #下面的代码判断是否log在write,有的话等待其完成
  15. if (log_sys->n_pending_writes > 0) {
  16. if (flush_to_disk    # 如果需要刷新到磁盘，如果正在flush的lsn包括了commit的lsn，只要等待操作完成就可以了
  17. && ut_dulint_cmp(log_sys->current_flush_lsn, lsn)
  18. >= 0) {
  19. goto do_waits;
  20. }
  21. if (!flush_to_disk  # 如果是刷到redo log file的那么如果在write的lsn包括了commit的lsn,也只要等待就可以了
  22. && ut_dulint_cmp(log_sys->write_lsn, lsn) >= 0) {
  23. goto do_waits;
  24. }
  25. ......
  26. if (!flush_to_disk  # 如果在当前IO空闲情况下 ，而且不需要flush到磁盘，那么 如果下次写的位置已经到达buf_free位置说明wirte操作都已经完成了，直接返回
  27. && log_sys->buf_free == log_sys->buf_next_to_write) {
  28. mutex_exit(&(log_sys->mutex));
  29. return;
  30. }

复制代码

下面取到group,设置相关write or flush相关字段，并且得到起始和结束位置的block号

  1. log_sys->n_pending_writes++;
  2. group = UT_LIST_GET_FIRST(log_sys->log_groups);
  3. group->n_pending_writes++;      /* We assume here that we have only
  4. one log group! */
  5. os_event_reset(log_sys->no_flush_event);
  6. os_event_reset(log_sys->one_flushed_event);
  7. start_offset = log_sys->buf_next_to_write;
  8. end_offset = log_sys->buf_free;
  9. area_start = ut_calc_align_down(start_offset, OS_FILE_LOG_BLOCK_SIZE);
  10. area_end = ut_calc_align(end_offset, OS_FILE_LOG_BLOCK_SIZE);
  11. ut_ad(area_end - area_start > 0);
  12. log_sys->write_lsn = log_sys->lsn;
  13. if (flush_to_disk) {
  14. log_sys->current_flush_lsn = log_sys->lsn;
  15. }

复制代码

log_block_set_checkpoint_no调用设置end_offset所在block的LOG_BLOCK_CHECKPOINT_NO为log_sys中下个检查点号

  1. log_block_set_flush_bit(log_sys->buf + area_start, TRUE);   # 这个没看明白
  2. log_block_set_checkpoint_no(
  3. log_sys->buf + area_end - OS_FILE_LOG_BLOCK_SIZE,
  4. log_sys->next_checkpoint_no);

复制代码

保存不属于end_offset但在其所在的block中的数据到下一个空闲的block

  1. ut_memcpy(log_sys->buf + area_end,
  2. log_sys->buf + area_end - OS_FILE_LOG_BLOCK_SIZE,
  3. OS_FILE_LOG_BLOCK_SIZE);

复制代码

对于每个group调用log_group_write_buf写redo log buffer

  1. while (group) {
  2. log_group_write_buf(
  3. group, log_sys->buf + area_start,
  4. area_end - area_start,
  5. ut_dulint_align_down(log_sys->written_to_all_lsn,
  6. OS_FILE_LOG_BLOCK_SIZE),
  7. start_offset - area_start);
  8. log_group_set_fields(group, log_sys->write_lsn);   # 计算这次写的lsn和offset来设置group->lsn和group->lsn_offset
  9. group = UT_LIST_GET_NEXT(log_groups, group);
  10. }
  11. ......
  12. if (srv_unix_file_flush_method == SRV_UNIX_O_DSYNC) {  # 这个是什么东西
  13. /* O_DSYNC means the OS did not buffer the log file at all:
  14. so we have also flushed to disk what we have written */
  15. log_sys->flushed_to_disk_lsn = log_sys->write_lsn;
  16. } else if (flush_to_disk) {
  17. group = UT_LIST_GET_FIRST(log_sys->log_groups);
  18. fil_flush(group->space_id);         # 最后调用fil_flush执行flush到磁盘
  19. log_sys->flushed_to_disk_lsn = log_sys->write_lsn;
  20. }

复制代码

接下来看log_group_write_buf做了点什么

在log_group_calc_size_offset中,从group中取到上次记录的lsn位置（注意是log files组成的1个环状buffer),并计算这次的lsn相对于上次的差值

  1. # 调用log_group_calc_size_offset计算group->lsn_offset除去多个LOG_FILE头部长度后的大小，比如lsn_offset落在第3个log file上，那么需要减掉3*LOG_FILE_HDR_SIZE的大小
  2. gr_lsn_size_offset = (ib_longlong)
  3. log_group_calc_size_offset(group->lsn_offset, group);
  4. group_size = (ib_longlong) log_group_get_capacity(group); # 计算group除去所有LOG_FILE_HDR_SIZE长度后的DATA部分大小
  5. # 下面是典型的环状结构差值计算
  6. if (ut_dulint_cmp(lsn, gr_lsn) >= 0) {
  7. difference = (ib_longlong) ut_dulint_minus(lsn, gr_lsn);
  8. } else {
  9. difference = (ib_longlong) ut_dulint_minus(gr_lsn, lsn);
  10. difference = difference % group_size;
  11. difference = group_size - difference;
  12. }
  13. offset = (gr_lsn_size_offset + difference) % group_size;
  14. # 最后算上每个log file 头部大小，返回真实的offset
  15. return(log_group_calc_real_offset((ulint)offset, group));

复制代码

接着看

  1. # 如果需要写的内容超过一个文件大小
  2. if ((next_offset % group->file_size) + len > group->file_size) {
  3. write_len = group->file_size                # 写到file末尾
  4. - (next_offset % group->file_size);
  5. } else {
  6. write_len = len;                              # 否者写len个block
  7. }
  8. # 最后真正的内容就是写buffer了，如果跨越file的话另外需要写file log file head部分
  9. if ((next_offset % group->file_size == LOG_FILE_HDR_SIZE)
  10. && write_header) {
  11. /* We start to write a new log file instance in the group */
  12. log_group_file_header_flush(group,
  13. next_offset / group->file_size,
  14. start_lsn);
  15. srv_os_log_written+= OS_FILE_LOG_BLOCK_SIZE;
  16. srv_log_writes++;
  17. }
  18. # 调用fil_io来执行buffer写
  19. if (log_do_write) {
  20. log_sys->n_log_ios++;
  21. srv_os_log_pending_writes++;
  22. fil_io(OS_FILE_WRITE | OS_FILE_LOG, TRUE, group->space_id,
  23. next_offset / UNIV_PAGE_SIZE,
  24. next_offset % UNIV_PAGE_SIZE, write_len, buf, group);
  25. srv_os_log_pending_writes--;
  26. srv_os_log_written+= write_len;
  27. srv_log_writes++;
  28. }

复制代码

上一篇：在 MVC 控制器中使用 构造函数时行依赖注入 （IoC)


下一篇：【转】如何确定Kafka的分区数、key和consumer线程数