Skip to content

Commit

Permalink
log store sync flush
Browse files Browse the repository at this point in the history
  • Loading branch information
@JacksonYao287
JacksonYao287 committed Jul 30, 2024
1 parent 8ee9a2b commit f064bc6
Show file tree
Hide file tree
Showing 14 changed files with 603 additions and 1,295 deletions.
1 change: 0 additions & 1 deletion conanfile.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@
class HomestoreConan(ConanFile):
name = "homestore"
version = "6.4.32"

homepage = "https://github.com/eBay/Homestore"
description = "HomeStore Storage Engine"
topics = ("ebay", "nublox")
Expand Down
154 changes: 25 additions & 129 deletions src/include/homestore/logstore/log_store.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
#include <set>
#include <unordered_map>
#include <vector>
#include <tuple>

#include <sisl/fds/buffer.hpp>
#include <sisl/fds/stream_tracker.hpp>
Expand Down Expand Up @@ -81,18 +82,6 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
*/
log_replay_done_cb_t get_log_replay_done_cb() const { return m_replay_done_cb; }

/**
* @brief Write the blob at the user specified seq number in sync manner. Under the covers it will call async
* write and then wait for its completion. As such this is much lesser performing than async version since it
* involves mutex/cv combination
*
* @param seq_num : Sequence number to insert data
* @param b : Data blob to write to log
*
* @return is write completed successfully.
*/
bool write_sync(logstore_seq_num_t seq_num, const sisl::io_blob& b);

/**
* @brief Write the blob at the user specified seq number - prepared as a request in async fashion.
*
Expand All @@ -110,17 +99,7 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
* @param cookie : Any cookie or context which will passed back in the callback
* @param cb Callback upon completion which is called with the status, seq_num and cookie that was passed.
*/
void write_async(logstore_seq_num_t seq_num, const sisl::io_blob& b, void* cookie, const log_write_comp_cb_t& cb,
bool flush_wait = false);

/**
* @brief This method appends the blob into the log and it returns the generated seq number
*
* @param b Blob of data to append
* @return logstore_seq_num_t Returns the seqnum generated by the log
*/
// This method is not implemented yet
logstore_seq_num_t append_sync(const sisl::io_blob& b);
void write_async(logstore_seq_num_t seq_num, const sisl::io_blob& b, void* cookie, const log_write_comp_cb_t& cb);

/**
* @brief This method appends the blob into the log and makes a callback at the end of the append.
Expand All @@ -132,6 +111,14 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
*/
logstore_seq_num_t append_async(const sisl::io_blob& b, void* cookie, const log_write_comp_cb_t& completion_cb);

/**
* @brief Write the blob at the user specified seq number and flush, just like write_sync
*
* @param seq_num: Seq number to write to
* @param b : Blob of data
*/
void write_and_flush(logstore_seq_num_t seq_num, const sisl::io_blob& b);

/**
* @brief Read the log provided the sequence number synchronously. This is not the most efficient way to read
* as reader will be blocked until read is completed. In addition, it is built on-top of async system by doing
Expand All @@ -144,25 +131,6 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
*/
log_buffer read_sync(logstore_seq_num_t seq_num);

/**
* @brief Read the log based on the logstore_req prepared. In case callback is supplied, it uses the callback
* to provide the data it has read. If not overridden, use default callback registered during initialization.
*
* @param req Request containing seq_num
* @param cb [OPTIONAL] Callback to get the data back, if it needs to be different from the default registered
* one.
*/
void read_async(logstore_req* req, const log_found_cb_t& cb = nullptr);

/**
* @brief Read the log for the seq_num and make the callback with the data
*
* @param seq_num Seqnumber to read the log from
* @param cookie Any cookie or context which will passed back in the callback
* @param cb Callback which contains seq_num, cookie and
*/
void read_async(logstore_seq_num_t seq_num, void* cookie, const log_found_cb_t& cb);

/**
* @brief Truncate the logs for this log store upto the seq_num provided (inclusive). Once truncated, the reads
* on seq_num <= upto_seq_num will return an error. The truncation in general is a 2 step process, where first
Expand All @@ -175,7 +143,6 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
* expensive and grouping them together yields better results.
*
* Note: this flag currently is not used, meaning all truncate is in memory only;
* @return number of records to truncate
*/
void truncate(logstore_seq_num_t upto_seq_num, bool in_memory_truncate_only = true);

Expand All @@ -188,30 +155,17 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
*/
void fill_gap(logstore_seq_num_t seq_num);

/**
* @brief Get the safe truncation log dev key from this log store perspective. Please note that the safe idx is
* not globally safe, but it is safe from this log store perspective only. To get global safe id, one should
* access all log stores and get the minimum of them before truncating.
*
* It could return invalid logdev_key which indicates that this log store does not have any valid logdev key
* to truncate. This could happen when there were no ios on this logstore since last truncation or at least no
* ios are flushed yet. The caller should simply ignore this return value.
*
* @return truncation_entry_t: Which contains the logdev key and its corresponding seq_num to truncate and also
* is that entry represents the entire log store.
*/
// truncation_entry_t get_safe_truncation_boundary() const;

