Add evict http api

Signed-off-by: JaySon-Huang <[email protected]>

Author: JaySon-Huang

dbms/src/Storages/KVStore/FFI/ProxyFFIStatusService.cpp

@@ -23,6 +23,7 @@
 #include <Storages/KVStore/FFI/ProxyFFICommon.h>
 #include <Storages/KVStore/KVStore.h>
 #include <Storages/KVStore/TMTContext.h>
+#include <Storages/S3/FileCache.h>
 #include <Storages/S3/S3GCManager.h>
 #include <TiDB/OwnerManager.h>
 #include <TiDB/Schema/SchemaSyncService.h>
@@ -32,6 +33,7 @@
 
 #include <boost/algorithm/string.hpp>
 #include <magic_enum.hpp>
+#include <string>
 
 namespace DB
 {
@@ -181,16 +183,19 @@ HttpRequestRes HandleHttpRequestStoreStatus(
 
 // Check whether the disaggregated mode is enabled.
 // If not, return a HttpRequestRes with error message.
-std::optional<HttpRequestRes> allowDisaggAPI(Context & global_ctx, std::string_view message)
+std::optional<HttpRequestRes> allowDisaggAPI(
+    Context & global_ctx,
+    DisaggregatedMode expect_mode,
+    std::string_view message)
 {
-    if (!global_ctx.getSharedContextDisagg()->isDisaggregatedStorageMode())
+    if (global_ctx.getSharedContextDisagg()->disaggregated_mode != expect_mode)
     {
         auto * body = RawCppString::New(fmt::format(
             R"json({{"message":"{}, disagg_mode={}"}})json",
             message,
             magic_enum::enum_name(global_ctx.getSharedContextDisagg()->disaggregated_mode)));
         return HttpRequestRes{
-            .status = HttpRequestStatus::ErrorParam,
+            .status = HttpRequestStatus::Ok,
             .res = CppStrWithView{
                 .inner = GenRawCppPtr(body, RawCppPtrTypeImpl::String),
                 .view = BaseBuffView{body->data(), body->size()},
@@ -209,7 +214,7 @@ HttpRequestRes HandleHttpRequestRemoteReUpload(
     std::string_view)
 {
     auto & global_ctx = server->tmt->getContext();
-    if (auto err_resp = allowDisaggAPI(global_ctx, "can not sync remote store"); err_resp)
+    if (auto err_resp = allowDisaggAPI(global_ctx, DisaggregatedMode::Storage, "can not sync remote store"); err_resp)
     {
         return err_resp.value();
     }
@@ -234,7 +239,7 @@ HttpRequestRes HandleHttpRequestRemoteOwnerInfo(
     std::string_view)
 {
     auto & global_ctx = server->tmt->getContext();
-    if (auto err_resp = allowDisaggAPI(global_ctx, "can not get gc owner"); err_resp)
+    if (auto err_resp = allowDisaggAPI(global_ctx, DisaggregatedMode::Storage, "can not get gc owner"); err_resp)
     {
         return err_resp.value();
     }
@@ -263,7 +268,7 @@ HttpRequestRes HandleHttpRequestRemoteOwnerResign(
     std::string_view)
 {
     auto & global_ctx = server->tmt->getContext();
-    if (auto err_resp = allowDisaggAPI(global_ctx, "can not resign gc owner"); err_resp)
+    if (auto err_resp = allowDisaggAPI(global_ctx, DisaggregatedMode::Storage, "can not resign gc owner"); err_resp)
     {
         return err_resp.value();
     }
@@ -290,7 +295,7 @@ HttpRequestRes HandleHttpRequestRemoteGC(
     std::string_view)
 {
     auto & global_ctx = server->tmt->getContext();
-    if (auto err_resp = allowDisaggAPI(global_ctx, "can not trigger gc"); err_resp)
+    if (auto err_resp = allowDisaggAPI(global_ctx, DisaggregatedMode::Storage, "can not trigger gc"); err_resp)
     {
         return err_resp.value();
     }
@@ -317,6 +322,79 @@ HttpRequestRes HandleHttpRequestRemoteGC(
     };
 }
 
+HttpRequestRes HandleHttpRequestRemoteCacheEvict(
+    EngineStoreServerWrap * server,
+    std::string_view path,
+    const std::string & api_name,
+    std::string_view,
+    std::string_view)
+{
+    auto & global_ctx = server->tmt->getContext();
+    if (auto err_resp = allowDisaggAPI(global_ctx, DisaggregatedMode::Compute, "can not trigger remote cache evict");
+        err_resp)
+    {
+        return err_resp.value();
+    }
+
+    auto log = Logger::get("HandleHttpRequestRemoteCacheEvict");
+    LOG_INFO(log, "handling remote cache evict request, path={} api_name={}", path, api_name);
+    FileSegment::FileType evict_until_type;
+    {
+        // schema: /tiflash/remote/cache/evict/{file_type_int}
+        auto query = path.substr(api_name.size());
+        if (query.empty() || query[0] != '/')
+        {
+            auto * body = RawCppString::New(
+                fmt::format(R"json({{"message":"invalid remote cache evict request: {}"}})json", path));
+            return HttpRequestRes{
+                .status = HttpRequestStatus::Ok,
+                .res = CppStrWithView{
+                    .inner = GenRawCppPtr(body, RawCppPtrTypeImpl::String),
+                    .view = BaseBuffView{body->data(), body->size()},
+                },
+            };
+        }
+        query.remove_prefix(1);
+        std::optional<FileSegment::FileType> opt_evict_until_type = std::nullopt;
+        try
+        {
+            auto itype = std::stoll(query.data());
+            opt_evict_until_type = magic_enum::enum_cast<FileSegment::FileType>(itype);
+        }
+        catch (...)
+        {
+            // ignore
+        }
+        if (!opt_evict_until_type.has_value())
+        {
+            auto * body = RawCppString::New(
+                fmt::format(R"json({{"message":"invalid file_type in remote cache evict request: {}"}})json", path));
+            return HttpRequestRes{
+                .status = HttpRequestStatus::Ok,
+                .res = CppStrWithView{
+                    .inner = GenRawCppPtr(body, RawCppPtrTypeImpl::String),
+                    .view = BaseBuffView{body->data(), body->size()},
+                },
+            };
+        }
+        // successfully parse the file type
+        evict_until_type = opt_evict_until_type.value();
+    }
+
+    auto released_size = DB::FileCache::instance()->evictUntil(evict_until_type);
+    auto * body = RawCppString::New(fmt::format(
+        R"json({{"file_type":"{}","released_size":"{}"}})json",
+        magic_enum::enum_name(evict_until_type),
+        released_size));
+    return HttpRequestRes{
+        .status = HttpRequestStatus::Ok,
+        .res = CppStrWithView{
+            .inner = GenRawCppPtr(body, RawCppPtrTypeImpl::String),
+            .view = BaseBuffView{body->data(), body->size()},
+        },
+    };
+}
+
 // Acquiring the all the region ids created in this TiFlash node with given keyspace id.
 HttpRequestRes HandleHttpRequestSyncRegion(
     EngineStoreServerWrap * server,
@@ -502,6 +580,7 @@ static const std::map<std::string, HANDLE_HTTP_URI_METHOD> AVAILABLE_HTTP_URI =
     {"/tiflash/remote/owner/resign", HandleHttpRequestRemoteOwnerResign},
     {"/tiflash/remote/gc", HandleHttpRequestRemoteGC},
     {"/tiflash/remote/upload", HandleHttpRequestRemoteReUpload},
+    {"/tiflash/remote/cache/evict", HandleHttpRequestRemoteCacheEvict},
 };
 
 uint8_t CheckHttpUriAvailable(BaseBuffView path_)
@@ -530,9 +609,9 @@ HttpRequestRes HandleHttpRequest(
             return method(
                 server,
                 path,
-                str,
-                std::string_view(query.data, query.len),
-                std::string_view(body.data, body.len));
+                /*api_name=*/str,
+                /*query=*/std::string_view(query.data, query.len),
+                /*body=*/std::string_view(body.data, body.len));
         }
     }
     return HttpRequestRes{

dbms/src/Storages/S3/FileCache.cpp

@@ -241,27 +241,27 @@ FileSegmentPtr FileCache::get(const S3::S3FilenameView & s3_fname, const std::op
         return nullptr;
     }
 
-    // File not exists, try to download and cache it in backgroud.
+    // File not exists, try to download and cache it in background.
 
     // We don't know the exact size of a object/file, but we need reserve space to save the object/file.
     // A certain amount of space is reserved for each file type.
-    auto estimzted_size = filesize ? *filesize : getEstimatedSizeOfFileType(file_type);
-    if (!reserveSpaceImpl(file_type, estimzted_size, EvictMode::TryEvict))
+    auto estimated_size = filesize ? *filesize : getEstimatedSizeOfFileType(file_type);
+    if (!reserveSpaceImpl(file_type, estimated_size, EvictMode::TryEvict))
     {
         // Space not enough.
         GET_METRIC(tiflash_storage_remote_cache, type_dtfile_full).Increment();
         LOG_DEBUG(
             log,
-            "s3_key={} space not enough(capacity={} used={} estimzted_size={}), skip cache",
+            "s3_key={} space not enough(capacity={} used={} estimated_size={}), skip cache",
             s3_key,
             cache_capacity,
             cache_used,
-            estimzted_size);
+            estimated_size);
         return nullptr;
     }
 
     auto file_seg
-        = std::make_shared<FileSegment>(toLocalFilename(s3_key), FileSegment::Status::Empty, estimzted_size, file_type);
+        = std::make_shared<FileSegment>(toLocalFilename(s3_key), FileSegment::Status::Empty, estimated_size, file_type);
     table.set(s3_key, file_seg);
     bgDownload(s3_key, file_seg);
 
@@ -300,7 +300,7 @@ FileSegmentPtr FileCache::getOrWait(const S3::S3FilenameView & s3_fname, const s
         GET_METRIC(tiflash_storage_remote_cache, type_dtfile_full).Increment();
         LOG_INFO(
             log,
-            "s3_key={} space not enough(capacity={} used={} estimzted_size={}), skip cache",
+            "s3_key={} space not enough(capacity={} used={} estimated_size={}), skip cache",
             s3_key,
             cache_capacity,
             cache_used,
@@ -378,7 +378,7 @@ std::pair<Int64, std::list<String>::iterator> FileCache::removeImpl(
     FileSegmentPtr & f,
     bool force)
 {
-    // Except currenly thread and the FileTable,
+    // Except current thread and the FileTable,
     // there are other threads hold this FileSegment object.
     if (f.use_count() > 2 && !force)
     {
@@ -422,30 +422,50 @@ bool FileCache::reserveSpaceImpl(FileType reserve_for, UInt64 size, EvictMode ev
 
 // The basic evict logic:
 // Distinguish cache priority according to file type. The larger the file type, the lower the priority.
+// If evict_same_type_first is true,
 // First, try to evict files which not be used recently with the same type. => Try to evict old files.
 // Second, try to evict files with lower priority. => Try to evict lower priority files.
 // Finally, evict files with higher priority, if space is still not sufficient. Higher priority files
 // are usually smaller. If we don't evict them, it is very possible that cache is full of these higher
 // priority small files and we can't effectively cache any lower-priority large files.
-std::vector<FileType> FileCache::getEvictFileTypes(FileType evict_for)
+std::vector<FileType> FileCache::getEvictFileTypes(FileType evict_for, bool evict_same_type_first)
 {
-    std::vector<FileType> evict_types;
-    evict_types.push_back(evict_for); // First, try evict with the same file type.
     constexpr auto all_file_types = magic_enum::enum_values<FileType>(); // all_file_types are sorted by enum value.
-    // Second, try evict from the lower proirity file type.
-    for (auto itr = std::rbegin(all_file_types); itr != std::rend(all_file_types); ++itr)
+    if (evict_same_type_first)
+    {
+        std::vector<FileType> evict_types;
+        evict_types.push_back(evict_for); // First, try evict with the same file type.
+        // Second, try evict from the lower priority file type.
+        for (auto itr = std::rbegin(all_file_types); itr != std::rend(all_file_types); ++itr)
+        {
+            if (*itr != evict_for)
+                evict_types.push_back(*itr);
+        }
+        return evict_types;
+    }
+    else
     {
-        if (*itr != evict_for)
+        std::vector<FileType> evict_types;
+        // Evict from the lower priority file type first.
+        for (auto itr = std::rbegin(all_file_types); itr != std::rend(all_file_types); ++itr)
+        {
             evict_types.push_back(*itr);
+            if (*itr == evict_for)
+            {
+                // Do not evict higher priority file type
+                break;
+                ;
+            }
+        }
+        return evict_types;
     }
-    return evict_types;
 }
 
 void FileCache::tryEvictFile(FileType evict_for, UInt64 size, EvictMode evict)
 {
     RUNTIME_CHECK(evict != EvictMode::NoEvict);
 
-    auto file_types = getEvictFileTypes(evict_for);
+    auto file_types = getEvictFileTypes(evict_for, /*evict_same_type_first*/ true);
     for (auto evict_from : file_types)
     {
         auto evicted_size = tryEvictFrom(evict_for, size, evict_from);
@@ -1015,4 +1035,45 @@ void FileCache::updateConfig(const Settings & settings)
     }
 }
 
+UInt64 FileCache::evictUntil(FileSegment::FileType file_type)
+{
+    std::lock_guard lock(mtx);
+    auto file_types = getEvictFileTypes(file_type, /*evict_same_type_first*/ false);
+    UInt64 total_released_size = 0;
+    for (auto evict_from : file_types)
+    {
+        UInt64 curr_released_size = 0;
+        auto & table = tables[static_cast<UInt64>(evict_from)];
+        for (auto itr = table.begin(); itr != table.end();)
+        {
+            auto s3_key = *itr;
+            auto f = table.get(s3_key, /*update_lru*/ false);
+            auto [released_size, next_itr] = removeImpl(table, s3_key, f, /*force*/ true);
+            if (released_size < 0) // not remove
+            {
+                ++itr;
+            }
+            else // removed
+            {
+                itr = next_itr;
+                curr_released_size += released_size;
+            }
+        }
+        total_released_size += curr_released_size;
+        LOG_INFO(
+            log,
+            "evictUntil layer evict finish, evict_from={} file_type={} released_size={} tot_release_size={}",
+            magic_enum::enum_name(evict_from),
+            magic_enum::enum_name(file_type),
+            curr_released_size,
+            total_released_size);
+    }
+    LOG_INFO(
+        log,
+        "evictUntil finish, file_type={} total_released_size={}",
+        magic_enum::enum_name(file_type),
+        total_released_size);
+    return total_released_size;
+}
+
 } // namespace DB

dbms/src/Storages/S3/FileCache.h

@@ -261,6 +261,9 @@ class FileCache
 
     void updateConfig(const Settings & settings);
 
+    // evict the cached files until no file of >= `file_type` is in cache.
+    UInt64 evictUntil(FileSegment::FileType file_type);
+
 #ifndef DBMS_PUBLIC_GTEST
 private:
 #else
@@ -346,7 +349,9 @@ class FileCache
     void releaseSpace(UInt64 size);
     bool reserveSpace(FileSegment::FileType reserve_for, UInt64 size, EvictMode evict);
     bool finalizeReservedSize(FileSegment::FileType reserve_for, UInt64 reserved_size, UInt64 content_length);
-    static std::vector<FileSegment::FileType> getEvictFileTypes(FileSegment::FileType evict_for);
+    static std::vector<FileSegment::FileType> getEvictFileTypes(
+        FileSegment::FileType evict_for,
+        bool evict_same_type_first);
     void tryEvictFile(FileSegment::FileType evict_for, UInt64 size, EvictMode evict);
     UInt64 tryEvictFrom(FileSegment::FileType evict_for, UInt64 size, FileSegment::FileType evict_from);
     UInt64 forceEvict(UInt64 size);

dbms/src/Storages/StorageDisaggregatedRemote.cpp

@@ -209,6 +209,7 @@ DM::SegmentReadTasks StorageDisaggregated::buildReadTaskWithBackoff(
         }
     }
 
+    LOG_INFO(log, "build read task done, num_read_tasks={} elapsed_s={:.3f}", read_task.size(), build_read_task_watch.elapsedSeconds());
     return read_task;
 }
 
@@ -400,6 +401,7 @@ void StorageDisaggregated::buildReadTaskForWriteNode(
         elapsed_seconds = watch.elapsedSeconds();
         scan_context->disagg_parse_read_task_ms += elapsed_seconds * 1000;
         GET_METRIC(tiflash_disaggregated_breakdown_duration_seconds, type_parse_read_tasks).Observe(elapsed_seconds);
+        LOG_INFO(log, "build read task for write node done, wn_addr={} elapsed_s={:.3f}", req->address(), elapsed_seconds);
     });
     // Now we have successfully established disaggregated read for this write node.
     // Let's parse the result and generate actual segment read tasks.