cfs/src/FsProc_Journal.cc
// blocking function call that checkpoints all journals.
blocking function call that checkpoints all journals.
blocking function call that checkpoints all journals .
void JournalManager::checkpointAllJournals(FileMng *mng,
                                           std::vector<FsProcWorker *> &workers,
                                           size_t n_workers,
                                           size_t this_worker_idx) {
  PrepareForCheckpointingCtx ctx[n_workers];
  CheckpointInput *per_worker_ci[n_workers];
  volatile bool per_worker_completions[n_workers];
  FsProcWorker *wkr_ptr = nullptr;
  FsProcMessage msg;
  msg.type = static_cast<uint8_t>(PREPARE_FOR_CHECKPOINTING);
  for (size_t i = 0; i < n_workers; i++) {
    per_worker_ci[i] = nullptr;
    per_worker_completions[i] = false;
    if (i == this_worker_idx) continue;  
    ctx[i].ci = &(per_worker_ci[i]);
    ctx[i].completed = &(per_worker_completions[i]);
    msg.ctx = (void *)(&(ctx[i]));
    wkr_ptr = workers[i];
    wkr_ptr->messenger->send_message(wkr_ptr->getWorkerIdx(), msg);
  }
  prepareForCheckpointing(&(per_worker_ci[this_worker_idx]), mng);
  per_worker_completions[this_worker_idx] = true;
  size_t num_successful = 0;
  for (size_t i = 0; i < n_workers; i++) {
    while (per_worker_completions[i] == false) continue;
    if (per_worker_ci[i] != nullptr) num_successful++;
  }
  if (num_successful != n_workers) {
    SPDLOG_ERROR("Some workers failed to prepare for checkpointing");
    notifyAllProcsOnCheckpointComplete(mng, workers, n_workers, this_worker_idx,
                                       per_worker_ci, false);
    return;
  }
  n_checkpoint_requests_issued = 0;
  n_checkpoint_requests_completed = 0;
  n_checkpoint_requests_failed = 0;
  for (size_t i = 0; i < n_workers; i++) {
    CheckpointInput *ci = per_worker_ci[i];
    for (auto it = ci->inodesToCheckpoint.cbegin();
         it != ci->inodesToCheckpoint.cend();) {  
      uint32_t disk_sector_no = FsImpl::ino2SectorNo(it->first);
      int rc =
          spdk_nvme_ns_cmd_write(dev_ns, dev_qpair, it->second, disk_sector_no,
                                 1, checkpointWriteComplete, this, 0);
      if (rc == 0) {
        n_checkpoint_requests_issued++;
        it++;
        continue;
      }
      if (rc != -ENOMEM) {
        SPDLOG_ERROR(
            "Failed to submit inode write request, errno={}, skipping it", rc);
        it++;  
        continue;
      }
      waitForCheckpointWriteCompletions();
    }
    std::unordered_map<uint64_t, void *> bitmap_block_to_data_arr[2];
    bitmap_block_to_data_arr[0] = ci->bitmapsToCheckpoint;
    bitmap_block_to_data_arr[1] = ci->inodeBitmapsToCheckpoint;
    for (auto &bitmap_block_to_data : bitmap_block_to_data_arr) {
      for (auto it = bitmap_block_to_data.cbegin();
           it != bitmap_block_to_data.cend();) {  
        int rc =
            spdk_nvme_ns_cmd_write(dev_ns, dev_qpair, it->second, it->first * 8,
                                   8, checkpointWriteComplete, this, 0);
        if (rc == 0) {
          n_checkpoint_requests_issued++;
          it++;
          continue;
        }
        if (rc != -ENOMEM) {
          SPDLOG_ERROR(
              "Failed to submit bitmap write request, errno={}, skipping it",
              rc);
          it++;  
          continue;
        }
        waitForCheckpointWriteCompletions();
      }
    }
  }
  waitForCheckpointWriteCompletions();
  if (n_checkpoint_requests_failed > 0) {
    SPDLOG_ERROR("Checkpoint failued: I/O errors");
    notifyAllProcsOnCheckpointComplete(mng, workers, n_workers, this_worker_idx,
                                       per_worker_ci, false);
    return;
  }
  SPDLOG_INFO("Checkpoint completed successfully");
  notifyAllProcsOnCheckpointComplete(mng, workers, n_workers, this_worker_idx,
                                     per_worker_ci, true);
}
void JournalManager :: checkpointAllJournals ( FileMng * mng , std :: vector < FsProcWorker * > & workers , size_t n_workers , size_t this_worker_idx ) { PrepareForCheckpointingCtx ctx [ n_workers ] ; CheckpointInput * per_worker_ci [ n_workers ] ; volatile bool per_worker_completions [ n_workers ] ; FsProcWorker * wkr_ptr = nullptr ; FsProcMessage msg ; msg . type = static_cast < uint8_t > ( PREPARE_FOR_CHECKPOINTING ) ; for ( size_t i = 0 ; i < n_workers ; i ++ ) { per_worker_ci [ i ] = nullptr ; per_worker_completions [ i ] = false ; if ( i == this_worker_idx ) continue ; ctx [ i ] . ci = & ( per_worker_ci [ i ] ) ; ctx [ i ] . completed = & ( per_worker_completions [ i ] ) ; msg . ctx = ( void * ) ( & ( ctx [ i ] ) ) ; wkr_ptr = workers [ i ] ; wkr_ptr -> messenger -> send_message ( wkr_ptr -> getWorkerIdx ( ) , msg ) ; } prepareForCheckpointing ( & ( per_worker_ci [ this_worker_idx ] ) , mng ) ; per_worker_completions [ this_worker_idx ] = true ; size_t num_successful = 0 ; for ( size_t i = 0 ; i < n_workers ; i ++ ) { while ( per_worker_completions [ i ] == false ) continue ; if ( per_worker_ci [ i ] != nullptr ) num_successful ++ ; } if ( num_successful != n_workers ) { SPDLOG_ERROR ( " " ) ; notifyAllProcsOnCheckpointComplete ( mng , workers , n_workers , this_worker_idx , per_worker_ci , false ) ; return ; } n_checkpoint_requests_issued = 0 ; n_checkpoint_requests_completed = 0 ; n_checkpoint_requests_failed = 0 ; for ( size_t i = 0 ; i < n_workers ; i ++ ) { CheckpointInput * ci = per_worker_ci [ i ] ; for ( auto it = ci -> inodesToCheckpoint . cbegin ( ) ; it != ci -> inodesToCheckpoint . cend ( ) ; ) { uint32_t disk_sector_no = FsImpl :: ino2SectorNo ( it -> first ) ; int rc = spdk_nvme_ns_cmd_write ( dev_ns , dev_qpair , it -> second , disk_sector_no , 1 , checkpointWriteComplete , this , 0 ) ; if ( rc == 0 ) { n_checkpoint_requests_issued ++ ; it ++ ; continue ; } if ( rc != - ENOMEM ) { SPDLOG_ERROR ( " " , rc ) ; it ++ ; continue ; } waitForCheckpointWriteCompletions ( ) ; } std :: unordered_map < uint64_t , void * > bitmap_block_to_data_arr [ 2 ] ; bitmap_block_to_data_arr [ 0 ] = ci -> bitmapsToCheckpoint ; bitmap_block_to_data_arr [ 1 ] = ci -> inodeBitmapsToCheckpoint ; for ( auto & bitmap_block_to_data : bitmap_block_to_data_arr ) { for ( auto it = bitmap_block_to_data . cbegin ( ) ; it != bitmap_block_to_data . cend ( ) ; ) { int rc = spdk_nvme_ns_cmd_write ( dev_ns , dev_qpair , it -> second , it -> first * 8 , 8 , checkpointWriteComplete , this , 0 ) ; if ( rc == 0 ) { n_checkpoint_requests_issued ++ ; it ++ ; continue ; } if ( rc != - ENOMEM ) { SPDLOG_ERROR ( " " , rc ) ; it ++ ; continue ; } waitForCheckpointWriteCompletions ( ) ; } } } waitForCheckpointWriteCompletions ( ) ; if ( n_checkpoint_requests_failed > 0 ) { SPDLOG_ERROR ( " " ) ; notifyAllProcsOnCheckpointComplete ( mng , workers , n_workers , this_worker_idx , per_worker_ci , false ) ; return ; } SPDLOG_INFO ( " " ) ; notifyAllProcsOnCheckpointComplete ( mng , workers , n_workers , this_worker_idx , per_worker_ci , true ) ; }
blocking function call that checkpoints all journals.
blocking function call that checkpoints all journals .
// TODO possibly just use one struct instead of these three for this code. It // could serve as the ctx as well // skip this thread // TODO: make async version where they all send messages back to this worker. // For each worker, we first write out the inodes and then the bitmaps. // increment on success // TODO check rc to see if queue is full. Wait for completions before // sending any more // TODO : maybe fail checkpointing entirely? // rc is -ENOMEM, need to wait till a few have finished and then resume // NOTE: not incrementing it, trying again for the same failed inode // increment on success // TODO (jing) instead of 8, would BSIZE / SSD_SEC_SIZE be correct for // lba_count? // TODO : maybe fail checkpointing entirely? // NOTE: not incrementing it, trying again for same failed bitmap // TODO all blocking waits for io completions can be made asynchronous, where // the worker still services requests and on process completions, we come back // into the checkpointer // Checkpoint succeeded, notify of success