<commit_before>/// The default initial capacity of the sector map.
const INITIAL_CAPACITY: usize = 256;

/// A cached disk.
///
/// This wrapper manages caching and the consistency issues originating from it.
struct Cache {
    /// The inner driver.
    driver: vdev::Driver,

    /// The cache replacement tracker.
    ///
    /// This tracks the state of the replacement algorithm, which chooses which cache block shall
    /// be replaced in favor of a new cache. It serves to estimate/guess which block is likely not
    /// used in the near future.
    tracker: mlcr::ConcurrentCache,
    /// The sector-number-to-data block map.
    sectors: AtomicHashMap<disk::Sector, disk::SectorBuf>,
}

impl From<vdev::Driver> for Cache {
    fn from(driver: vdev::Driver) -> Cache {
        Cache {
            // Store the driver.
            driver: driver,
            // Set empty/default values.
            tracker: mlcr::ConcurrentCache::new(),
            sectors: CHashMap::with_capacity(INITIAL_CAPACITY),
        }
    }
}

impl Cache {
    /// Write a sector.
    ///
    /// This writes `buf` into sector `sector`. If it fails, the error is returned.
    fn write(&self, sector: disk::Sector, buf: Box<disk::SectorBuf>) -> impl Future<(), disk::Error> {
        debug!(self, "writing sector"; "sector" => sector);

        // Then insert it into the cache.
        self.sectors.insert(sector, buf);
        // Write the data to the disk.
        self.driver.write(sector, &buf)
    }

    /// Drop a sector from the cache.
    fn forget(&self, sector: disk::Sector) {
        debug!(self, "removing sector from cache"; "sector" => sector);

        // Update the cache tracker.
        self.tracker.remove(sector);
        // Update the sector map.
        self.sectors.remove(sector);
    }

    /// Read a sector.
    ///
    /// This reads sector `sector`, and applies the closure `map`. If `sector` needs to be fetched
    /// from the disk, and `map` fails, data recovery is attempted.
    ///
    /// If an I/O operation fails, the error is returned. Otherwise, the return value of `map` is
    /// returned.
    fn read_then<F, T, E>(&self, sector: disk::Sector, map: F) -> impl Future<T, E>
        where F: Fn(atomic_hash_map::Value<disk::SectorBuf>) -> Result<T, E>,
              E: From<disk::Error> {
        debug!(self, "reading sector"; "sector" => sector);

        // Check if the sector is already available in the cache.
        if let Some(accessor) = self.sectors.get(sector) {
            // Yup, we found the sector in the cache.
            trace!(self, "cache hit; reading from cache"; "sector" => sector);

            // Touch the sector.
            self.tracker.touch(sector);

            map(accessor)
        } else {
            trace!(self, "cache miss; reading from disk"; "sector" => sector);

            // Insert the sector into the cache tracker.
            self.tracker.touch(sector):

            // Fetch the data from the disk.
            self.disk.read(sector).map(|buf| {
                // Insert the read data into the hash table.
                self.sectors.get_mut_or(sector, buf)
            }).and_then(map)
            // TODO: If the above failed, try to recover the data through the vdev redundancy.
        }
    }

    /// Trim the cache.
    ///
    /// This reduces the cache to exactly `to` blocks.
    fn trim(&self, to: usize) {
        info!(self, "trimming cache"; "to" => to);

        // Lock the cache tracker.<commit_msg>Remove trace of striped hash map.<commit_after>/// The default initial capacity of the sector map.
const INITIAL_CAPACITY: usize = 256;

/// A cached disk.
///
/// This wrapper manages caching and the consistency issues originating from it.
struct Cache {
    /// The inner driver.
    driver: vdev::Driver,

    /// The cache replacement tracker.
    ///
    /// This tracks the state of the replacement algorithm, which chooses which cache block shall
    /// be replaced in favor of a new cache. It serves to estimate/guess which block is likely not
    /// used in the near future.
    tracker: mlcr::ConcurrentCache,
    /// The sector-number-to-data block map.
    sectors: AtomicHashMap<disk::Sector, disk::SectorBuf>,
}

impl From<vdev::Driver> for Cache {
    fn from(driver: vdev::Driver) -> Cache {
        Cache {
            // Store the driver.
            driver: driver,
            // Set empty/default values.
            tracker: mlcr::ConcurrentCache::new(),
            sectors: CHashMap::with_capacity(INITIAL_CAPACITY),
        }
    }
}

impl Cache {
    /// Write a sector.
    ///
    /// This writes `buf` into sector `sector`. If it fails, the error is returned.
    fn write(&self, sector: disk::Sector, buf: Box<disk::SectorBuf>) -> impl Future<(), disk::Error> {
        debug!(self, "writing sector"; "sector" => sector);

        // Then insert it into the cache.
        self.sectors.insert(sector, buf);
        // Write the data to the disk.
        self.driver.write(sector, &buf)
    }

    /// Drop a sector from the cache.
    fn forget(&self, sector: disk::Sector) {
        debug!(self, "removing sector from cache"; "sector" => sector);

        // Update the cache tracker.
        self.tracker.remove(sector);
        // Update the sector map.
        self.sectors.remove(sector);
    }

    /// Read a sector.
    ///
    /// This reads sector `sector`, and applies the closure `map`. If `sector` needs to be fetched
    /// from the disk, and `map` fails, data recovery is attempted.
    ///
    /// If an I/O operation fails, the error is returned. Otherwise, the return value of `map` is
    /// returned.
    fn read_then<F, T, E>(&self, sector: disk::Sector, map: F) -> impl Future<T, E>
        where F: Fn(atomic_hash_map::Value<disk::SectorBuf>) -> Result<T, E>,
              E: From<disk::Error> {
        debug!(self, "reading sector"; "sector" => sector);

        // Check if the sector is already available in the cache.
        if let Some(buf) = self.sectors.get(sector) {
            // Yup, we found the sector in the cache.
            trace!(self, "cache hit; reading from cache"; "sector" => sector);

            // Touch the sector.
            self.tracker.touch(sector);

            map(buf)
        } else {
            trace!(self, "cache miss; reading from disk"; "sector" => sector);

            // Insert the sector into the cache tracker.
            self.tracker.touch(sector):

            // Fetch the data from the disk.
            self.disk.read(sector).map(|buf| {
                // Insert the read data into the hash table.
                self.sectors.get_mut_or(sector, buf)
            }).and_then(map)
            // TODO: If the above failed, try to recover the data through the vdev redundancy.
        }
    }

    /// Trim the cache.
    ///
    /// This reduces the cache to exactly `to` blocks.
    fn trim(&self, to: usize) {
        info!(self, "trimming cache"; "to" => to);

        // Lock the cache tracker.
{'fraction_non_alphanumeric': 0.12678265830005706, 'fraction_numerical': 0.0008556759840273817, 'mean_word_length': 2.045158488927486, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '311245', 'n_tokens_mistral': 2034, 'n_tokens_neox': 1953, 'n_words': 714}