buffered car append (#561)

Signed-off-by: turuslan <[email protected]>

Author: turuslan

core/api/full_node/make.cpp

@@ -557,6 +557,7 @@ namespace fc::api {
           OUTCOME_CB(auto it, find(ts_branch, context.tipset->height()));
           OUTCOME_CB(info.prev_beacon, latestBeacon(ts_load, it));
           OUTCOME_CB(auto it2, getLookbackTipSetForRound(it, epoch));
+          OUTCOME_CB(auto lookback_ts, ts_load->lazyLoad(it2.second->second));
           OUTCOME_CB(auto cached,
                      interpreter_cache->get(it2.second->second.key));
           ts_lock.unlock();
@@ -570,7 +571,7 @@ namespace fc::api {
               [=, FWD(cb), MOVE(context), MOVE(info)](auto _beacons) mutable {
                 OUTCOME_CB(info.beacons, _beacons);
                 TipsetContext lookback{
-                    nullptr,
+                    lookback_ts,
                     {withVersion(ipld, epoch), std::move(cached.state_root)},
                     {}};
                 OUTCOME_CB(auto actor, lookback.state_tree.tryGet(miner));

core/node/main/builder.cpp

@@ -219,6 +219,17 @@ namespace fc::node {
       }
     }
 
+    while (true) {
+      const auto it{std::prev(o.ts_main->chain.end())};
+      if (it == o.ts_main->chain.begin()
+          || o.env_context.interpreter_cache->tryGet(it->second.key)) {
+        break;
+      }
+      log()->warn("missing state at {}, reverting", it->first);
+      o.ts_main->updater->revert();
+      o.ts_main->chain.erase(it);
+    }
+
     log()->info("chain loaded");
     assert(o.ts_main->bottom().second.key == genesis_tsk);
   }
@@ -231,6 +242,7 @@ namespace fc::node {
         car_path, true, 1 << 30, o.ipld, log());
     // estimated
     ipld->flush_on = 200000;
+    ipld->car_flush_on = 100;
     o.ipld_flush_thread = std::make_shared<IoThread>();
     ipld->io = o.ipld_flush_thread->io;
     return ipld;
@@ -452,6 +464,10 @@ namespace fc::node {
     o.io_context = injector.create<std::shared_ptr<boost::asio::io_context>>();
     o.scheduler = injector.create<std::shared_ptr<Scheduler>>();
 
+    timerLoop(o.scheduler, std::chrono::minutes{1}, [ipld{o.compacter}] {
+      ipld->carFlush();
+    });
+
     o.events = std::make_shared<sync::events::Events>(o.io_context);
 
     o.host = injector.create<std::shared_ptr<libp2p::Host>>();

core/storage/car/cids_index/util.hpp

@@ -158,7 +158,7 @@ namespace fc::storage::cids_index {
     _ipld->index = index;
     _ipld->ipld = ipld;
     if (writable) {
-      _ipld->writable.open(car_path, std::ios::app | std::ios::binary);
+      _ipld->writable = {fopen(car_path.c_str(), "ab"), fclose};
     }
     _ipld->car_offset = car_size;
     _ipld->car_path = car_path;

core/storage/compacter/compacter.cpp

@@ -49,6 +49,11 @@ namespace fc::storage::compacter {
     }
   }
 
+  void CompacterIpld::carFlush() {
+    std::shared_lock lock{ipld_mutex};
+    (use_new_ipld ? new_ipld : old_ipld)->carFlush();
+  }
+
   void CompacterIpld::open() {
     if (start_head_key.has()) {
       resume();
@@ -82,6 +87,7 @@ namespace fc::storage::compacter {
     new_ipld = *car;
     new_ipld->io = old_ipld->io;
     new_ipld->flush_on = old_ipld->flush_on;
+    new_ipld->car_flush_on = old_ipld->car_flush_on;
     queue->visited = new_ipld;
     queue->open(true);
     std::unique_lock vm_lock{*interpreter->mutex};
@@ -98,6 +104,7 @@ namespace fc::storage::compacter {
     start_head_key.setCbor(start_head->key.cids());
     {
       std::unique_lock ipld_lock{ipld_mutex};
+      old_ipld->carFlush();
       use_new_ipld = true;
     }
     vm_lock.unlock();
@@ -118,6 +125,7 @@ namespace fc::storage::compacter {
     new_ipld = *car;
     new_ipld->io = old_ipld->io;
     new_ipld->flush_on = old_ipld->flush_on;
+    new_ipld->car_flush_on = old_ipld->car_flush_on;
     queue->visited = new_ipld;
     queue->open(false);
     start_head =
@@ -265,6 +273,7 @@ namespace fc::storage::compacter {
       std::unique_lock old_flush_lock{old_ipld->flush_mutex};
       std::unique_lock new_flush_lock{new_ipld->flush_mutex};
       std::unique_lock ipld_lock{ipld_mutex};
+      new_ipld->carFlush();
       // keep last car copy for debug
       boost::filesystem::rename(old_ipld->car_path,
                                 old_ipld->car_path + ".old_ipld");

core/storage/compacter/compacter.hpp

@@ -45,6 +45,8 @@ namespace fc::storage::compacter {
     bool get(const CbCid &key, Bytes *value) const override;
     void put(const CbCid &key, BytesCow &&value) override;
 
+    void carFlush();
+
     void open();
     bool asyncStart();
     void doStart();

core/storage/ipld/cids_ipld.cpp

@@ -9,6 +9,7 @@
 #include <boost/filesystem/operations.hpp>
 
 #include "cbor_blake/ipld_any.hpp"
+#include "codec/cbor/light_reader/cid.hpp"
 #include "codec/uvarint.hpp"
 #include "common/error_text.hpp"
 #include "common/logger.hpp"
@@ -20,7 +21,7 @@ namespace fc::storage::ipld {
   using cids_index::MergeRange;
 
   boost::optional<Row> CidsIpld::findWritten(const CbCid &key) const {
-    assert(writable.is_open());
+    assert(writable != nullptr);
     auto it{written.lower_bound(Row{key, {}, {}})};
     if (it != written.end() && it->key == key) {
       return *it;
@@ -42,7 +43,7 @@ namespace fc::storage::ipld {
 
   outcome::result<void> CidsIpld::set(const CID &cid, BytesCow &&value) {
     if (auto key{asBlake(cid)}) {
-      if (writable.is_open()) {
+      if (writable != nullptr) {
         put(*key, std::move(value));
         return outcome::success();
       }
@@ -128,7 +129,7 @@ namespace fc::storage::ipld {
     std::shared_lock index_lock{index_mutex};
     auto row{index->find(key).value()};
     index_lock.unlock();
-    if (!row && writable.is_open()) {
+    if (!row && writable != nullptr) {
       std::shared_lock written_lock{written_mutex};
       row = findWritten(key);
     }
@@ -139,6 +140,9 @@ namespace fc::storage::ipld {
       return false;
     }
     if (value != nullptr) {
+      if (carGet(*row, *value)) {
+        return true;
+      }
       std::unique_lock car_lock{car_mutex};
       auto [good, size]{readCarItem(car_file, *row, nullptr)};
       if (!good) {
@@ -155,7 +159,7 @@ namespace fc::storage::ipld {
   }
 
   void CidsIpld::put(const CbCid &key, BytesCow &&value) {
-    if (!writable.is_open()) {
+    if (writable == nullptr) {
       outcome::raise(ERROR_TEXT("CidsIpld.put: not writable"));
     }
     if (has(key)) {
@@ -179,22 +183,64 @@ namespace fc::storage::ipld {
     row.key = key;
     row.offset = car_offset;
     row.max_size64 = maxSize64(item.size());
-    if (!common::write(writable, item)) {
-      spdlog::error("CidsIpld.put write error");
-      outcome::raise(ERROR_TEXT("CidsIpld.put: write error"));
-    }
-    if (!writable.flush().good()) {
-      spdlog::error("CidsIpld.put flush error");
-      outcome::raise(ERROR_TEXT("CidsIpld.put: flush error"));
-    }
     car_offset += item.size();
+    carPut(row, std::move(item));
     written.insert(row);
     if (flush_on != 0 && written.size() >= flush_on) {
       written_lock.unlock();
       asyncFlush();
     }
   }
 
+  void CidsIpld::carPut(const Row &row, Bytes &&item) {
+    std::unique_lock lock{car_flush_mutex};
+    car_queue.emplace(row.offset.value(), car_queue_buffer.size());
+    append(car_queue_buffer, item);
+    if (car_queue.size() >= car_flush_on) {
+      carFlush(std::adopt_lock);
+    }
+  }
+
+  bool CidsIpld::carGet(const Row &row, Bytes &value) const {
+    std::shared_lock lock{car_flush_mutex};
+    const auto it{car_queue.find(row.offset.value())};
+    if (it == car_queue.end()) {
+      return false;
+    }
+    auto item{BytesIn{car_queue_buffer}.subspan(it->second)};
+    BytesIn input;
+    const CbCid *hash;
+    if (codec::uvarint::readBytes(input, item)
+        && codec::cbor::light_reader::readCborBlake(hash, input)) {
+      copy(value, input);
+      return true;
+    }
+    outcome::raise(ERROR_TEXT("CidsIpld.carGet decode error"));
+  }
+
+  void CidsIpld::carFlush(std::adopt_lock_t) {
+    if (car_queue.empty()) {
+      return;
+    }
+    if (fwrite(
+            car_queue_buffer.data(), car_queue_buffer.size(), 1, writable.get())
+        != 1) {
+      spdlog::error("CidsIpld.carFlush write error");
+      outcome::raise(ERROR_TEXT("CidsIpld.carFlush: write error"));
+    }
+    if (fflush(writable.get()) != 0) {
+      spdlog::error("CidsIpld.carFlush flush error");
+      outcome::raise(ERROR_TEXT("CidsIpld.carFlush: flush error"));
+    }
+    car_queue.clear();
+    car_queue_buffer.resize(0);
+  }
+
+  void CidsIpld::carFlush() {
+    std::unique_lock lock{car_flush_mutex};
+    carFlush(std::adopt_lock);
+  }
+
   void CidsIpld::asyncFlush() {
     if (!flushing.test_and_set()) {
       if (io) {

core/storage/ipld/cids_ipld.hpp

@@ -31,6 +31,11 @@ namespace fc::storage::ipld {
     bool get(const CbCid &key, Bytes *value) const override;
     void put(const CbCid &key, BytesCow &&value) override;
 
+    void carPut(const Row &row, Bytes &&item);
+    bool carGet(const Row &row, Bytes &value) const;
+    void carFlush(std::adopt_lock_t);
+    void carFlush();
+
     void asyncFlush();
 
     inline boost::optional<Row> findWritten(const CbCid &key) const;
@@ -41,7 +46,7 @@ namespace fc::storage::ipld {
     mutable std::shared_mutex index_mutex;
     std::shared_ptr<Index> index;
     IpldPtr ipld;
-    std::ofstream writable;
+    std::shared_ptr<FILE> writable;
     mutable std::shared_mutex written_mutex;
     std::set<Row> written;
     uint64_t car_offset{};
@@ -52,5 +57,9 @@ namespace fc::storage::ipld {
     std::string index_path;
     std::string car_path;
     boost::optional<size_t> max_memory;
+    mutable std::shared_mutex car_flush_mutex;
+    std::map<uint64_t, size_t> car_queue;
+    Bytes car_queue_buffer;
+    size_t car_flush_on{};
   };
 }  // namespace fc::storage::ipld

core/storage/mpool/mpool.cpp

@@ -794,8 +794,9 @@ namespace fc::storage::mpool {
               remove(msg->from, msg->nonce);
             } else {
               if (bls) {
-                std::lock_guard bls_cache_lock{bls_cache_mutex_};
+                std::unique_lock bls_cache_lock{bls_cache_mutex_};
                 if (auto sig{bls_cache.get(cid)}) {
+                  bls_cache_lock.unlock();
                   OUTCOME_TRY(add({*msg, *sig}));
                 }
               } else {

test/core/storage/car/cids_index_test.cpp

@@ -67,7 +67,7 @@ namespace fc::storage::cids_index {
     // inserted only once
     auto car_value1{*common::readFile(car_path)};
     EXPECT_OUTCOME_TRUE_1(setCbor(ipld, value1));
-    ipld->writable.flush();
+    fflush(ipld->writable.get());
     EXPECT_EQ(fs::file_size(car_path), car_value1.size());
 
     // truncated car drops index
@@ -133,4 +133,19 @@ namespace fc::storage::cids_index {
     IoThread thread;
     testFlush(thread.io);
   }
+
+  TEST_F(CidsIndexTest, CarFlush) {
+    ipld = *load(true);
+    const auto header{*common::readFile(car_path)};
+    ipld->car_flush_on = 2;
+
+    const auto c1{setCbor(ipld, 1).value()};
+    EXPECT_EQ(fs::file_size(car_path), header.size());
+    EXPECT_OUTCOME_EQ(getCbor<int>(ipld, c1), 1);
+
+    const auto c2{setCbor(ipld, 2).value()};
+    EXPECT_EQ(fs::file_size(car_path), header.size() + 2 * 40);
+    EXPECT_OUTCOME_EQ(getCbor<int>(ipld, c1), 1);
+    EXPECT_OUTCOME_EQ(getCbor<int>(ipld, c2), 2);
+  }
 }  // namespace fc::storage::cids_index