fault_tolerant_reader.cpp

Go to the documentation of this file.

#include "fault_tolerant_reader.hpp"
 
#include <algorithm>
#include <chrono>
#include <thread>
 
bool FaultTolerantReader::open(const std::string& path, std::string& error) {
    return reader_.open_readonly(path, error);
}
 
void FaultTolerantReader::close() {
    std::scoped_lock lock(mutex_);
    reader_.close();
    bad_map_ = {};
}
 
bool FaultTolerantReader::is_open() const {
    return reader_.is_open();
}
 
uint64_t FaultTolerantReader::size() const {
    return reader_.size();
}
 
uint32_t FaultTolerantReader::logical_sector_size() const {
    return reader_.logical_sector_size();
}
 
const std::string& FaultTolerantReader::path() const {
    return reader_.path();
}
 
void FaultTolerantReader::set_config(const FaultTolerantReaderConfig& cfg) {
    config_ = cfg;
}
 
void FaultTolerantReader::set_cancel_callback(CancelCallback cb) {
    cancel_cb_ = std::move(cb);
}
 
ReadResult FaultTolerantReader::read_at(uint64_t offset, void* buffer, size_t size) {
    std::scoped_lock lock(mutex_);
    ReadResult result;
    if (cancel_cb_ && cancel_cb_()) {
        result.error = "cancelled";
        return result;
    }
    result.success = reader_.read_at(offset, buffer, size);
    result.bytes_read = result.success ? size : 0;
    if (result.success) {
        bad_map_.mark_readable(offset, size);
    } else {
        result.error = "read failed";
    }
    return result;
}
 
ReadResult FaultTolerantReader::read_rescue(uint64_t offset, void* buffer, size_t size) {
    std::scoped_lock lock(mutex_);
    return read_rescue_recursive(offset, static_cast<unsigned char*>(buffer), size);
}
 
const BadBlockMap& FaultTolerantReader::bad_block_map() const {
    return bad_map_;
}
 
BadBlockMap& FaultTolerantReader::bad_block_map() {
    return bad_map_;
}
 
ReadResult FaultTolerantReader::read_rescue_recursive(uint64_t offset, unsigned char* buffer, size_t size) {
    ReadResult result;
    if (size == 0) {
        result.success = true;
        return result;
    }
    if (cancel_cb_ && cancel_cb_()) {
        result.error = "cancelled";
        return result;
    }
 
    const auto start = std::chrono::steady_clock::now();
    for (int attempt = 0; attempt < std::max(1, config_.max_retries); ++attempt) {
        if (reader_.read_at(offset, buffer, size)) {
            result.success = true;
            result.bytes_read = size;
            bad_map_.mark_readable(offset, size);
            const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
                std::chrono::steady_clock::now() - start).count();
            if (config_.mark_slow_ranges && elapsed > 250) {
                bad_map_.mark_slow(offset, size);
            }
            return result;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(25 * (attempt + 1)));
    }
 
    if (size <= config_.min_chunk_size) {
        bad_map_.mark_bad(offset, size);
        result.error = "unreadable range";
        return result;
    }
 
    const size_t left_size = size / 2;
    const size_t right_size = size - left_size;
    auto left = read_rescue_recursive(offset, buffer, left_size);
    auto right = read_rescue_recursive(offset + left_size, buffer + left_size, right_size);
 
    result.bytes_read = left.bytes_read + right.bytes_read;
    result.partial = result.bytes_read > 0 && result.bytes_read < size;
    result.success = (result.bytes_read == size);
    if (!result.success && result.partial) {
        result.error = "partial read";
    } else if (!result.success) {
        result.error = left.error.empty() ? right.error : left.error;
    }
    return result;
}