fix(http): harden FileCache concurrent access

http/server/FileCache.cpp

@@ -38,9 +38,13 @@ FileCache::FileCache(size_t capacity) : hv::LRUCache<std::string, file_cache_ptr
 }
 
 file_cache_ptr FileCache::Open(const char* filepath, OpenParam* param) {
+    BeginPathAccess(filepath);
+    defer(EndPathAccess(filepath);)
+
     file_cache_ptr fc = Get(filepath);
     bool modified = false;
     if (fc) {
+        std::lock_guard<std::mutex> lock(fc->mutex);
         time_t now = time(NULL);
         if (now - fc->stat_time > stat_interval) {
             fc->stat_time = now;
@@ -65,62 +69,63 @@ file_cache_ptr FileCache::Open(const char* filepath, OpenParam* param) {
             return NULL;
         }
         defer(if (fd > 0) { close(fd); })
-        if (fc == NULL) {
-            if (S_ISREG(st.st_mode) ||
-                (S_ISDIR(st.st_mode) &&
-                 filepath[strlen(filepath)-1] == '/')) {
-                fc = std::make_shared<file_cache_t>();
-                fc->filepath = filepath;
-                fc->st = st;
-                time(&fc->open_time);
-                fc->stat_time = fc->open_time;
-                fc->stat_cnt = 1;
-                put(filepath, fc);
-            }
-            else {
-                param->error = ERR_MISMATCH;
-                return NULL;
-            }
+        size_t filepath_len = strlen(filepath);
+        if (!S_ISREG(st.st_mode) &&
+            !(S_ISDIR(st.st_mode) && filepath_len > 0 && filepath[filepath_len - 1] == '/')) {
+            param->error = ERR_MISMATCH;
+            return NULL;
         }
-        if (S_ISREG(fc->st.st_mode)) {
-            param->filesize = fc->st.st_size;
+
+        // Build a complete replacement before publishing it to the LRU. Existing
+        // readers keep their immutable shared_ptr while a refresh is in progress.
+        file_cache_ptr refreshed = std::make_shared<file_cache_t>();
+        refreshed->filepath = filepath;
+        refreshed->st = st;
+        time(&refreshed->open_time);
+        refreshed->stat_time = refreshed->open_time;
+        refreshed->stat_cnt = 1;
+
+        if (S_ISREG(st.st_mode)) {
+            param->filesize = st.st_size;
             // FILE
             if (param->need_read) {
-                if (fc->st.st_size > param->max_read) {
+                if (st.st_size > param->max_read) {
                     param->error = ERR_OVER_LIMIT;
                     return NULL;
                 }
-                fc->resize_buf(fc->st.st_size);
-                int nread = read(fd, fc->filebuf.base, fc->filebuf.len);
-                if (nread != fc->filebuf.len) {
+                refreshed->resize_buf(st.st_size);
+                int nread = read(fd, refreshed->filebuf.base, refreshed->filebuf.len);
+                if (nread != refreshed->filebuf.len) {
                     hloge("Failed to read file: %s", filepath);
                     param->error = ERR_READ_FILE;
-                    Close(filepath);
                     return NULL;
                 }
             }
             const char* suffix = strrchr(filepath, '.');
             if (suffix) {
                 http_content_type content_type = http_content_type_enum_by_suffix(suffix+1);
                 if (content_type == TEXT_HTML) {
-                    fc->content_type = "text/html; charset=utf-8";
+                    refreshed->content_type = "text/html; charset=utf-8";
                 } else if (content_type == TEXT_PLAIN) {
-                    fc->content_type = "text/plain; charset=utf-8";
+                    refreshed->content_type = "text/plain; charset=utf-8";
                 } else {
-                    fc->content_type = http_content_type_str_by_suffix(suffix+1);
+                    refreshed->content_type = http_content_type_str_by_suffix(suffix+1);
                 }
             }
         }
-        else if (S_ISDIR(fc->st.st_mode)) {
+        else if (S_ISDIR(st.st_mode)) {
             // DIR
             std::string page;
             make_index_of_page(filepath, page, param->path);
-            fc->resize_buf(page.size());
-            memcpy(fc->filebuf.base, page.c_str(), page.size());
-            fc->content_type = "text/html; charset=utf-8";
+            refreshed->resize_buf(page.size());
+            memcpy(refreshed->filebuf.base, page.c_str(), page.size());
+            refreshed->content_type = "text/html; charset=utf-8";
         }
-        gmtime_fmt(fc->st.st_mtime, fc->last_modified);
-        snprintf(fc->etag, sizeof(fc->etag), ETAG_FMT, (size_t)fc->st.st_mtime, (size_t)fc->st.st_size);
+        gmtime_fmt(refreshed->st.st_mtime, refreshed->last_modified);
+        snprintf(refreshed->etag, sizeof(refreshed->etag), ETAG_FMT,
+                 (size_t)refreshed->st.st_mtime, (size_t)refreshed->st.st_size);
+        put(filepath, refreshed);
+        fc = refreshed;
     }
     return fc;
 }
@@ -141,9 +146,27 @@ file_cache_ptr FileCache::Get(const char* filepath) {
     return NULL;
 }
 
+void FileCache::BeginPathAccess(const std::string& filepath) {
+    std::unique_lock<std::mutex> lock(loading_mutex_);
+    loading_cv_.wait(lock, [this, &filepath]() {
+        return loading_files_.find(filepath) == loading_files_.end();
+    });
+    loading_files_.insert(filepath);
+}
+
+void FileCache::EndPathAccess(const std::string& filepath) {
+    {
+        std::lock_guard<std::mutex> lock(loading_mutex_);
+        loading_files_.erase(filepath);
+    }
+    loading_cv_.notify_all();
+}
+
 void FileCache::RemoveExpiredFileCache() {
     time_t now = time(NULL);
     remove_if([this, now](const std::string& filepath, const file_cache_ptr& fc) {
+        std::unique_lock<std::mutex> lock(fc->mutex, std::try_to_lock);
+        if (!lock.owns_lock()) return false;
         return (now - fc->stat_time > expired_time);
     });
 }

http/server/FileCache.h

@@ -5,6 +5,8 @@
 #include <map>
 #include <string>
 #include <mutex>
+#include <condition_variable>
+#include <set>
 
 #include "hbuf.h"
 #include "hstring.h"
@@ -15,6 +17,7 @@
 #define FILE_CACHE_MAX_SIZE         (1 << 22)   // 4M
 
 typedef struct file_cache_s {
+    mutable std::mutex mutex;
     std::string filepath;
     struct stat st;
     time_t      open_time;
@@ -31,7 +34,7 @@ typedef struct file_cache_s {
         stat_cnt = 0;
     }
 
-    bool is_modified() {
+    bool is_modified() const {
         struct stat new_st;
         // stat failed: treat as modified to force re-validation via reopen
 #ifdef OS_WIN
@@ -41,9 +44,9 @@ typedef struct file_cache_s {
         if (stat(filepath.c_str(), &new_st) != 0)
             return true;
 #endif
-        time_t mtime = st.st_mtime;
-        st = new_st;
-        return mtime != st.st_mtime;
+        return st.st_mtime != new_st.st_mtime ||
+               st.st_size != new_st.st_size ||
+               st.st_mode != new_st.st_mode;
     }
 
     bool is_complete() {
@@ -101,6 +104,14 @@ class FileCache : public hv::LRUCache<std::string, file_cache_ptr> {
 
 protected:
     file_cache_ptr Get(const char* filepath);
+
+private:
+    void BeginPathAccess(const std::string& filepath);
+    void EndPathAccess(const std::string& filepath);
+
+    std::mutex loading_mutex_;
+    std::condition_variable loading_cv_;
+    std::set<std::string> loading_files_;
 };
 
 #endif // HV_FILE_CACHE_H_

http/server/HttpHandler.cpp

@@ -957,6 +957,13 @@ int HttpHandler::SendHttpResponse(bool submit) {
     if (!io || !parser) return -1;
     char* data = NULL;
     size_t len = 0, total_len = 0;
+    // Keep the cache entry alive until the mutex is released. GetSendData may
+    // evict the entry after its buffer has been handed to the writer.
+    file_cache_ptr fc_keepalive = fc;
+    std::unique_lock<std::mutex> file_cache_lock;
+    if (fc_keepalive) {
+        file_cache_lock = std::unique_lock<std::mutex>(fc_keepalive->mutex);
+    }
     if (submit) parser->SubmitResponse(resp.get());
     while (GetSendData(&data, &len)) {
         // printf("GetSendData %d\n", (int)len);

scripts/unittest.sh

@@ -35,6 +35,9 @@ bin/http_router_test
 if [ -x bin/websocket_parser_test ]; then
     bin/websocket_parser_test || exit $?
 fi
+if [ -x bin/file_cache_concurrency_test ]; then
+    bin/file_cache_concurrency_test || exit $?
+fi
 if [ -x bin/tlv_test ]; then
     bin/tlv_test
 fi

unittest/CMakeLists.txt

@@ -94,7 +94,10 @@ if(WITH_HTTP)
 add_executable(websocket_parser_test websocket_parser_test.cpp)
 target_include_directories(websocket_parser_test PRIVATE .. ../base ../http)
 target_link_libraries(websocket_parser_test ${HV_LIBRARIES})
-set(HTTP_UNITTEST_TARGETS websocket_parser_test)
+add_executable(file_cache_concurrency_test file_cache_concurrency_test.cpp)
+target_include_directories(file_cache_concurrency_test PRIVATE .. ../base ../cpputil ../http ../http/server)
+target_link_libraries(file_cache_concurrency_test ${HV_LIBRARIES})
+set(HTTP_UNITTEST_TARGETS websocket_parser_test file_cache_concurrency_test)
 endif()
 
 if(WITH_LUA)

unittest/file_cache_concurrency_test.cpp

@@ -0,0 +1,55 @@
+#include "FileCache.h"
+
+#include <condition_variable>
+#include <cstdio>
+#include <fstream>
+#include <mutex>
+#include <thread>
+#include <unistd.h>
+#include <vector>
+
+int main() {
+    const std::string path = "/tmp/hv-file-cache-concurrency-" + std::to_string(getpid()) + ".txt";
+    {
+        std::ofstream out(path.c_str(), std::ios::binary);
+        out << "file-cache-concurrency";
+    }
+
+    FileCache cache;
+    const size_t thread_count = 8;
+    std::vector<file_cache_ptr> results(thread_count);
+    std::vector<std::thread> workers;
+    std::mutex start_mutex;
+    std::condition_variable start_cv;
+    size_t ready = 0;
+    bool start = false;
+
+    for (size_t i = 0; i < thread_count; ++i) {
+        workers.emplace_back([&, i]() {
+            {
+                std::unique_lock<std::mutex> lock(start_mutex);
+                ++ready;
+                start_cv.notify_all();
+                start_cv.wait(lock, [&]() { return start; });
+            }
+            FileCache::OpenParam param;
+            results[i] = cache.Open(path.c_str(), &param);
+        });
+    }
+
+    {
+        std::unique_lock<std::mutex> lock(start_mutex);
+        start_cv.wait(lock, [&]() { return ready == thread_count; });
+        start = true;
+    }
+    start_cv.notify_all();
+    for (auto& worker : workers) worker.join();
+
+    bool ok = results[0] != NULL;
+    for (size_t i = 1; i < results.size(); ++i) {
+        ok = ok && results[i] == results[0];
+    }
+
+    std::remove(path.c_str());
+    return ok ? 0 : 1;
+}