src/server/pegasus_server_impl.cpp
// not thread safe, should be protected by caller
not thread safe, should be protected by caller
not thread safe should be protected by caller
::dsn::error_code pegasus_server_impl::copy_checkpoint_to_dir_unsafe(const char *checkpoint_dir,
                                                                     int64_t *checkpoint_decree,
                                                                     bool flush_memtable)
{
    rocksdb::Checkpoint *chkpt_raw = nullptr;
    auto status = rocksdb::Checkpoint::Create(_db, &chkpt_raw);
    if (!status.ok()) {
        derror_replica("create Checkpoint object failed, error = {}", status.ToString());
        return ::dsn::ERR_LOCAL_APP_FAILURE;
    }
    std::unique_ptr<rocksdb::Checkpoint> chkpt(chkpt_raw);
    if (::dsn::utils::filesystem::directory_exists(checkpoint_dir)) {
        ddebug_replica("checkpoint directory {} is already existed, remove it first",
                       checkpoint_dir);
        if (!::dsn::utils::filesystem::remove_path(checkpoint_dir)) {
            derror_replica("remove checkpoint directory {} failed", checkpoint_dir);
            return ::dsn::ERR_FILE_OPERATION_FAILED;
        }
    }
    status = chkpt->CreateCheckpoint(checkpoint_dir,
                                     flush_memtable ? 0 : std::numeric_limits<uint64_t>::max());
    if (!status.ok()) {
        derror_replica("CreateCheckpoint failed, error = {}", status.ToString());
        if (!::dsn::utils::filesystem::remove_path(checkpoint_dir)) {
            derror_replica("remove checkpoint directory {} failed", checkpoint_dir);
        }
        return ::dsn::ERR_LOCAL_APP_FAILURE;
    }
    ddebug_replica("copy checkpoint to dir({}) succeed", checkpoint_dir);
    if (checkpoint_decree != nullptr) {
        rocksdb::DB *snapshot_db = nullptr;
        std::vector<rocksdb::ColumnFamilyHandle *> handles_opened;
        auto cleanup = [&](bool remove_checkpoint) {
            if (remove_checkpoint && !::dsn::utils::filesystem::remove_path(checkpoint_dir)) {
                derror_replica("remove checkpoint directory {} failed", checkpoint_dir);
            }
            if (snapshot_db) {
                for (auto handle : handles_opened) {
                    if (handle) {
                        snapshot_db->DestroyColumnFamilyHandle(handle);
                        handle = nullptr;
                    }
                }
                delete snapshot_db;
                snapshot_db = nullptr;
            }
        };
        std::vector<rocksdb::ColumnFamilyDescriptor> column_families(
            {{DATA_COLUMN_FAMILY_NAME, rocksdb::ColumnFamilyOptions()},
             {META_COLUMN_FAMILY_NAME, rocksdb::ColumnFamilyOptions()}});
        status = rocksdb::DB::OpenForReadOnly(
            rocksdb::DBOptions(), checkpoint_dir, column_families, &handles_opened, &snapshot_db);
        if (!status.ok()) {
            derror_replica(
                "OpenForReadOnly from {} failed, error = {}", checkpoint_dir, status.ToString());
            snapshot_db = nullptr;
            cleanup(true);
            return ::dsn::ERR_LOCAL_APP_FAILURE;
        }
        dcheck_eq_replica(handles_opened.size(), 2);
        dcheck_eq_replica(handles_opened[1]->GetName(), META_COLUMN_FAMILY_NAME);
        uint64_t last_flushed_decree =
            _meta_store->get_decree_from_readonly_db(snapshot_db, handles_opened[1]);
        *checkpoint_decree = last_flushed_decree;
        cleanup(false);
    }
    return ::dsn::ERR_OK;
}
:: dsn :: error_code pegasus_server_impl :: copy_checkpoint_to_dir_unsafe ( const char * checkpoint_dir , int64_t * checkpoint_decree , bool flush_memtable ) { rocksdb :: Checkpoint * chkpt_raw = nullptr ; auto status = rocksdb :: Checkpoint :: Create ( _db , & chkpt_raw ) ; if ( ! status . ok ( ) ) { derror_replica ( " " , status . ToString ( ) ) ; return :: dsn :: ERR_LOCAL_APP_FAILURE ; } std :: unique_ptr < rocksdb :: Checkpoint > chkpt ( chkpt_raw ) ; if ( :: dsn :: utils :: filesystem :: directory_exists ( checkpoint_dir ) ) { ddebug_replica ( " " , checkpoint_dir ) ; if ( ! :: dsn :: utils :: filesystem :: remove_path ( checkpoint_dir ) ) { derror_replica ( " " , checkpoint_dir ) ; return :: dsn :: ERR_FILE_OPERATION_FAILED ; } } status = chkpt -> CreateCheckpoint ( checkpoint_dir , flush_memtable ? 0 : std :: numeric_limits < uint64_t > :: max ( ) ) ; if ( ! status . ok ( ) ) { derror_replica ( " " , status . ToString ( ) ) ; if ( ! :: dsn :: utils :: filesystem :: remove_path ( checkpoint_dir ) ) { derror_replica ( " " , checkpoint_dir ) ; } return :: dsn :: ERR_LOCAL_APP_FAILURE ; } ddebug_replica ( " " , checkpoint_dir ) ; if ( checkpoint_decree != nullptr ) { rocksdb :: DB * snapshot_db = nullptr ; std :: vector < rocksdb :: ColumnFamilyHandle * > handles_opened ; auto cleanup = [ & ] ( bool remove_checkpoint ) { if ( remove_checkpoint && ! :: dsn :: utils :: filesystem :: remove_path ( checkpoint_dir ) ) { derror_replica ( " " , checkpoint_dir ) ; } if ( snapshot_db ) { for ( auto handle : handles_opened ) { if ( handle ) { snapshot_db -> DestroyColumnFamilyHandle ( handle ) ; handle = nullptr ; } } delete snapshot_db ; snapshot_db = nullptr ; } } ; std :: vector < rocksdb :: ColumnFamilyDescriptor > column_families ( { { DATA_COLUMN_FAMILY_NAME , rocksdb :: ColumnFamilyOptions ( ) } , { META_COLUMN_FAMILY_NAME , rocksdb :: ColumnFamilyOptions ( ) } } ) ; status = rocksdb :: DB :: OpenForReadOnly ( rocksdb :: DBOptions ( ) , checkpoint_dir , column_families , & handles_opened , & snapshot_db ) ; if ( ! status . ok ( ) ) { derror_replica ( " " , checkpoint_dir , status . ToString ( ) ) ; snapshot_db = nullptr ; cleanup ( true ) ; return :: dsn :: ERR_LOCAL_APP_FAILURE ; } dcheck_eq_replica ( handles_opened . size ( ) , 2 ) ; dcheck_eq_replica ( handles_opened [ 1 ] -> GetName ( ) , META_COLUMN_FAMILY_NAME ) ; uint64_t last_flushed_decree = _meta_store -> get_decree_from_readonly_db ( snapshot_db , handles_opened [ 1 ] ) ; * checkpoint_decree = last_flushed_decree ; cleanup ( false ) ; } return :: dsn :: ERR_OK ; }
not thread safe, should be protected by caller
not thread safe should be protected by caller
// CreateCheckpoint() will not flush memtable when log_size_for_flush = max // Because of RocksDB's restriction, we have to to open default column family even though // not use it