技术分享 | 从实现原理来看为什么 Clone 插件比 Xtrabackup 更好用？

Clone_Handle::process_chunk(Clone_Task*, unsigned int, unsigned int, Ha_clone_cbk*) (/mysql-8.0.33/storage/innobase/clone/clone0copy.cc:1440)
Clone_Handle::copy(unsigned int, Ha_clone_cbk*) (/mysql-8.0.33/storage/innobase/clone/clone0copy.cc:1379)
innodb_clone_copy(handlerton*, THD*, unsigned char const*, unsigned int, unsigned int, Ha_clone_cbk*) (/mysql-8.0.33/storage/innobase/clone/clone0api.cc:561)
hton_clone_copy(THD*, std::__1::vector<myclone::Locator, std::__1::allocator<myclone::Locator>>&, std::__1::vector<unsigned int, std::__1::allocator<unsigned int>>&, Ha_clone_cbk*) (/mysql-8.0.33/plugin/clone/src/clone_hton.cc:152)
myclone::Local::clone_exec() (/mysql-8.0.33/plugin/clone/src/clone_local.cc:172)
myclone::Local::clone() (/mysql-8.0.33/plugin/clone/src/clone_local.cc:73)
plugin_clone_local(THD*, char const*) (/mysql-8.0.33/plugin/clone/src/clone_plugin.cc:456)
Clone_handler::clone_local(THD*, char const*) (/mysql-8.0.33/sql/clone_handler.cc:135)
Sql_cmd_clone::execute(THD*) (/mysql-8.0.33/sql/sql_admin.cc:2017)
mysql_execute_command(THD*, bool) (/mysql-8.0.33/sql/sql_parse.cc:4714)
dispatch_sql_command(THD*, Parser_state*) (/mysql-8.0.33/sql/sql_parse.cc:5363)
dispatch_command(THD*, COM_DATA const*, enum_server_command) (/mysql-8.0.33/sql/sql_parse.cc:2050)
do_command(THD*) (/mysql-8.0.33/sql/sql_parse.cc:1439)
handle_connection(void*) (/mysql-8.0.33/sql/conn_handler/connection_handler_per_thread.cc:302)
pfs_spawn_thread(void*) (/mysql-8.0.33/storage/perfschema/pfs.cc:3042)
_pthread_start (@_pthread_start:40)

buf0flu.cc
if (flush) {
    /* We are committed to flushing by the time we get here */

    mutex_enter(&buf_pool->flush_state_mutex);
....

    arch_page_sys->track_page(bpage, buf_pool->track_page_lsn, frame_lsn,
                                false);
}

/** Check and add page ID to archived data.
Check for duplicate page.
@param[in]      bpage           page to track
@param[in]      track_lsn       LSN when tracking started
@param[in]      frame_lsn       current LSN of the page
@param[in]      force           if true, add page ID without check */
void Arch_Page_Sys::track_page(buf_page_t *bpage, lsn_t track_lsn,
                               lsn_t frame_lsn, bool force) {
  Arch_Block *cur_blk;
  uint count = 0;
  ... ...

  /* We need to track this page. */
  arch_oper_mutex_enter();

  while (true) {
    if (m_state != ARCH_STATE_ACTIVE) {
      break;
    }
    ... ...

    cur_blk = m_data.get_block(&m_write_pos, ARCH_DATA_BLOCK);

    if (cur_blk->get_state() == ARCH_BLOCK_ACTIVE) {
      if (cur_blk->add_page(bpage, &m_write_pos)) {
        /* page added successfully. */
        break;
      }

      /* Current block is full. Move to next block. */
      cur_blk->end_write();

      m_write_pos.set_next();

      /* Writing to a new file so move to the next reset block. */
      if (m_write_pos.m_block_num % ARCH_PAGE_FILE_DATA_CAPACITY == 0) {
        Arch_Block *reset_block =
            m_data.get_block(&m_reset_pos, ARCH_RESET_BLOCK);
        reset_block->end_write();

        m_reset_pos.set_next();
      }

      os_event_set(page_archiver_thread_event);

      ++count;
      continue;

    } else if (cur_blk->get_state() == ARCH_BLOCK_INIT ||
               cur_blk->get_state() == ARCH_BLOCK_FLUSHED) {
      ut_ad(m_write_pos.m_offset == ARCH_PAGE_BLK_HEADER_LENGTH);

      cur_blk->begin_write(m_write_pos);

      if (!cur_blk->add_page(bpage, &m_write_pos)) {
        /* Should always succeed. */
        ut_d(ut_error);
      }

      /* page added successfully. */
      break;

    } else {
      bool success;
      ... ...
      /* Might release operation mutex temporarily. Need to
      loop again verifying the state. */
      success = wait_flush_archiver(cbk);
      count = success ? 0 : 2;

      continue;
    }
  }
  arch_oper_mutex_exit();
}

