<commit_before>// Copyright (c) The Libra Core Contributors
// SPDX-License-Identifier: Apache-2.0

//! This module provides `Pruner` which manages a thread pruning old data in the background and is
//! meant to be triggered by other threads as they commit new data to the DB.

use crate::{
    schema::{
        jellyfish_merkle_node::JellyfishMerkleNodeSchema, stale_node_index::StaleNodeIndexSchema,
    },
    OP_COUNTER,
};
use failure::prelude::*;
use logger::prelude::*;
use schemadb::{ReadOptions, SchemaBatch, DB};
use std::{
    sync::{
        mpsc::{channel, Receiver, Sender},
        Arc, Mutex,
    },
    thread::JoinHandle,
};
use types::transaction::Version;

use failure::_core::sync::atomic::Ordering;
use std::sync::atomic::AtomicU64;
#[cfg(test)]
use std::thread::sleep;
#[cfg(test)]
use std::time::{Duration, Instant};

/// The `Pruner` is meant to be part of a `LibraDB` instance and runs in the background to prune old
/// data.
///
/// It creates a worker thread on construction and joins it on destruction. When destructed, it
/// quits the worker thread eagerly without waiting for all pending work to be done.
pub(crate) struct Pruner {
    /// Other than the latest version, how many historical versions to keep being readable. For
    /// example, this being 0 means keep only the latest version.
    num_historical_versions_to_keep: u64,
    /// The worker thread handle, created upon Pruner instance construction and joined upon its
    /// destruction. It only becomes `None` after joined in `drop()`.
    worker_thread: Option<JoinHandle<()>>,
    /// The sender side of the channel talking to the worker thread.
    command_sender: Mutex<Sender<Command>>,
    /// (For tests) A way for the worker thread to inform the `Pruner` the pruning progress. If it
    /// sets this atomic value to `V`, all versions before `V` can no longer be accessed.
    #[allow(dead_code)]
    worker_progress: Arc<AtomicU64>,
}

impl Pruner {
    /// Creates a worker thread that waits on a channel for pruning commands.
    pub fn new(db: Arc<DB>, num_historical_versions_to_keep: u64) -> Self {
        let (command_sender, command_receiver) = channel();
        let worker_progress = Arc::new(AtomicU64::new(0));
        let worker_progress_clone = Arc::clone(&worker_progress);

        let worker_thread = std::thread::Builder::new()
            .name("libradb_pruner".into())
            .spawn(move || Worker::new(db, command_receiver, worker_progress_clone).work_loop())
            .expect("Creating pruner thread should succeed.");

        Self {
            num_historical_versions_to_keep,
            worker_thread: Some(worker_thread),
            command_sender: Mutex::new(command_sender),
            worker_progress,
        }
    }

    /// Sends pruning command to the worker thread when necessary.
    pub fn wake(&self, latest_version: Version) {
        if latest_version > self.num_historical_versions_to_keep {
            let least_readable_version = latest_version - self.num_historical_versions_to_keep;
            self.command_sender
                .lock()
                .expect("command_sender to pruner thread should lock.")
                .send(Command::Prune {
                    least_readable_version,
                })
                .expect("Receiver should not destruct prematurely.");
        }
    }