/**
* @brief Get the last truncated seqnum upto which we have truncated. If called after recovery, it returns the
* first seq_num it has seen-1.
*
* @return logstore_seq_num_t
* @return the last truncated seqnum upto which we have truncated
*/
logstore_seq_num_t truncated_upto() const {
const auto ts{m_safe_truncation_boundary.seq_num.load(std::memory_order_acquire)};
return (ts == std::numeric_limits< logstore_seq_num_t >::max()) ? -1 : ts;
}
logstore_seq_num_t truncated_upto() const { return m_start_lsn.load(std::memory_order_acquire) - 1; }

logdev_key get_trunc_ld_key() const { return m_trunc_ld_key; }

std::tuple< logstore_seq_num_t, logdev_key, logstore_seq_num_t > truncate_info() const;

sisl::StreamTracker< logstore_record >& log_records() { return m_records; }

Expand Down Expand Up @@ -256,42 +210,23 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
*
* @param seq_num Sequence number upto which logs are to be flushed. If not provided, will wait to flush all seq
* numbers issued prior.
* @return True on success
*/
void flush_sync(logstore_seq_num_t upto_seq_num = invalid_lsn());
void flush(logstore_seq_num_t upto_seq_num = invalid_lsn());

/**
* @brief Rollback the given instance to the given sequence number
*
* @param seq_num Sequence number back which logs are to be rollbacked
* @param to_lsn Sequence number back which logs are to be rollbacked
* @return True on success
*/
uint64_t rollback_async(logstore_seq_num_t to_lsn, on_rollback_cb_t cb);

auto seq_num() const { return m_seq_num.load(std::memory_order_acquire); }
bool rollback(logstore_seq_num_t to_lsn);

std::shared_ptr< LogDev > get_logdev() { return m_logdev; }
auto start_lsn() const { return m_start_lsn.load(std::memory_order_acquire); }

nlohmann::json dump_log_store(const log_dump_req& dump_req = log_dump_req());

nlohmann::json get_status(int verbosity) const;

/**
* Retrieves the truncation information before device truncation.
*
* @return A constant reference to the truncation_info object representing the truncation information.
*/
const truncation_info& pre_device_truncation();

/**
* \brief post device truncation processing.
*
* This function is used to update safe truncation boundary to the specified `trunc_upto_key`.
*
* \param trunc_upto_key The key indicating the log entry up to which truncation has been performed.
*/
void post_device_truncation(const logdev_key& trunc_upto_key);

/**
* Handles the completion of a write operation in the log store.
*
Expand All @@ -300,17 +235,6 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
*/
void on_write_completion(logstore_req* req, const logdev_key& ld_key);

/**
* \brief Handles the completion of a read operation in the log store.
*
* This function is called when a read operation in the log store has completed.
* It takes a pointer to a logstore_req object and a logdev_key object as parameters.
*
* \param req The pointer to the logstore_req object representing the read request.
* \param ld_key The logdev_key object representing the key used for the read operation.
*/
void on_read_completion(logstore_req* req, const logdev_key& ld_key);

/**
* @brief Handles the event when a log is found.
*
Expand All @@ -327,27 +251,8 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
*/
void on_log_found(logstore_seq_num_t seq_num, const logdev_key& ld_key, const logdev_key& flush_ld_key,
log_buffer buf);
/**
* @brief Handles the completion of a batch flush operation to update internal state.
*
* This function is called when a batch flush operation is completed.
* It takes a `logdev_key` parameter that represents the key of the flushed batch.
*
* This function is also called during log store recovery;
*
* @param flush_batch_ld_key The key of the flushed batch.
*/
void on_batch_completion(const logdev_key& flush_batch_ld_key);

private:
/**
* Truncates the log store up to the specified sequence number.
*
* @param upto_seq_num The sequence number up to which the log store should be truncated.
*/
void do_truncate(logstore_seq_num_t upto_seq_num);

int search_max_le(logstore_seq_num_t input_sn);
std::shared_ptr< LogDev > get_logdev() { return m_logdev; }

private:
logstore_id_t m_store_id;
Expand All @@ -357,21 +262,12 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
log_req_comp_cb_t m_comp_cb;
log_found_cb_t m_found_cb;
log_replay_done_cb_t m_replay_done_cb;
std::atomic< logstore_seq_num_t > m_seq_num;
std::atomic< logstore_seq_num_t > m_start_lsn;
std::atomic< logstore_seq_num_t > m_next_lsn;
std::atomic< logstore_seq_num_t > m_tail_lsn;
std::string m_fq_name;
LogStoreServiceMetrics& m_metrics;

// seq_ld_key_pair m_flush_batch_max = {-1, {0, 0}}; // The maximum seqnum we have seen in the prev flushed
// batch
logstore_seq_num_t m_flush_batch_max_lsn{std::numeric_limits< logstore_seq_num_t >::min()};

// Sync flush sections
std::atomic< logstore_seq_num_t > m_sync_flush_waiter_lsn{invalid_lsn()};
std::mutex m_sync_flush_mtx;
std::mutex m_single_sync_flush_mtx;
std::condition_variable m_sync_flush_cv;

