Implement homeblock write path.

Write flow writes data to disk, add key value to index
and flush journal entry to journal. Add test case to test
the normal io flow.

Author: sanebay

conanfile.py

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

src/CMakeLists.txt

@@ -2,6 +2,7 @@ cmake_minimum_required (VERSION 3.11)
 
 add_flags("-Wall -Wextra -Werror -Wpedantic")
 include_directories (BEFORE include/)
+include_directories (BEFORE lib)
 include_directories (AFTER .)
 
 find_package(Threads QUIET REQUIRED)

src/include/homeblks/volume_mgr.hpp

@@ -11,11 +11,11 @@
 namespace homeblocks {
 
 ENUM(VolumeError, uint16_t, UNKNOWN = 1, INVALID_ARG, TIMEOUT, UNKNOWN_VOLUME, UNSUPPORTED_OP, CRC_MISMATCH,
-     NO_SPACE_LEFT, DRIVE_WRITE_ERROR, INTERNAL_ERROR);
+     NO_SPACE_LEFT, DRIVE_WRITE_ERROR, INTERNAL_ERROR, INDEX_ERROR);
+
+using lba_t = uint64_t;
+using lba_count_t = uint32_t;
 
-    using lba_t = uint64_t;
-    using lba_count_t = uint32_t;
-    
 struct vol_interface_req : public sisl::ObjLifeCounter< vol_interface_req > {
     uint8_t* buffer{nullptr};
     lba_t lba;
@@ -62,7 +62,7 @@ struct VolumeInfo {
 
     std::string to_string() {
         return fmt::format("VolumeInfo: id={} size_bytes={}, page_size={}, name={}", boost::uuids::to_string(id),
-                            size_bytes, page_size, name);
+                           size_bytes, page_size, name);
     }
 };
 