    /// (For tests only.) Notifies the worker thread and waits for it to finish its job by polling
    /// an internal counter.
    #[cfg(test)]
    pub fn wake_and_wait(&self, latest_version: Version) -> Result<()> {
        self.wake(latest_version);

        if latest_version > self.num_historical_versions_to_keep {
            let least_readable_version = latest_version - self.num_historical_versions_to_keep;
            const TIMEOUT: Duration = Duration::from_millis(100);<commit_msg>[storage] trivial: enlarges timeout in `Pruner::wake_and_wait()` (tests only)<commit_after>// Copyright (c) The Libra Core Contributors
// SPDX-License-Identifier: Apache-2.0

//! This module provides `Pruner` which manages a thread pruning old data in the background and is
//! meant to be triggered by other threads as they commit new data to the DB.

use crate::{
    schema::{
        jellyfish_merkle_node::JellyfishMerkleNodeSchema, stale_node_index::StaleNodeIndexSchema,
    },
    OP_COUNTER,
};
use failure::prelude::*;
use logger::prelude::*;
use schemadb::{ReadOptions, SchemaBatch, DB};
use std::{
    sync::{
        mpsc::{channel, Receiver, Sender},
        Arc, Mutex,
    },
    thread::JoinHandle,
};
use types::transaction::Version;

use failure::_core::sync::atomic::Ordering;
use std::sync::atomic::AtomicU64;
#[cfg(test)]
use std::thread::sleep;
#[cfg(test)]
use std::time::{Duration, Instant};

/// The `Pruner` is meant to be part of a `LibraDB` instance and runs in the background to prune old
/// data.
///
/// It creates a worker thread on construction and joins it on destruction. When destructed, it
/// quits the worker thread eagerly without waiting for all pending work to be done.
pub(crate) struct Pruner {
    /// Other than the latest version, how many historical versions to keep being readable. For
    /// example, this being 0 means keep only the latest version.
    num_historical_versions_to_keep: u64,
    /// The worker thread handle, created upon Pruner instance construction and joined upon its
    /// destruction. It only becomes `None` after joined in `drop()`.
    worker_thread: Option<JoinHandle<()>>,
    /// The sender side of the channel talking to the worker thread.
    command_sender: Mutex<Sender<Command>>,
    /// (For tests) A way for the worker thread to inform the `Pruner` the pruning progress. If it
    /// sets this atomic value to `V`, all versions before `V` can no longer be accessed.
    #[allow(dead_code)]
    worker_progress: Arc<AtomicU64>,
}

impl Pruner {
    /// Creates a worker thread that waits on a channel for pruning commands.
    pub fn new(db: Arc<DB>, num_historical_versions_to_keep: u64) -> Self {
        let (command_sender, command_receiver) = channel();
        let worker_progress = Arc::new(AtomicU64::new(0));
        let worker_progress_clone = Arc::clone(&worker_progress);

        let worker_thread = std::thread::Builder::new()
            .name("libradb_pruner".into())
            .spawn(move || Worker::new(db, command_receiver, worker_progress_clone).work_loop())
            .expect("Creating pruner thread should succeed.");

        Self {
            num_historical_versions_to_keep,
            worker_thread: Some(worker_thread),
            command_sender: Mutex::new(command_sender),
            worker_progress,
        }
    }

    /// Sends pruning command to the worker thread when necessary.
    pub fn wake(&self, latest_version: Version) {
        if latest_version > self.num_historical_versions_to_keep {
            let least_readable_version = latest_version - self.num_historical_versions_to_keep;
            self.command_sender
                .lock()
                .expect("command_sender to pruner thread should lock.")
                .send(Command::Prune {
                    least_readable_version,
                })
                .expect("Receiver should not destruct prematurely.");
        }
    }

    /// (For tests only.) Notifies the worker thread and waits for it to finish its job by polling
    /// an internal counter.
    #[cfg(test)]
    pub fn wake_and_wait(&self, latest_version: Version) -> Result<()> {
        self.wake(latest_version);

        if latest_version > self.num_historical_versions_to_keep {
            let least_readable_version = latest_version - self.num_historical_versions_to_keep;
            // Assuming no big pruning chunks will be issued by a test.
            const TIMEOUT: Duration = Duration::from_secs(10);
{'fraction_non_alphanumeric': 0.1046733732331593, 'fraction_numerical': 0.004202215713739972, 'mean_word_length': 3.025627883136853, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 25, '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': '5540968', 'n_tokens_mistral': 2270, 'n_tokens_neox': 2208, 'n_words': 721}