int Page_Arch_Client_Ctx::start(bool recovery, uint64_t *start_id) {
  ... ...
  /* Start archiving. */
  err = arch_page_sys->start(&m_group, &m_last_reset_lsn, &m_start_pos,
                             m_is_durable, reset, recovery);
  ... ...
}

int Arch_Page_Sys::start(Arch_Group **group, lsn_t *start_lsn,
                         Arch_Page_Pos *start_pos, bool is_durable,
                         bool restart, bool recovery) {
   ... ...
    
   log_sys_lsn = (recovery ? m_last_lsn : log_get_lsn(*log_sys));
   /* Enable/Reset buffer pool page tracking. */
        set_tracking_buf_pool(log_sys_lsn); //  page_id
   ... ...
   auto err = start_page_archiver_background();  sp_id, page_id
   ... ...
   if (!recovery) {
    /* Request checkpoint */
    log_request_checkpoint(*log_sys, true);  checkpoint
  }
}

/** Archiver background thread */
void page_archiver_thread() {
  bool page_wait = false;

  ... ... 

  while (true) {
    /* Archive in memory data blocks to disk. */
    auto page_abort = arch_page_sys->archive(&page_wait);

    if (page_abort) {
      ib::info(ER_IB_MSG_14) << "Exiting Page Archiver";
      break;
    }

    if (page_wait) {
      /* Nothing to archive. Wait until next trigger. */
      os_event_wait(page_archiver_thread_event);
      os_event_reset(page_archiver_thread_event);
    }
  }
}

bool Arch_Page_Sys::archive(bool *wait) {
  ... ...

  db_err = flush_blocks(wait);

  if (db_err != DB_SUCCESS) {
    is_abort = true;
  }

  ... ...
  return (is_abort);
}

dberr_t Arch_Page_Sys::flush_blocks(bool *wait) {
  ... ...
  err = flush_inactive_blocks(cur_pos, end_pos);
  ... ...
}

dberr_t Arch_Page_Sys::flush_inactive_blocks(Arch_Page_Pos &cur_pos,
                                             Arch_Page_Pos end_pos) {
  
  /* Write all blocks that are ready for flushing. */
  while (cur_pos.m_block_num < end_pos.m_block_num) {
    cur_blk = m_data.get_block(&cur_pos, ARCH_DATA_BLOCK);

    err = cur_blk->flush(m_current_group, ARCH_FLUSH_NORMAL);

    if (err != DB_SUCCESS) {
      break;
    }

   ... ...
  }

  return (err);
}

/** Flush this block to the file group.
@param[in]      file_group      current archive group
@param[in]      type            flush type
@return error code. */
dberr_t Arch_Block::flush(Arch_Group *file_group, Arch_Blk_Flush_Type type) {
  ... ...
  switch (m_type) {
    case ARCH_RESET_BLOCK:
      err = file_group->write_file_header(m_data, m_size);
      break;

    case ARCH_DATA_BLOCK: {
      bool is_partial_flush = (type == ARCH_FLUSH_PARTIAL);

      /* Callback responsible for setting up file's header starting at offset 0.
      This header is left empty within this flush operation. */
      auto get_empty_file_header_cbk = [](uint64_t, byte *) {
        return DB_SUCCESS;
      };

      /* We allow partial flush to happen even if there were no pages added
      since the last partial flush as the block's header might contain some
      useful info required during recovery. */
      err = file_group->write_to_file(nullptr, m_data, m_size, is_partial_flush,
                                      true, get_empty_file_header_cbk);
      break;
    }

    default:
      ut_d(ut_error);
  }

  return (err);
}

dberr_t Arch_Group::write_to_file(Arch_File_Ctx *from_file, byte *from_buffer,
                                  uint length, bool partial_write,
                                  bool do_persist,
                                  Get_file_header_callback get_header) {
  ... ...
    if (do_persist) {
      Arch_Page_Dblwr_Offset dblwr_offset =
          (partial_write ? ARCH_PAGE_DBLWR_PARTIAL_FLUSH_PAGE
                         : ARCH_PAGE_DBLWR_FULL_FLUSH_PAGE);

      /** Write to the doublewrite buffer before writing archived data to a file.
      The source is either a file context or buffer. Caller must ensure that data
  is in single file in source file context. **/
      Arch_Group::write_to_doublewrite_file(from_file, from_buffer, write_size,
                                            dblwr_offset);
    }

   ... ...

  return (DB_SUCCESS);
}