std::vector< seq_ld_key_pair > m_truncation_barriers; // List of truncation barriers
truncation_info m_safe_truncation_boundary;
logdev_key m_trunc_ld_key;
};
} // namespace homestore
40 changes: 9 additions & 31 deletions src/include/homestore/logstore/log_store_internal.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@
#include <nlohmann/json.hpp>

namespace homestore {

///////////////////// All typedefs ///////////////////////////////
class logstore_req;
class HomeLogStore;
Expand All @@ -50,16 +49,14 @@ typedef std::function< void(logstore_seq_num_t, sisl::io_blob&, logdev_key, void
typedef std::function< void(logstore_seq_num_t, log_buffer, void*) > log_found_cb_t;
typedef std::function< void(std::shared_ptr< HomeLogStore >) > log_store_opened_cb_t;
typedef std::function< void(std::shared_ptr< HomeLogStore >, logstore_seq_num_t) > log_replay_done_cb_t;
typedef std::function< void(const std::unordered_map< logdev_id_t, logdev_key >&) > device_truncate_cb_t;

typedef int64_t logid_t;

struct logdev_key {
logid_t idx;
off_t dev_offset;

constexpr logdev_key(const logid_t idx = std::numeric_limits< logid_t >::min(),
const off_t dev_offset = std::numeric_limits< uint64_t >::min()) :
constexpr logdev_key(const logid_t idx = -1, const off_t dev_offset = std::numeric_limits< uint64_t >::min()) :
idx{idx}, dev_offset{dev_offset} {}
logdev_key(const logdev_key&) = default;
logdev_key& operator=(const logdev_key&) = default;
Expand All @@ -72,11 +69,11 @@ struct logdev_key {
operator bool() const { return is_valid(); }
bool is_valid() const { return !is_lowest() && !is_highest(); }

bool is_lowest() const { return (idx == std::numeric_limits< logid_t >::min()); }
bool is_lowest() const { return (idx == -1); }
bool is_highest() const { return (idx == std::numeric_limits< logid_t >::max()); }

void set_lowest() {
idx = std::numeric_limits< logid_t >::min();
idx = -1;
dev_offset = std::numeric_limits< uint64_t >::min();
}

Expand Down Expand Up @@ -109,9 +106,12 @@ struct log_dump_req {

struct logstore_record {
logdev_key m_dev_key;
logdev_key m_trunc_key;
// only when out-of-order write happens and the higher lsn is flushed in a lower LogGroup while the lower lsn is
// flushed in a higher LogGroup , they are different

logstore_record() = default;
logstore_record(const logdev_key& key) : m_dev_key{key} {}
logstore_record(const logdev_key& key, const logdev_key& trunc_key) : m_dev_key{key}, m_trunc_key{trunc_key} {}
};

class HomeLogStore;
Expand All @@ -121,7 +121,6 @@ struct logstore_req {
logstore_seq_num_t seq_num; // Log store specific seq_num (which could be monotonically increaseing with logstore)
sisl::io_blob data; // Data blob containing data
void* cookie; // User generated cookie (considered as opaque)
bool is_write; // Directon of IO
bool is_internal_req; // If the req is created internally by HomeLogStore itself
log_req_comp_cb_t cb; // Callback upon completion of write (overridden than default)
Clock::time_point start_time;
Expand All @@ -138,13 +137,11 @@ struct logstore_req {
// TODO: Implement this method
return 0;
}
static logstore_req* make(HomeLogStore* store, logstore_seq_num_t seq_num, const sisl::io_blob& data,
bool is_write_req = true) {
static logstore_req* make(HomeLogStore* store, logstore_seq_num_t seq_num, const sisl::io_blob& data) {
logstore_req* req = new logstore_req();
req->log_store = store;
req->seq_num = seq_num;
req->data = data;
req->is_write = is_write_req;
req->is_internal_req = true;
req->cb = nullptr;

Expand All @@ -158,25 +155,6 @@ struct logstore_req {
logstore_req() = default;
};

struct seq_ld_key_pair {
logstore_seq_num_t seq_num{-1};
logdev_key ld_key;
};

struct truncation_info {
// Safe log dev location upto which it is truncatable
logdev_key ld_key{std::numeric_limits< logid_t >::min(), 0};

// LSN of this log store upto which it is truncated
std::atomic< logstore_seq_num_t > seq_num{-1};

// Is there any entry which is already store truncated but waiting for device truncation
bool pending_dev_truncation{false};

// Any truncation entries/barriers which are not part of this truncation
bool active_writes_not_part_of_truncation{false};
};

#pragma pack(1)
struct logstore_superblk {
logstore_superblk(const logstore_seq_num_t seq_num = 0) : m_first_seq_num{seq_num} {}
Expand All @@ -194,4 +172,4 @@ struct logstore_superblk {
logstore_seq_num_t m_first_seq_num{0};
};
#pragma pack()
} // namespace homestore
} // namespace homestore
Loading

0 comments on commit f064bc6

Please sign in to comment.