read API implementation

Author: Ravi Nagarjun Akella

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomeBlocksConan(ConanFile):
     name = "homeblocks"
-    version = "1.0.16"
+    version = "1.0.17"
     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

@@ -105,26 +105,14 @@ 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;
 
     void submit_io_batch() final;
 
-
-    NullAsyncResult write(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,
-        bool part_of_batch = false) final;
-
-    NullAsyncResult unmap(const VolumePtr& vol, const vol_interface_req_ptr& req) final;
-
-    void submit_io_batch() final;
-
-
     // see api comments in base class;
     bool get_stats(volume_id_t id, VolumeStats& stats) const final;
     void get_volume_ids(std::vector< volume_id_t >& vol_ids) const final;
@@ -160,6 +148,8 @@ 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,
+                        std::vector< std::pair< VolumeIndexKey, VolumeIndexValue > >& out_vector);
 };
 
 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,
+                         std::vector< std::pair< VolumeIndexKey, VolumeIndexValue > >& out_vector) {
+    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, out_vector); 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;
 
@@ -249,7 +252,7 @@ class VolumeIndexValue : public homestore::BtreeIntervalValue {
         // Get the next blk num and checksum
         m_blkid_suffix += n;
         auto curr_lba = ctx->start_lba + n;
-        LOGINFO("shift n={} blk_num={} curr_lba={}", n, m_blkid_suffix, curr_lba);
+        LOGTRACE("shift n={} blk_num={} curr_lba={}", n, m_blkid_suffix, curr_lba);
         DEBUG_ASSERT(ctx->block_info->find(curr_lba) != ctx->block_info->end(), "Invalid index");
         m_checksum = (*ctx->block_info)[curr_lba].checksum;
     }
@@ -281,4 +284,4 @@ class VolumeIndexValue : public homestore::BtreeIntervalValue {
                 (m_checksum == other.m_checksum));
     }
 };
-} // namespace homeblocks
+} // namespace homeblocks

src/lib/volume/tests/test_common.hpp

@@ -213,6 +213,12 @@ class HBTestHelper {
         }
     }
 
+    static void validate_zeros(uint8_t const* buf, uint64_t size) {
+        sisl::io_blob_safe blob(size, 512);
+        std::memset(blob.bytes(), 0, size);
+        RELEASE_ASSERT_EQ(std::memcmp(buf, blob.bytes(), size), 0, "data_buf mismatch");
+    }
+
 private:
     void init_devices(bool is_file, uint64_t dev_size = 0) {
         if (is_file) {
@@ -277,4 +283,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

@@ -96,7 +96,7 @@ class VolumeIOImpl {
             data_bytes += page_size;
             // Store the lba to pattern mapping
             m_lba_data[lba] = data_pattern;
-            LOGINFO("Generate data lba={} pattern={}", lba, data_pattern);
+            //LOGDEBUG("Generate data lba={} pattern={}", lba, data_pattern);
             lba++;
         }
 
@@ -131,18 +131,44 @@ class VolumeIOImpl {
         });
     }
 
-    void verify_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);
+            } else {
+                test_common::HBTestHelper::validate_zeros(buf, m_vol_ptr->info()->page_size);
+            }
+            
+            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();
-            // LOGINFO("Data read={}", fmt::format("{}", spdlog::to_hex((buffer), (buffer) + (128))));
-            test_common::HBTestHelper::validate_data_buf(buffer, 4096, data_pattern);
-            LOGINFO("Verify data lba={} pattern={} {}", lba, data_pattern, *r_cast< uint64_t* >(buffer));
-            iomanager.iobuf_free(buffer);
+    void verify_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
@@ -191,22 +217,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_data(shared< VolumeIOImpl > vol_impl = nullptr) {
+    void verify_data(shared< VolumeIOImpl > vol_impl = nullptr, uint64_t nlbas_per_io = 1) {
         if (vol_impl) {
-            vol_impl->verify_data();
+            vol_impl->verify_data(nlbas_per_io);
             return;
         }
 
         for (auto& vol_impl : m_vols_impl) {
-            vol_impl->verify_data();
+            vol_impl->verify_data(nlbas_per_io);
         }
     }
 
@@ -233,18 +258,38 @@ TEST_F(VolumeIOTest, SingleVolumeWriteData) {
     vol->reset();
 
     LOGINFO("Verify data");
-    verify_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_data(vol);
+
+    // verify random lba ranges
+
     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 = 500;
+    lba_t start_lba = 1000;
+    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);
+
+    LOGINFO("SingleVolumeReadHoles test done.");
+}
+
 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 >());