[feat][cache]: Add prefetcher.

confv2/dingo-cache.conf

@@ -3,12 +3,12 @@
 --loglevel=0
 
 ############################# cache group node
---group_nane=default
+--group_name=default
 --listen_ip=127.0.0.1
 --listen_port=20000
 --group_weight=100
 --max_range_size_kb=128
---metadata_filepath="/tmp/cache_group_metadata"
+--metadata_filepath=/tmp/cache_group_metadata
 --send_heartbeat_interval_s=10
 --mds_rpc_addrs=127.0.0.1:6700,127.0.0.1:6701,127.0.0.1:6702
 --mds_rpc_retry_total_ms=16000

src/cache/benchmark/factory.cpp

@@ -104,24 +104,34 @@ RangeTaskFactory::RangeTaskFactory(BlockCacheSPtr block_cache)
     : block_cache_(block_cache) {}
 
 Task RangeTaskFactory::GenTask(const BlockKey& key) {
-  return [this, key]() { Range(key); };
+  return [this, key]() { RangeAll(key); };
 }
 
-void RangeTaskFactory::Range(const BlockKey& key) {
+void RangeTaskFactory::RangeAll(const BlockKey& key) {
+  IOBuffer buffer;
+  off_t offset = FLAGS_offset;
+  size_t blksize = FLAGS_blksize;
+  while (blksize) {
+    size_t length = std::min(FLAGS_blksize - offset, FLAGS_length);
+    Range(key, offset, length, &buffer);
+
+    offset += length;
+    blksize -= length;
+  }
+}
+
+void RangeTaskFactory::Range(const BlockKey& key, off_t offset, size_t length,
+                             IOBuffer* buffer) {
   auto option = RangeOption();
   option.retrive = FLAGS_retrive;
   option.block_size = FLAGS_blksize;
+  auto status =
+      block_cache_->Range(NewContext(), key, offset, length, buffer, option);
 
-  IOBuffer buffer;
-  auto status = block_cache_->Range(NewContext(), key, FLAGS_offset,
-                                    FLAGS_length, &buffer, option);
   if (!status.ok()) {
     LOG(ERROR) << "Range block (key=" << key.Filename()
                << ") failed: " << status.ToString();
   }
-
-  // auto hole = std::make_unique<char[]>(FLAGS_blksize);
-  // buffer.CopyTo(hole.get());
 }
 
 TaskFactoryUPtr NewFactory(BlockCacheSPtr block_cache, const std::string& op) {

src/cache/benchmark/factory.h

@@ -88,7 +88,10 @@ class RangeTaskFactory final : public TaskFactory {
   Task GenTask(const BlockKey& key) override;
 
  private:
-  void Range(const BlockKey& key);
+ private:
+  void RangeAll(const BlockKey& key);
+  void Range(const BlockKey& key, off_t offset, size_t length,
+             IOBuffer* buffer);
 
   BlockCacheSPtr block_cache_;
 };

src/cache/benchmark/worker.cpp

@@ -59,8 +59,8 @@ void Worker::ExecTask(std::function<void()> task) {
   timer.stop();
 
   collector_->Submit([this, timer](Stat* stat, Stat* total) {
-    stat->Add(FLAGS_length, timer.u_elapsed());
-    total->Add(FLAGS_length, timer.u_elapsed());
+    stat->Add(FLAGS_blksize, timer.u_elapsed());
+    total->Add(FLAGS_blksize, timer.u_elapsed());
   });
 }
 

src/cache/blockcache/block_cache_impl.cpp

@@ -66,7 +66,7 @@ BlockCacheImpl::BlockCacheImpl(BlockCacheOption option,
   if (HasCacheStore()) {
     store_ = std::make_shared<DiskCacheGroup>(option.disk_cache_options);
   } else {
-    store_ = std::make_shared<MemCache>();
+    store_ = std::make_shared<MemStore>();
   }
   uploader_ = std::make_shared<BlockCacheUploader>(store_, storage_pool_);
 }

src/cache/blockcache/cache_store.h

@@ -92,11 +92,12 @@ struct BlockKey {
 
 // block
 struct Block {
+  Block() = default;
   Block(IOBuffer buffer) : buffer(buffer), size(buffer.Size()) {}
   Block(const char* data, size_t size) : buffer(data, size), size(size) {}
 
   IOBuffer buffer;
-  size_t size;
+  size_t size{0};
 };
 
 // block context

src/cache/blockcache/lru_cache.cpp

@@ -77,6 +77,11 @@ bool LRUCache::Delete(const CacheKey& key, CacheValue* deleted) {
   return true;
 }
 
+bool LRUCache::Exist(const CacheKey& key) {
+  ListNode* node;
+  return HashLookup(key.Filename(), &node);
+}
+
 CacheItems LRUCache::Evict(FilterFunc filter) {
   CacheItems evicted;
   if (EvictNode(&inactive_, filter, &evicted)) {  // continue
@@ -116,7 +121,7 @@ CacheItem LRUCache::KV(ListNode* node) {
   CacheKey key;
   // we use CacheKey.Filename() as hash key
   auto filename = hash_->Key(node->handle);
-  CHECK(key.ParseFromFilename(filename));
+  CHECK(key.ParseFromFilename(filename)) << "filename = " << filename;
   return CacheItem(key, node->value);
 }
 

src/cache/blockcache/lru_cache.h

@@ -50,6 +50,7 @@ class LRUCache {
   virtual void Add(const CacheKey& key, const CacheValue& value);
   virtual bool Get(const CacheKey& key, CacheValue* value);
   virtual bool Delete(const CacheKey& key, CacheValue* deleted);
+  virtual bool Exist(const CacheKey& key);
   virtual CacheItems Evict(FilterFunc filter);
 
   virtual size_t Size();

src/cache/blockcache/mem_cache.h

@@ -31,10 +31,10 @@
 namespace dingofs {
 namespace cache {
 
-class MemCache final : public CacheStore {
+class MemStore final : public CacheStore {
  public:
-  MemCache() = default;
-  ~MemCache() override = default;
+  MemStore() = default;
+  ~MemStore() override = default;
 
   Status Start(UploadFunc) override { return Status::OK(); }
   Status Shutdown() override { return Status::OK(); }

src/cache/cachegroup/cache_group_node_server.cpp

@@ -27,10 +27,6 @@
 #include "cache/common/macro.h"
 #include "cache/utils/offload_thread_pool.h"
 
-namespace brpc {
-DECLARE_bool(graceful_quit_on_sigterm);
-}  // namespace brpc
-
 namespace dingofs {
 namespace cache {
 
@@ -83,7 +79,6 @@ Status CacheGroupNodeServerImpl::Start() {
 
   CHECK_RUNNING("Cache group node server");
 
-  brpc::FLAGS_graceful_quit_on_sigterm = true;
   server_->RunUntilAskedToQuit();
 
   return Status::OK();

src/cache/common/const.h

@@ -62,6 +62,7 @@ static const std::string kNodeRange = "node_range";
 static const std::string kNodeCache = "node_cache";
 static const std::string kNodePrefetch = "node_prefetch";
 static const std::string kSendResponse = "send_response";
+static const std::string kRetrieveCache = "retrieve_cache";
 
 // step: disk cache
 static const std::string kStageBlock = "stage";

src/cache/common/macro.h

@@ -25,8 +25,6 @@
 
 #include <absl/strings/str_format.h>
 
-#include "cache/storage/filesystem.h"
-
 namespace dingofs {
 namespace cache {
 
@@ -66,31 +64,31 @@ namespace cache {
 #define VLOG_6(...) VLOG(6) << absl::StrFormat(__VA_ARGS__)
 #define VLOG_9(...) VLOG(9) << absl::StrFormat(__VA_ARGS__)
 
-#define LOG_PUT_ERROR()                                          \
+#define GENERIC_LOG_PUT_ERROR()                                  \
   do {                                                           \
     LOG(ERROR) << "[" << ctx->TraceId()                          \
                << "] Put block failed: key = " << key.Filename() \
                << ", length = " << block.size                    \
                << ", status = " << status.ToString();            \
   } while (0);
 
-#define LOG_RANGE_ERROR()                                            \
+#define GENERIC_LOG_RANGE_ERROR()                                    \
   do {                                                               \
     LOG(ERROR) << "[" << ctx->TraceId()                              \
                << "] Range block failed: key = " << key.Filename()   \
                << ", offset = " << offset << ", length = " << length \
                << ", status = " << status.ToString();                \
   } while (0);
 
-#define LOG_CACHE_ERROR()                                          \
+#define GENERIC_LOG_CACHE_ERROR()                                  \
   do {                                                             \
     LOG(ERROR) << "[" << ctx->TraceId()                            \
                << "] Cache block failed: key = " << key.Filename() \
                << ", length = " << block.size                      \
                << ", status = " << status.ToString();              \
   } while (0);
 
-#define LOG_PREFETCH_ERROR()                                          \
+#define GENERIC_LOG_PREFETCH_ERROR()                                  \
   do {                                                                \
     LOG(ERROR) << "[" << ctx->TraceId()                               \
                << "] Prefetch block failed: key = " << key.Filename() \

src/cache/main.cpp

@@ -20,14 +20,6 @@
  * Author: Jingli Chen (Wine93)
  */
 
-#include <gflags/gflags.h>
-
 #include "cache/server.h"
-#include "cache/utils/logging.h"
-
-int main(int argc, char** argv) {
-  google::ParseCommandLineFlags(&argc, &argv, false);
-  dingofs::cache::InitLogging(argv[0]);
 
-  return dingofs::cache::RunServer();
-}
+int main(int argc, char** argv) { return dingofs::cache::Run(argc, argv); }