Read API impl

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomeBlocksConan(ConanFile):
     name = "homeblocks"
-    version = "1.0.18"
+    version = "1.0.19"
     homepage = "https://github.com/eBay/HomeBlocks"
     description = "Block Store built on HomeStore"
     topics = ("ebay")

src/include/homeblks/volume_mgr.hpp

@@ -36,6 +36,7 @@ struct vol_interface_req : public sisl::ObjLifeCounter< vol_interface_req > {
 
     virtual ~vol_interface_req() = default; // override; sisl::ObjLifeCounter should have virtual destructor
     virtual void free_yourself() { delete this; }
+    lba_t end_lba() const { return lba + nlbas - 1; }
 };
 
 using vol_interface_req_ptr = boost::intrusive_ptr< vol_interface_req >;
@@ -95,29 +96,27 @@ class VolumeManager : public Manager< VolumeError > {
      *
      * @param vol Pointer to the volume
      * @param req Request created which contains all the write parameters
-     * @param part_of_batch Is this request part of a batch request. If so, implementation can wait for batch_submit
+     * req.part_of_batch field can be used if this request is part of a batch request. If so, implementation can wait for batch_submit
      * call before issuing the writes. IO might already be started or even completed (in case of errors) before
      * batch_sumbit call, so application cannot assume IO will be started only after submit_batch call.
      *
      * @return std::error_condition no_error or error in issuing writes
      */
-    virtual NullAsyncResult write(const VolumePtr& vol, const vol_interface_req_ptr& req,
-                                  bool part_of_batch = false) = 0;
+    virtual NullAsyncResult write(const VolumePtr& vol, const vol_interface_req_ptr& req) = 0;
 
     /**
      * @brief Read the data from the volume asynchronously, created from the request. After completion the attached
      * callback function will be called with this req ptr.
      *
      * @param vol Pointer to the volume
      * @param req Request created which contains all the read parameters
-     * @param part_of_batch Is this request part of a batch request. If so, implementation can wait for batch_submit
+     * req.part_of_batch field can be used if this request is part of a batch request. If so, implementation can wait for batch_submit
      * call before issuing the reads. IO might already be started or even completed (in case of errors) before
      * batch_sumbit call, so application cannot assume IO will be started only after submit_batch call.
      *
      * @return std::error_condition no_error or error in issuing reads
      */
-    virtual NullAsyncResult read(const VolumePtr& vol, const vol_interface_req_ptr& req,
-                                 bool part_of_batch = false) = 0;
+    virtual NullAsyncResult read(const VolumePtr& vol, const vol_interface_req_ptr& req) = 0;
 
     /**
      * @brief unmap the given block range

src/lib/homeblks_impl.hpp

@@ -40,6 +40,16 @@ struct VolJournalEntry {
     uint16_t num_old_blks;
 };
 
+using index_kv_list_t = std::vector< std::pair< VolumeIndexKey, VolumeIndexValue > >;
+using read_blks_list_t = std::vector< std::pair< lba_t, homestore::MultiBlkId > >;
+
+struct vol_read_ctx {
+    uint8_t* buf;
+    lba_t start_lba;
+    uint32_t blk_size;
+    index_kv_list_t index_kvs{};
+};
+
 class HomeBlocksImpl : public HomeBlocks, public VolumeManager, public std::enable_shared_from_this< HomeBlocksImpl > {
     struct homeblks_sb_t {
         uint64_t magic;
@@ -105,9 +115,9 @@ class HomeBlocksImpl : public HomeBlocks, public VolumeManager, public std::enab
 
     VolumePtr lookup_volume(const volume_id_t& id) final;
 
-    NullAsyncResult write(const VolumePtr& vol, const vol_interface_req_ptr& req, bool part_of_batch = false) final;
+    NullAsyncResult write(const VolumePtr& vol, const vol_interface_req_ptr& req) final;
 
-    NullAsyncResult read(const VolumePtr& vol, const vol_interface_req_ptr& req, bool part_of_batch = false) final;
+    NullAsyncResult read(const VolumePtr& vol, const vol_interface_req_ptr& req) final;
 
     NullAsyncResult unmap(const VolumePtr& vol, const vol_interface_req_ptr& req) final;
 
@@ -133,6 +143,8 @@ class HomeBlocksImpl : public HomeBlocks, public VolumeManager, public std::enab
     void on_write(int64_t lsn, const sisl::blob& header, const sisl::blob& key,
                   const std::vector< homestore::MultiBlkId >& blkids, cintrusive< homestore::repl_req_ctx >& ctx);
 
+    VolumeManager::Result< folly::Unit > verify_checksum(vol_read_ctx const& read_ctx);
+
 private:
     // Should only be called for first-time-boot
     void superblk_init();
@@ -150,6 +162,11 @@ class HomeBlocksImpl : public HomeBlocks, public VolumeManager, public std::enab
 
     VolumeManager::Result< folly::Unit > write_to_index(const VolumePtr& vol_ptr, lba_t start_lba, lba_t end_lba,
                                                         std::unordered_map< lba_t, BlockInfo >& blocks_info);
+    VolumeManager::Result< folly::Unit > read_from_index(const VolumePtr& vol_ptr, const vol_interface_req_ptr& req,
+                                                         index_kv_list_t& index_kvs);
+    void generate_blkids_to_read(const index_kv_list_t& index_kvs, read_blks_list_t& blks_to_read);
+    void submit_read_to_backend(read_blks_list_t const& blks_to_read, const vol_interface_req_ptr& req, 
+                                const VolumePtr& vol, std::vector< folly::Future< std::error_code > >& futs);
 };
 
 class HBIndexSvcCB : public homestore::IndexServiceCallbacks {

src/lib/index.cpp

@@ -33,7 +33,7 @@ HomeBlocksImpl::write_to_index(const VolumePtr& vol_ptr, lba_t start_lba, lba_t
     // For value shift() will get the blk_num and checksum for each lba.
     IndexValueContext app_ctx{&blocks_info, start_lba};
     const BlkId& start_blkid = blocks_info[start_lba].new_blkid;
-    VolumeIndexValue value{start_blkid};
+    VolumeIndexValue value{start_blkid, blocks_info[start_lba].checksum};
 
     auto req = homestore::BtreeRangePutRequest< VolumeIndexKey >{
         homestore::BtreeKeyRange< VolumeIndexKey >{VolumeIndexKey{start_lba}, true, VolumeIndexKey{end_lba}, true},
@@ -51,4 +51,16 @@ HomeBlocksImpl::write_to_index(const VolumePtr& vol_ptr, lba_t start_lba, lba_t
     return folly::Unit();
 }
 
+VolumeManager::Result< folly::Unit > HomeBlocksImpl::read_from_index(const VolumePtr& vol_ptr, const vol_interface_req_ptr& req,
+                                                                    index_kv_list_t& index_kvs) {
+    homestore::BtreeQueryRequest< VolumeIndexKey > qreq{homestore::BtreeKeyRange< VolumeIndexKey >{VolumeIndexKey{req->lba}, 
+                                VolumeIndexKey{req->end_lba()}}, homestore::BtreeQueryType::SWEEP_NON_INTRUSIVE_PAGINATION_QUERY};
+    auto index_table = vol_ptr->indx_table();
+    RELEASE_ASSERT(index_table != nullptr, "Index table is null for volume id: {}", boost::uuids::to_string(vol_ptr->id()));
+    if (auto ret = index_table->query(qreq, index_kvs); ret != homestore::btree_status_t::success) {
+        return folly::makeUnexpected(VolumeError::INDEX_ERROR);
+    }
+    return folly::Unit();
+}
+
 } // namespace homeblocks

src/lib/tests/CMakeLists.txt

@@ -6,4 +6,4 @@ target_sources(test_fixture PRIVATE
 )
 target_link_libraries(test_fixture
     ${COMMON_TEST_DEPS}
-)
+)

src/lib/volume/index.hpp

@@ -188,15 +188,18 @@ class VolumeIndexValue : public homestore::BtreeIntervalValue {
 #pragma pack()
 
 public:
-    VolumeIndexValue(const BlkId& base_blkid) : homestore::BtreeIntervalValue() {
+    VolumeIndexValue(const BlkId& base_blkid, homestore::csum_t csum) : homestore::BtreeIntervalValue() {
         m_blkid_suffix = uint32_cast(base_blkid.to_integer() & 0xFFFFFFFF) >> 1;
         m_blkid_prefix = uint32_cast(base_blkid.to_integer() >> 32);
+        m_checksum = csum;
     }
+    VolumeIndexValue(const BlkId& base_blkid) : VolumeIndexValue(base_blkid, 0) {}
     VolumeIndexValue() = default;
     VolumeIndexValue(const VolumeIndexValue& other) :
             homestore::BtreeIntervalValue(),
             m_blkid_prefix(other.m_blkid_prefix),
-            m_blkid_suffix(other.m_blkid_suffix) {}
+            m_blkid_suffix(other.m_blkid_suffix),
+            m_checksum(other.m_checksum) {}
     VolumeIndexValue(const sisl::blob& b, bool copy) : homestore::BtreeIntervalValue() { this->deserialize(b, copy); }
     virtual ~VolumeIndexValue() = default;
 
@@ -280,4 +283,4 @@ class VolumeIndexValue : public homestore::BtreeIntervalValue {
                 (m_checksum == other.m_checksum));
     }
 };
-} // namespace homeblocks
+} // namespace homeblocks

src/lib/volume/tests/test_common.hpp

@@ -277,4 +277,4 @@ class HBTestHelper {
     Waiter waiter_;
 };
 
-} // namespace test_common
+} // namespace test_common

src/lib/volume/tests/test_volume.cpp

@@ -26,8 +26,8 @@
 
 SISL_LOGGING_INIT(HOMEBLOCKS_LOG_MODS)
 SISL_OPTION_GROUP(test_volume_setup,
-                  (num_vols, "", "num_vols", "number of volumes", ::cxxopts::value< uint32_t >()->default_value("2"),
-                   "number"));
+                (num_vols, "", "num_vols", "number of volumes", ::cxxopts::value< uint32_t >()->default_value("2"),
+                "number"));
 SISL_OPTIONS_ENABLE(logging, test_common_setup, test_volume_setup, homeblocks)
 SISL_LOGGING_DECL(test_volume)
 
@@ -193,4 +193,4 @@ int main(int argc, char* argv[]) {
     g_helper->teardown();
 
     return ret;
-}
+}

src/lib/volume/tests/test_volume_io.cpp

@@ -174,18 +174,42 @@ class VolumeIOImpl {
             });
     }
 
-    void verify_all_data() {
-        for (auto& [lba, data_pattern] : m_lba_data) {
-            auto buffer = iomanager.iobuf_alloc(512, 4096);
-            vol_interface_req_ptr req(new vol_interface_req{buffer, lba, 1});
+    void read_and_verify(lba_t start_lba, uint32_t nlbas) {
+        auto sz = nlbas * m_vol_ptr->info()->page_size;
+        sisl::io_blob_safe read_blob(sz, 512);
+        auto buf = read_blob.bytes();
+        vol_interface_req_ptr req(new vol_interface_req{buf, start_lba, nlbas});
+        auto read_resp = g_helper->inst()->volume_manager()->read(m_vol_ptr, req).get();
+        if(read_resp.hasError()) {
+            LOGERROR("Read failed with error={}", read_resp.error());
+        }
+        RELEASE_ASSERT(!read_resp.hasError(), "Read failed with error={}", read_resp.error());
+        auto read_sz = m_vol_ptr->info()->page_size;
+        for(auto lba = start_lba; lba < start_lba + nlbas; lba++, buf += read_sz) {
+            uint64_t data_pattern = 0;
+            if(auto it = m_lba_data.find(lba); it != m_lba_data.end()) {
+                data_pattern = it->second;
+                test_common::HBTestHelper::validate_data_buf(buf, m_vol_ptr->info()->page_size, data_pattern);
+            }
+
+            LOGDEBUG("Verify data lba={} pattern expected={} actual={}", lba, data_pattern, *r_cast< uint64_t* >(read_blob.bytes()));
+        } 
+    }
 
-            auto vol_mgr = g_helper->inst()->volume_manager();
-            vol_mgr->read(m_vol_ptr, req).get();
-            test_common::HBTestHelper::validate_data_buf(buffer, 4096, data_pattern);
-            LOGDEBUG("Verify data vol={} lba={} pattern={} {}", m_vol_name, lba, data_pattern,
-                     *r_cast< uint64_t* >(buffer));
-            iomanager.iobuf_free(buffer);
+    void verify_all_data(uint64_t nlbas_per_io = 1) {
+        auto start_lba = m_lba_data.begin()->first;
+        auto max_lba = m_lba_data.rbegin()->first;
+        verify_data(start_lba, max_lba, nlbas_per_io);
+    }
+
+    void verify_data(lba_t start_lba, lba_t max_lba, uint64_t nlbas_per_io) {
+        uint64_t num_lbas_verified = 0;
+        for(auto lba = start_lba; lba < max_lba; lba += nlbas_per_io) {
+            auto num_lbas_this_round = std::min(nlbas_per_io, max_lba - lba);
+            read_and_verify(lba, num_lbas_this_round);
+            num_lbas_verified += num_lbas_this_round;
         }
+        LOGINFO("Verified {} lbas for volume {}", num_lbas_verified, m_vol_ptr->info()->name);
     }
 
 #ifdef _PRERELEASE
@@ -238,22 +262,21 @@ class VolumeIOTest : public ::testing::Test {
                 // Get a random volume.
                 vol = m_vols_impl[rand() % m_vols_impl.size()];
             }
-
             vol->generate_io(start_lba, nblks);
         });
 
         if (wait) { g_helper->runner().execute().get(); }
         LOGINFO("IO completed");
     }
 
-    void verify_all_data(shared< VolumeIOImpl > vol_impl = nullptr) {
+    void verify_all_data(shared< VolumeIOImpl > vol_impl = nullptr, uint64_t nlbas_per_io = 1) {
         if (vol_impl) {
-            vol_impl->verify_all_data();
+            vol_impl->verify_all_data(nlbas_per_io);
             return;
         }
 
         for (auto& vol_impl : m_vols_impl) {
-            vol_impl->verify_all_data();
+            vol_impl->verify_all_data(nlbas_per_io);
         }
     }
 
@@ -266,6 +289,14 @@ class VolumeIOTest : public ::testing::Test {
 
     std::vector< shared< VolumeIOImpl > >& volume_list() { return m_vols_impl; }
 
+    template < typename T >
+    T get_random_number(T min, T max) {
+        static std::random_device rd;
+        static std::mt19937 gen(rd());
+        std::uniform_int_distribution< T > dis(min, max);
+        return dis(gen);
+    }
+
 private:
     std::vector< shared< VolumeIOImpl > > m_vols_impl;
 };
@@ -287,17 +318,76 @@ TEST_F(VolumeIOTest, SingleVolumeWriteData) {
 
     LOGINFO("Verify data");
     verify_all_data(vol);
+    //verify_data(vol, 30 /* nlbas_per_io */);
 
     // Write and verify again on same LBA range to single volume multiple times.
     LOGINFO("Write and verify data with num_iter={} start={} nblks={}", num_iter, start_lba, nblks);
     for (uint32_t i = 0; i < num_iter; i++) {
         generate_io_single(vol, start_lba, nblks);
     }
 
