Remove logstore family. (#281)

Remove the fixed number of logstore families and instead
dynamically create logdev and logstores under them.
Logservice managers logdev's and logdev manages logstores.
Create a single journal vdev and multiple logdev use
the journal vdev with journal descriptor maintaing the state.
Remove LOG_REPLICATED and LOG_LOCAL and use LOG service.

conanfile.py

@@ -5,7 +5,7 @@
 
 class HomestoreConan(ConanFile):
     name = "homestore"
-    version = "5.1.2"
+    version = "5.1.3"
 
     homepage = "https://github.com/eBay/Homestore"
     description = "HomeStore Storage Engine"

src/include/homestore/homestore.hpp

@@ -56,8 +56,7 @@ class ReplApplication;
 
 using HomeStoreSafePtr = std::shared_ptr< HomeStore >;
 
-VENUM(hs_vdev_type_t, uint32_t, DATA_VDEV = 1, INDEX_VDEV = 2, META_VDEV = 3, DATA_LOGDEV_VDEV = 4,
-      CTRL_LOGDEV_VDEV = 5);
+VENUM(hs_vdev_type_t, uint32_t, DATA_VDEV = 1, INDEX_VDEV = 2, META_VDEV = 3, LOGDEV_VDEV = 4);
 
 #pragma pack(1)
 struct hs_vdev_context {
@@ -76,11 +75,10 @@ struct hs_stats {
 
 struct HS_SERVICE {
     static constexpr uint32_t META = 1 << 0;
-    static constexpr uint32_t LOG_REPLICATED = 1 << 1;
-    static constexpr uint32_t LOG_LOCAL = 1 << 2;
-    static constexpr uint32_t DATA = 1 << 3;
-    static constexpr uint32_t INDEX = 1 << 4;
-    static constexpr uint32_t REPLICATION = 1 << 5;
+    static constexpr uint32_t LOG = 1 << 1;
+    static constexpr uint32_t DATA = 1 << 2;
+    static constexpr uint32_t INDEX = 1 << 3;
+    static constexpr uint32_t REPLICATION = 1 << 4;
 
     uint32_t svcs;
 
@@ -91,8 +89,7 @@ struct HS_SERVICE {
         if (svcs & META) { str += "meta,"; }
         if (svcs & DATA) { str += "data,"; }
         if (svcs & INDEX) { str += "index,"; }
-        if (svcs & LOG_REPLICATED) { str += "log_replicated,"; }
-        if (svcs & LOG_LOCAL) { str += "log_local,"; }
+        if (svcs & LOG) { str += "log,"; }
         if (svcs & REPLICATION) { str += "replication,"; }
         return str;
     }

src/include/homestore/logstore/log_store.hpp

@@ -35,7 +35,6 @@
 
 namespace homestore {
 
-class LogStoreFamily;
 class LogDev;
 class LogStoreServiceMetrics;
 
@@ -44,7 +43,7 @@ typedef std::function< void(logstore_seq_num_t) > on_rollback_cb_t;
 
 class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
 public:
-    HomeLogStore(LogStoreFamily& family, logstore_id_t id, bool append_mode, logstore_seq_num_t start_lsn);
+    HomeLogStore(std::shared_ptr< LogDev > logdev, logstore_id_t id, bool append_mode, logstore_seq_num_t start_lsn);
     HomeLogStore(const HomeLogStore&) = delete;
     HomeLogStore(HomeLogStore&&) noexcept = delete;
     HomeLogStore& operator=(const HomeLogStore&) = delete;
@@ -266,7 +265,7 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
 
     auto seq_num() const { return m_seq_num.load(std::memory_order_acquire); }
 
-    LogStoreFamily& get_family() { return m_logstore_family; }
+    std::shared_ptr< LogDev > get_logdev() { return m_logdev; }
 
     nlohmann::json dump_log_store(const log_dump_req& dump_req = log_dump_req());
 
@@ -285,8 +284,7 @@ class HomeLogStore : public std::enable_shared_from_this< HomeLogStore > {
     int search_max_le(logstore_seq_num_t input_sn);
 
     logstore_id_t m_store_id;
-    LogStoreFamily& m_logstore_family;
-    LogDev& m_logdev;
+    std::shared_ptr< LogDev > m_logdev;
     sisl::StreamTracker< logstore_record > m_records;
     bool m_append_mode{false};
     log_req_comp_cb_t m_comp_cb;

src/include/homestore/logstore/log_store_internal.hpp

@@ -43,13 +43,14 @@ typedef int64_t logstore_seq_num_t;
 typedef std::function< void(logstore_req*, logdev_key) > log_req_comp_cb_t;
 typedef sisl::byte_view log_buffer;
 typedef uint32_t logstore_id_t;
-typedef uint8_t logstore_family_id_t;
+typedef uint32_t logdev_id_t;
 
 typedef std::function< void(logstore_req*, logdev_key) > log_req_comp_cb_t;
 typedef std::function< void(logstore_seq_num_t, sisl::io_blob&, logdev_key, void*) > log_write_comp_cb_t;
 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;
 

src/include/homestore/logstore_service.hpp

@@ -41,26 +41,20 @@ class LogStoreServiceMetrics : public sisl::MetricsGroup {
     LogStoreServiceMetrics& operator=(LogStoreServiceMetrics&&) noexcept = delete;
 };
 
-class LogStoreFamily;
 class HomeLogStore;
 class LogDev;
 struct logdev_key;
 class VirtualDev;
 class JournalVirtualDev;
 struct vdev_info;
 struct log_dump_req;
+struct logdev_superblk;
 
 class LogStoreService {
     friend class HomeLogStore;
-    friend class LogStoreFamily;
     friend class LogDev;
 
 public:
-    static constexpr logstore_family_id_t DATA_LOG_FAMILY_IDX{0};
-    static constexpr logstore_family_id_t CTRL_LOG_FAMILY_IDX{1};
-    static constexpr size_t num_log_families = CTRL_LOG_FAMILY_IDX + 1;
-    typedef std::function< void(const std::array< logdev_key, num_log_families >&) > device_truncate_cb_t;
-
     LogStoreService();
     LogStoreService(const LogStoreService&) = delete;
     LogStoreService(LogStoreService&&) noexcept = delete;
@@ -81,36 +75,53 @@ class LogStoreService {
      */
     void stop();
 
+    /**
+     * @brief Create a brand new log dev. A logdev manages a list of chunks and state about the log offsets.
+     * Internally each logdev has a journal descriptor which maintains the data start and tail offsets and list of
+     * chunks. Logdev can start with zero chunks and dynamically add chunks based on write request.
+     * @return Newly created log dev id.
+     */
+    logdev_id_t create_new_logdev();
+
+    /**
+     * @brief Open a log dev.
+     *
+     * @param logdev_id: Logdev ID
+     * @return Newly created log dev id.
+     */
+    void open_logdev(logdev_id_t logdev_id);
+
     /**
      * @brief Create a brand new log store (both in-memory and on device) and returns its instance. It also book
      * keeps the created log store and user can get this instance of log store by using logstore_id
      *
-     * @param family_id: Logstores can be created on different log_devs. As of now we only support data log_dev and
-     * ctrl log dev. The idx indicates which log device it is from. Its a mandatory parameter.
+     * @param logdev_id: Logstores can be created on different log_devs.
      * @param append_mode: If the log store have to be in append mode, user can call append_async and do not need to
      * maintain the log_idx. Else user is expected to keep track of the log idx. Default to false
      *
      * @return std::shared_ptr< HomeLogStore >
      */
-    shared< HomeLogStore > create_new_log_store(logstore_family_id_t family_id, bool append_mode = false);
+    std::shared_ptr< HomeLogStore > create_new_log_store(logdev_id_t logdev_id, bool append_mode = false);
 
     /**
      * @brief Open an existing log store and does a recovery. It then creates an instance of this logstore and
      * returns
-     *
+     * @param logdev_id: Logdev ID of the log store to close
      * @param store_id: Store ID of the log store to open
+     * @param append_mode: Append or not.
+     * @param on_open_cb: Callback to be called once log store is opened.
      * @return std::shared_ptr< HomeLogStore >
      */
-    folly::Future< shared< HomeLogStore > > open_log_store(logstore_family_id_t family_id, logstore_id_t store_id,
+    folly::Future< shared< HomeLogStore > > open_log_store(logdev_id_t logdev_id, logstore_id_t store_id,
                                                            bool append_mode);
 
     /**
      * @brief Close the log store instance and free-up the resources
-     *
+     * @param logdev_id: Logdev ID of the log store to close
      * @param store_id: Store ID of the log store to close
      * @return true on success
      */
-    bool close_log_store(const logstore_family_id_t family_id, const logstore_id_t store_id) {
+    bool close_log_store(logdev_id_t logdev_id, logstore_id_t store_id) {
         // TODO: Implement this method
         return true;
     }
@@ -121,7 +132,7 @@ class LogStoreService {
      *
      * @param store_id
      */
-    void remove_log_store(const logstore_family_id_t family_id, const logstore_id_t store_id);
+    void remove_log_store(logdev_id_t logdev_id, logstore_id_t store_id);
 
     /**
      * @brief Schedule a truncate all the log stores physically on the device.
@@ -131,38 +142,37 @@ class LogStoreService {
      * Default to false
      * @param dry_run: If the truncate is a real one or just dry run to simulate the truncation
      */
-    void device_truncate(const device_truncate_cb_t& cb = nullptr, const bool wait_till_done = false,
-                         const bool dry_run = false);
+    void device_truncate(const device_truncate_cb_t& cb = nullptr, bool wait_till_done = false, bool dry_run = false);
 
-    folly::Future< std::error_code > create_vdev(uint64_t size, logstore_family_id_t family, uint32_t chunk_size);
-    shared< VirtualDev > open_vdev(const vdev_info& vinfo, logstore_family_id_t family, bool load_existing);
-    shared< JournalVirtualDev > get_vdev(logstore_family_id_t family) const {
-        return (family == DATA_LOG_FAMILY_IDX) ? m_data_logdev_vdev : m_ctrl_logdev_vdev;
-    }
+    folly::Future< std::error_code > create_vdev(uint64_t size, uint32_t chunk_size);
+    std::shared_ptr< VirtualDev > open_vdev(const vdev_info& vinfo, bool load_existing);
+    std::shared_ptr< JournalVirtualDev > get_vdev() const { return m_logdev_vdev; }
+    std::vector< std::shared_ptr< LogDev > > get_all_logdevs();
+    std::shared_ptr< LogDev > get_logdev(logdev_id_t id);
 
     nlohmann::json dump_log_store(const log_dump_req& dum_req);
-    nlohmann::json get_status(const int verbosity) const;
+    nlohmann::json get_status(int verbosity) const;
 
     LogStoreServiceMetrics& metrics() { return m_metrics; }
-    LogStoreFamily* data_log_family() { return m_logstore_families[DATA_LOG_FAMILY_IDX].get(); }
-    LogStoreFamily* ctrl_log_family() { return m_logstore_families[CTRL_LOG_FAMILY_IDX].get(); }
-
-    LogDev& data_logdev();
-    LogDev& ctrl_logdev();
 
     uint32_t used_size() const;
     uint32_t total_size() const;
     iomgr::io_fiber_t flush_thread() { return m_flush_fiber; }
     iomgr::io_fiber_t truncate_thread() { return m_truncate_fiber; }
 
 private:
+    std::shared_ptr< LogDev > create_new_logdev_internal(logdev_id_t logdev_id);
+    void logdev_super_blk_found(const sisl::byte_view& buf, void* meta_cookie);
+    void rollback_super_blk_found(const sisl::byte_view& buf, void* meta_cookie);
     void start_threads();
     void flush_if_needed();
 
 private:
-    std::array< std::unique_ptr< LogStoreFamily >, num_log_families > m_logstore_families;
-    std::shared_ptr< JournalVirtualDev > m_data_logdev_vdev;
-    std::shared_ptr< JournalVirtualDev > m_ctrl_logdev_vdev;
+    std::unordered_map< logdev_id_t, std::shared_ptr< LogDev > > m_id_logdev_map;
+    std::unique_ptr< sisl::IDReserver > m_id_reserver;
+    folly::SharedMutexWritePriority m_logdev_map_mtx;
+
+    std::shared_ptr< JournalVirtualDev > m_logdev_vdev;
     iomgr::io_fiber_t m_truncate_fiber;
     iomgr::io_fiber_t m_flush_fiber;
     LogStoreServiceMetrics m_metrics;

src/lib/common/homestore_config.fbs

@@ -24,12 +24,12 @@ table BlkAllocator {
     /* Count of free blks cache in-terms of device size */
     free_blk_cache_count_by_vdev_percent: double = 80.0;
 
-    /* Percentage of overall memory allocated for blkallocator free blks cache. The memory allocated effictively is the 
+    /* Percentage of overall memory allocated for blkallocator free blks cache. The memory allocated effictively is the
      * min of memory occupied by (free_blk_cache_size_by_vdev_percent, max_free_blk_cache_memory_percent) */
     max_free_blk_cache_memory_percent: double = 1.0;
 
     /* Free blk cache slab distribution percentage
-     * An example assuming blk_size=4K is      [4K,   8K,   16K,  32K,  64K,  128K, 256K, 512K, 1M,  2M,  4M,  8M,  16M]  
+     * An example assuming blk_size=4K is      [4K,   8K,   16K,  32K,  64K,  128K, 256K, 512K, 1M,  2M,  4M,  8M,  16M]
      * free_blk_slab_distribution : [double] = [20.0, 10.0, 10.0, 10.0, 35.0, 3.0,  3.0,  3.0,  2.0, 1.0, 1.0, 1.0, 1.0] */
     free_blk_slab_distribution : [double];
 
@@ -54,7 +54,7 @@ table BlkAllocator {
 }
 
 table Btree {
-    max_nodes_to_rebalance: uint32 = 3; 
+    max_nodes_to_rebalance: uint32 = 3;
 
     mem_btree_page_size: uint32 = 8192;
 }
@@ -78,9 +78,9 @@ table Device {
 
     // Max completions to process per event in a thread
     max_completions_process_per_event_per_thread: uint32 = 200;
-    
+
     // DIRECT_IO mode, switch for HDD IO mode;
-    direct_io_mode: bool = false; 
+    direct_io_mode: bool = false;
 }
 
 table LogStore {
@@ -107,8 +107,7 @@ table LogStore {
     try_flush_iteration: uint64 = 10240(hotswap);
 
     // Logdev flushes in multiples of this size, setting to 0 will make it use default device optimal size
-    flush_size_multiple_data_logdev: uint64 = 0;
-    flush_size_multiple_ctrl_logdev: uint64 = 512;
+    flush_size_multiple_logdev: uint64 = 512;
 
     // Logdev will flush the logs only in a dedicated thread. Turn this on, if flush IO doesn't want to
     // intervene with data IO path.
@@ -137,7 +136,7 @@ table Generic {
     // number of threads for btree writes;
     num_btree_write_threads : uint32 = 2;
 
-    // percentage of cache used to create indx mempool. It should be more than 100 to 
+    // percentage of cache used to create indx mempool. It should be more than 100 to
     // take into account some floating buffers in writeback cache.
     indx_mempool_percent : uint32 = 110;
 
@@ -148,13 +147,13 @@ table Generic {
 table ResourceLimits {
     /* it is going to use 2 times of this space because of two concurrent cps */
     dirty_buf_percent: uint32 = 1 (hotswap);
-    
+
     /* it is going to use 2 times of this space because of two concurrent cps */
     free_blk_cnt: uint32 = 10000000 (hotswap);
     free_blk_size_percent: uint32 = 2 (hotswap);
-    
+
     /* Percentage of memory allocated for homestore cache */
-    cache_size_percent: uint32 = 65; 
+    cache_size_percent: uint32 = 65;
 
     /* precentage of memory used during recovery */
     memory_in_recovery_precent: uint32 = 40;
@@ -167,18 +166,18 @@ table ResourceLimits {
 }
 
 table MetaBlkStore {
-    // turn on/off compression feature 
+    // turn on/off compression feature
     compress_feature_on : bool = true (hotswap);
 
     // turn on/off skip header check
     skip_header_size_check : bool = false (hotswap);
 
-    // Compress buffer larger than this memory limit in MB will not trigger compress; 
+    // Compress buffer larger than this memory limit in MB will not trigger compress;
     max_compress_memory_size_mb: uint32 = 512 (hotswap);
-    
+
     // Inital memory allocated for compress buffer
     init_compress_memory_size_mb: uint32 = 10 (hotswap);
-    
+
     // Try to do compress only when input buffer is larger than this size
     min_compress_size_mb: uint32 = 1 (hotswap);
 
@@ -188,7 +187,7 @@ table MetaBlkStore {
     // percentage of *free* root fs while dump to file for get_status;
     percent_of_free_space: uint32 = 10 (hotswap);
 
-    // meta sanity check interval 
+    // meta sanity check interval
     sanity_check_interval: uint32 = 10 (hotswap);
 }
 

src/lib/device/journal_vdev.hpp

@@ -25,12 +25,11 @@
 #include "device.h"
 #include "physical_dev.hpp"
 #include "virtual_dev.hpp"
+#include <homestore/logstore/log_store_internal.hpp>
 
 namespace homestore {
 typedef std::function< void(const off_t ret_off) > alloc_next_blk_cb_t;
 using journal_id_t = uint64_t;
-// Each logstore family is associated to a single logdevice.
-using logdev_id_t = uint64_t;
 
 // Chunks used for journal vdev has journal related info stored in chunk private data.
 // Each log device has a list of journal chunk data with next_chunk.