@@ -102,36 +102,36 @@ class VolumeManager : public Manager< VolumeError > {
      * @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;
+                                  bool part_of_batch = false) = 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
-    * 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
-    */
+     * @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
+     * 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;
+                                 bool part_of_batch = false) = 0;
 
     /**
-    * @brief unmap the given block range
-    *
-    * @param vol Pointer to the volume
-    * @param req Request created which contains all the read parameters
-    */
+     * @brief unmap the given block range
+     *
+     * @param vol Pointer to the volume
+     * @param req Request created which contains all the read parameters
+     */
     virtual NullAsyncResult unmap(const VolumePtr& vol, const vol_interface_req_ptr& req) = 0;
 
     /**
-    * @brief Submit the io batch, which is a mandatory method to be called if read/write are issued with part_of_batch
-    * is set to true. In those cases, without this method, IOs might not be even issued. No-op if previous io requests
-    * are not part of batch.
-    */
+     * @brief Submit the io batch, which is a mandatory method to be called if read/write are issued with part_of_batch
+     * is set to true. In those cases, without this method, IOs might not be even issued. No-op if previous io requests
+     * are not part of batch.
+     */
     virtual void submit_io_batch() = 0;
 
     /**

src/lib/CMakeLists.txt

@@ -7,6 +7,7 @@ target_sources(${PROJECT_NAME}_core PRIVATE
     homeblks_impl.cpp
     volume_mgr.cpp
     listener.cpp
+    index.cpp
     common.cpp
 )
 target_link_libraries(${PROJECT_NAME}_core

src/lib/homeblks_impl.hpp

@@ -18,6 +18,8 @@
 #include <map>
 #include <string>
 #include <sisl/logging/logging.h>
+#include <sisl/utility/obj_life_counter.hpp>
+#include <homestore/crc.h>
 #include <homestore/homestore.hpp>
 #include <homestore/index/index_table.hpp>
 #include <homestore/superblk_handler.hpp>
@@ -30,6 +32,14 @@ namespace homeblocks {
 
 class Volume;
 
+ENUM(HomeBlksMsgType, uint8_t, READ, WRITE, UNMAP);
+
+struct VolJournalEntry {
+    lba_t start_lba;
+    lba_count_t nlbas;
+    uint16_t num_old_blks;
+};
+
 class HomeBlocksImpl : public HomeBlocks, public VolumeManager, public std::enable_shared_from_this< HomeBlocksImpl > {
     struct homeblks_sb_t {
         uint64_t magic;
@@ -119,6 +129,9 @@ class HomeBlocksImpl : public HomeBlocks, public VolumeManager, public std::enab
 
     void on_init_complete();
 
+    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);
+
 private:
     // Should only be called for first-time-boot
     void superblk_init();
@@ -133,6 +146,10 @@ class HomeBlocksImpl : public HomeBlocks, public VolumeManager, public std::enab
     // recovery apis
     void on_hb_meta_blk_found(sisl::byte_view const& buf, void* cookie);
     void on_vol_meta_blk_found(sisl::byte_view const& buf, void* cookie);
+
+    VolumeManager::Result< folly::Unit > write_to_index(const VolumePtr& vol_ptr, const vol_interface_req_ptr& vol_req,
+                                                        lba_t start_lba, lba_t end_lba,
+                                                        std::unordered_map< lba_t, BlockInfo >& blocks_info);
 };
 
 class HBIndexSvcCB : public homestore::IndexServiceCallbacks {
@@ -147,4 +164,87 @@ class HBIndexSvcCB : public homestore::IndexServiceCallbacks {
 private:
     HomeBlocksImpl* hb_;
 };
+
+const homestore::csum_t init_crc_16 = 0x8005;
+
+struct HomeBlksMessageHeader {
+    HomeBlksMessageHeader() = default;
+    HomeBlksMsgType msg_type;
+    volume_id_t volume_id;
+
+    std::string to_string() const {
+        return fmt::format(" msg_type={}volume={}\n", enum_name(msg_type), boost::uuids::to_string(volume_id));
+    }
+};
+
+struct homeblks_repl_ctx : public homestore::repl_req_ctx {
+    sisl::io_blob_safe hdr_buf_;
+    sisl::io_blob_safe key_buf_;
+
+    // Data bufs corresponding to data_sgs_. Since data_sgs are raw pointers, we need to keep the data bufs alive
+    folly::small_vector< sisl::io_blob_safe, 3 > data_bufs_;
+    sisl::sg_list data_sgs_;
+
+    homeblks_repl_ctx(uint32_t hdr_extn_size, uint32_t key_size = 0) : homestore::repl_req_ctx{} {
+        hdr_buf_ = std::move(sisl::io_blob_safe{uint32_cast(sizeof(HomeBlksMessageHeader) + hdr_extn_size), 0});
+        new (hdr_buf_.bytes()) HomeBlksMessageHeader();
+
+        if (key_size) { key_buf_ = std::move(sisl::io_blob_safe{key_size, 0}); }
+        data_sgs_.size = 0;
+    }
+
+    ~homeblks_repl_ctx() {
+        if (hdr_buf_.bytes()) { header()->~HomeBlksMessageHeader(); }
+    }
+
+    template < typename T >
+    T* to() {
+        return r_cast< T* >(this);
+    }
+
+    HomeBlksMessageHeader* header() { return r_cast< HomeBlksMessageHeader* >(hdr_buf_.bytes()); }
+    uint8_t* header_extn() { return hdr_buf_.bytes() + sizeof(HomeBlksMessageHeader); }
+
+    sisl::io_blob_safe& header_buf() { return hdr_buf_; }
+    sisl::io_blob_safe const& cheader_buf() const { return hdr_buf_; }
+
+    sisl::io_blob_safe& key_buf() { return key_buf_; }
+    sisl::io_blob_safe const& ckey_buf() const { return key_buf_; }
+
+    void add_data_sg(uint8_t* buf, uint32_t size) {
+        data_sgs_.iovs.emplace_back(iovec{.iov_base = buf, .iov_len = size});
+        data_sgs_.size += size;
+    }
+
+    void add_data_sg(sisl::io_blob_safe&& buf) {
+        add_data_sg(buf.bytes(), buf.size());
+        data_bufs_.emplace_back(std::move(buf));
+    }
+
+    sisl::sg_list& data_sgs() { return data_sgs_; }
+    std::string data_sgs_string() const {
+        fmt::memory_buffer buf;
+        fmt::format_to(fmt::appender(buf), "total_size={} iovcnt={} [", data_sgs_.size, data_sgs_.iovs.size());
+        for (auto const& iov : data_sgs_.iovs) {
+            fmt::format_to(fmt::appender(buf), "<base={},len={}> ", iov.iov_base, iov.iov_len);
+        }
+        fmt::format_to(fmt::appender(buf), "]");
+        return fmt::to_string(buf);
+    }
+};
+
+template < typename T >
+struct repl_result_ctx : public homeblks_repl_ctx {
+    folly::Promise< T > promise_;
+    VolumePtr vol_ptr_{nullptr};
+
+    template < typename... Args >
+    static intrusive< repl_result_ctx< T > > make(Args&&... args) {
+        return intrusive< repl_result_ctx< T > >{new repl_result_ctx< T >(std::forward< Args >(args)...)};
+    }
+
+    repl_result_ctx(uint32_t hdr_extn_size, uint32_t key_size = 0) : homeblks_repl_ctx{hdr_extn_size, key_size} {}
+    folly::SemiFuture< T > result() { return promise_.getSemiFuture(); }
+};
+
 } // namespace homeblocks

src/lib/index.cpp

@@ -0,0 +1,54 @@
+/*********************************************************************************
+ * Modifications Copyright 2017-2019 eBay Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *    https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed
+ * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+ * CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ *********************************************************************************/
+#include "volume/volume.hpp"
+#include "homeblks_impl.hpp"
+
+namespace homeblocks {
+
+VolumeManager::Result< folly::Unit >
+HomeBlocksImpl::write_to_index(const VolumePtr& vol_ptr, const vol_interface_req_ptr& vol_req, lba_t start_lba,
+                               lba_t end_lba, std::unordered_map< lba_t, BlockInfo >& blocks_info) {
+
+    // Use filter callback to get the old blkid.
+    homestore::put_filter_cb_t filter_cb = [&blocks_info](BtreeKey const& key, BtreeValue const& existing_value,
+                                                          BtreeValue const& value) {
+        auto lba = r_cast< const VolumeIndexKey& >(key).key();
+        blocks_info[lba].old_blkid = r_cast< const VolumeIndexValue& >(existing_value).blkid();
+        return homestore::put_filter_decision::replace;
+    };
+
+    // Write to prefix btree with key ranging from start_lba to end_lba.
+    // 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};
+
+    auto req = homestore::BtreeRangePutRequest< VolumeIndexKey >{
+        homestore::BtreeKeyRange< VolumeIndexKey >{VolumeIndexKey{start_lba}, true, VolumeIndexKey{end_lba}, true},
+        homestore::btree_put_type::UPSERT,
+        r_cast< VolumeIndexValue* >(&value),
+        r_cast< void* >(&app_ctx),
+        std::numeric_limits< uint32_t >::max(),
+        filter_cb};
+    auto result = vol_ptr->indx_table()->put(req);
+    if (result != homestore::btree_status_t::success) {
+        LOGERROR("Failed to put to index range=({},{}) error={}", start_lba, end_lba, result);
+        return folly::makeUnexpected(VolumeError::INDEX_ERROR);
+    }
+
+    return folly::Unit();
+}
+
+} // namespace homeblocks

src/lib/listener.cpp

@@ -19,7 +19,22 @@ namespace homeblocks {
 
 void HBListener::on_commit(int64_t lsn, sisl::blob const& header, sisl::blob const& key,
                            std::vector< homestore::MultiBlkId > const& blkids,
-                           cintrusive< homestore::repl_req_ctx >& ctx) {}
+                           cintrusive< homestore::repl_req_ctx >& ctx) {
+    // on_commit called whenever journal has flushed log entries. header contains the msg type and volume
+    // id, key contains the list of checksum, list of old blkids for write case. blkid's are the new blkid's
+    // where data is written.
+    const HomeBlksMessageHeader* msg_header = r_cast< const HomeBlksMessageHeader* >(header.cbytes());
+    switch (msg_header->msg_type) {
+    case HomeBlksMsgType::WRITE:
+        hb_->on_write(lsn, header, key, blkids, ctx);
+        break;
+
+    case HomeBlksMsgType::READ:
+        break;
+    case HomeBlksMsgType::UNMAP:
+        break;
+    }
+}
 
 bool HBListener::on_pre_commit(int64_t lsn, const sisl::blob& header, const sisl::blob& key,
                                cintrusive< homestore::repl_req_ctx >& ctx) {