-    verify_all_data(vol);
+    verify_all_data(vol, 30 /* nlbas_per_io */);
+
     LOGINFO("SingleVolumeWriteData test done.");
 }
 
+TEST_F(VolumeIOTest, SingleVolumeReadData) {
+    // Write and verify fixed LBA range to single volume multiple times.
+    auto vol = volume_list().back();
+    uint32_t nblks = 5000;
+    lba_t start_lba = 500;
+    uint32_t num_iter = 1;
+    LOGINFO("Write and verify data with num_iter={} start={} nblks={}", num_iter, start_lba, nblks);
+    for (uint32_t i = 0; i < num_iter; i++) {
+        generate_io_single(vol, start_lba, nblks);
+    }
+
+    vol->verify_data(300, 800, 40);
+    vol->verify_data(2000, 3000, 40);
+    vol->verify_data(800, 1800, 40);
+
+    // random reads
+    num_iter = 100;
+    for(uint32_t i = 0; i < num_iter; i++) {
+        auto start_lba = get_random_number< lba_t >(0, 10000);
+        auto nblks = get_random_number< uint32_t >(1, 64);
+        auto no_lbas_per_io = get_random_number< uint64_t >(1, 50);
+        LOGINFO("iter {}: Read data start={} nblks={} no_lbas_per_io {}", i, start_lba, nblks, no_lbas_per_io);
+        vol->verify_data(start_lba, start_lba + nblks, no_lbas_per_io);
+    }
+
+    LOGINFO("SingleVolumeRead test done.");
+}
+
+TEST_F(VolumeIOTest, SingleVolumeReadHoles) {
+    auto vol = volume_list().back();
+    uint32_t nblks = 5000;
+    lba_t start_lba = 500;
+    generate_io_single(vol, start_lba, nblks);
+
+    // Verify with no holes in the range
+    vol->verify_data(1000, 2000, 40);
+
+    start_lba = 10000;
+    nblks = 50;
+    for(uint32_t i = 0; i/2 < nblks; i+=2) {
+        generate_io_single(vol, start_lba+i, 1);
+    }
+
+    // Verfy with hole after each lba
+    vol->verify_data(10000, 10100, 50);
+
+    start_lba = 20000;
+    for(uint32_t i = 0; i < 100; i++) {
+        if(i%7 > 2) {
+            generate_io_single(vol, start_lba+i, 1);
+        }
+    }
+    // Verify with mixed holes in the range
+    vol->verify_data(20000, 20100, 50);
+
+}
+
 TEST_F(VolumeIOTest, MultipleVolumeWriteData) {
     LOGINFO("Write data randomly on num_vols={} num_io={}", SISL_OPTIONS["num_vols"].as< uint32_t >(),
             SISL_OPTIONS["num_io"].as< uint64_t >());