Binary cache: async push_success

include/vcpkg/base/batch-queue.h

@@ -0,0 +1,65 @@
+#pragma once
+
+#include <condition_variable>
+#include <mutex>
+#include <vector>
+
+template<class T>
+class BatchQueue
+{
+public:
+    template<class... Args>
+    void push(Args&&... args)
+    {
+        forward.emplace_back(std::forward<Args>(args)...);
+    }
+
+    bool empty() const { return forward.empty(); }
+
+    void pop(std::vector<T>& out)
+    {
+        out.clear();
+        swap(out, forward);
+    }
+
+private:
+    std::vector<T> forward;
+};
+
+template<class WorkItem>
+struct BGThreadBatchQueue
+{
+    template<class... Args>
+    void push(Args&&... args)
+    {
+        std::lock_guard<std::mutex> lock(m_mtx);
+        m_tasks.push(std::forward<Args>(args)...);
+        m_cv.notify_all();
+    }
+
+    void wait_for_items(std::vector<WorkItem>& out)
+    {
+        std::unique_lock<std::mutex> lock(m_mtx);
+        m_cv.wait(lock, [this]() { return !m_tasks.empty() || !m_running; });
+        m_tasks.pop(out);
+    }
+
+    void stop()
+    {
+        std::lock_guard<std::mutex> lock(m_mtx);
+        m_running = false;
+        m_cv.notify_all();
+    }
+
+    bool stopped()
+    {
+        std::lock_guard<std::mutex> lock(m_mtx);
+        return !m_running;
+    }
+
+private:
+    std::mutex m_mtx;
+    std::condition_variable m_cv;
+    BatchQueue<WorkItem> m_tasks;
+    bool m_running = true;
+};

include/vcpkg/base/fwd/message_sinks.h

@@ -11,4 +11,5 @@ namespace vcpkg
 
     struct FileSink;
     struct CombiningSink;
+    struct BGMessageSink;
 }

include/vcpkg/base/message-data.inc.h

@@ -3200,6 +3200,7 @@ DECLARE_MESSAGE(VSExaminedPaths, (), "", "The following paths were examined for
 DECLARE_MESSAGE(VSNoInstances, (), "", "Could not locate a complete Visual Studio instance")
 DECLARE_MESSAGE(WaitingForChildrenToExit, (), "", "Waiting for child processes to exit...")
 DECLARE_MESSAGE(WaitingToTakeFilesystemLock, (msg::path), "", "waiting to take filesystem lock on {path}...")
+DECLARE_MESSAGE(WaitUntilPackagesUploaded, (msg::count), "", "Wait until the remaining packages ({count}) are uploaded")
 DECLARE_MESSAGE(WarningsTreatedAsErrors, (), "", "previous warnings being interpreted as errors")
 DECLARE_MESSAGE(WarnOnParseConfig, (msg::path), "", "Found the following warnings in configuration {path}:")
 DECLARE_MESSAGE(WhileCheckingOutBaseline, (msg::commit_sha), "", "while checking out baseline {commit_sha}")

include/vcpkg/base/message_sinks.h

@@ -4,6 +4,8 @@
 
 #include <vcpkg/base/messages.h>
 
+#include <mutex>
+
 namespace vcpkg
 {
 
@@ -75,4 +77,30 @@ namespace vcpkg
         CombiningSink(MessageSink& first, MessageSink& second) : m_first(first), m_second(second) { }
         void print(Color c, StringView sv) override;
     };
+
+    struct BGMessageSink final : MessageSink
+    {
+        BGMessageSink(MessageSink& out_sink) : out_sink(out_sink) { }
+        ~BGMessageSink() { publish_directly_to_out_sink(); }
+        // must be called from producer
+        void print(Color c, StringView sv) override;
+        using MessageSink::print;
+
+        // must be called from consumer (synchronizer of out)
+        void print_published();
+
+        void publish_directly_to_out_sink();
+
+    private:
+        MessageSink& out_sink;
+
+        std::mutex m_published_lock;
+        std::vector<std::pair<Color, std::string>> m_published;
+
+        // buffers messages until newline is reached
+        // guarded by m_print_directly_lock
+        std::vector<std::pair<Color, std::string>> m_unpublished;
+        std::mutex m_print_directly_lock;
+        bool m_print_directly_to_out_sink = false;
+    };
 }

include/vcpkg/base/strings.h

@@ -198,6 +198,8 @@ namespace vcpkg::Strings
 
     const char* find_first_of(StringView searched, StringView candidates);
 
+    [[nodiscard]] std::string::size_type find_last(StringView searched, char c);
+
     [[nodiscard]] std::vector<StringView> find_all_enclosed(StringView input,
                                                             StringView left_delim,
                                                             StringView right_delim);

include/vcpkg/binarycaching.h

@@ -8,8 +8,10 @@
 #include <vcpkg/fwd/tools.h>
 #include <vcpkg/fwd/vcpkgpaths.h>
 
+#include <vcpkg/base/batch-queue.h>
 #include <vcpkg/base/downloads.h>
 #include <vcpkg/base/expected.h>
+#include <vcpkg/base/message_sinks.h>
 #include <vcpkg/base/path.h>
 
 #include <vcpkg/archives.h>
@@ -20,6 +22,7 @@
 #include <iterator>
 #include <set>
 #include <string>
+#include <thread>
 #include <unordered_map>
 #include <vector>
 
@@ -200,29 +203,63 @@ namespace vcpkg
         std::vector<CacheAvailability> precheck(View<InstallPlanAction> actions);
 
     protected:
-        BinaryProviders m_config;
+        struct ReadOnlyBinaryCacheData
+        {
+            BinaryProviders m_config;
 
-        std::unordered_map<std::string, CacheStatus> m_status;
+            std::unordered_map<std::string, CacheStatus> m_status;
+        };
+        std::unique_ptr<ReadOnlyBinaryCacheData> data = std::make_unique<ReadOnlyBinaryCacheData>();
     };
 
+    // compression and upload of binary cache entries happens on a single 'background' thread, `m_push_thread`
+    // Thread safety is achieved within the binary cache providers by:
+    //   1. Only using one thread in the background for this work.
+    //   2. Forming a queue of work for that thread to consume in `m_actions_to_push`, which maintains its own thread
+    //   safety
+    //   3. Sending any replies from the background thread through `m_bg_msg_sink`
+    //   4. Ensuring any supporting data, such as tool exes, is provided before the background thread is started.
+    //   5. Ensuring that work is not submitted to the background thread until the corresponding `packages` directory to
+    //   upload is no longer being actively touched by the foreground thread.
     struct BinaryCache : ReadOnlyBinaryCache
     {
         static ExpectedL<BinaryCache> make(const VcpkgCmdArguments& args, const VcpkgPaths& paths, MessageSink& sink);
 
         BinaryCache(const Filesystem& fs);
         BinaryCache(const BinaryCache&) = delete;
         BinaryCache(BinaryCache&&) = default;
+        ~BinaryCache();
 
         /// Called upon a successful build of `action` to store those contents in the binary cache.
         void push_success(CleanPackages clean_packages, const InstallPlanAction& action);
 
+        void print_push_success_messages();
+        void wait_for_async_complete_and_join();
+
     private:
         BinaryCache(BinaryProviders&& providers, const Filesystem& fs);
-
-        const Filesystem& m_fs;
-        Optional<ZipTool> m_zip_tool;
-        bool m_needs_nuspec_data = false;
-        bool m_needs_zip_file = false;
+        struct ActionToPush
+        {
+            BinaryPackageWriteInfo request;
+            CleanPackages clean_after_push;
+        };
+        struct BinaryCacheData
+        {
+            BinaryCacheData(const Filesystem& fs, ReadOnlyBinaryCacheData* data);
+            const Filesystem& m_fs;
+            Optional<ZipTool> m_zip_tool;
+            bool m_needs_nuspec_data = false;
+            bool m_needs_zip_file = false;
+
+            BGMessageSink m_bg_msg_sink;
+            BGThreadBatchQueue<ActionToPush> m_actions_to_push;
+            std::atomic_int m_remaining_packages_to_push = 0;
+            std::thread m_push_thread;
+
+            void push_thread_main(ReadOnlyBinaryCacheData* ro_data);
+            void wait_for_async_complete_and_join();
+        };
+        std::unique_ptr<BinaryCacheData> bc_data;
     };
 
     ExpectedL<DownloadManagerConfig> parse_download_configuration(const Optional<std::string>& arg);

locales/messages.json

@@ -1678,6 +1678,8 @@
   "_VersionSpecMismatch.comment": "An example of {path} is /foo/bar. An example of {expected_version} is 1.3.8. An example of {actual_version} is 1.3.8.",
   "VersionVerifiedOK": "{version_spec} is correctly in the version database ({git_tree_sha})",
   "_VersionVerifiedOK.comment": "An example of {version_spec} is zlib:[email protected]. An example of {git_tree_sha} is 7cfad47ae9f68b183983090afd6337cd60fd4949.",
+  "WaitUntilPackagesUploaded": "Wait until the remaining packages ({count}) are uploaded",
+  "_WaitUntilPackagesUploaded.comment": "An example of {count} is 42.",
   "WaitingForChildrenToExit": "Waiting for child processes to exit...",
   "WaitingToTakeFilesystemLock": "waiting to take filesystem lock on {path}...",
   "_WaitingToTakeFilesystemLock.comment": "An example of {path} is /foo/bar.",

src/vcpkg-test/strings.cpp

@@ -69,6 +69,14 @@ TEST_CASE ("find_first_of", "[strings]")
     REQUIRE(find_first_of("abcdefg", "gb") == std::string("bcdefg"));
 }
 
+TEST_CASE ("find_last", "[strings]")
+{
+    using vcpkg::Strings::find_last;
+    REQUIRE(find_last("abcdefg", 'a') == 0);
+    REQUIRE(find_last("abcdefg", 'g') == 6);
+    REQUIRE(find_last("abcdefg", 'z') == std::string::npos);
+}
+
 TEST_CASE ("contains_any_ignoring_c_comments", "[strings]")
 {
     using Strings::contains_any_ignoring_c_comments;

src/vcpkg/base/message_sinks.cpp

@@ -1,5 +1,6 @@
 #include <vcpkg/base/file_sink.h>
 #include <vcpkg/base/message_sinks.h>
+#include <vcpkg/base/strings.h>
 
 namespace
 {
@@ -58,4 +59,74 @@ namespace vcpkg
         m_second.print(c, sv);
     }
 
+    void BGMessageSink::print(Color c, StringView sv)
+    {
+        std::lock_guard<std::mutex> print_lk(m_print_directly_lock);
+        if (m_print_directly_to_out_sink)
+        {
+            out_sink.print(c, sv);
+            return;
+        }
+
+        auto pos = Strings::find_last(sv, '\n');
+        if (pos != std::string::npos)
+        {
+            {
+                std::lock_guard<std::mutex> lk(m_published_lock);
+                m_published.insert(m_published.end(),
+                                   std::make_move_iterator(m_unpublished.begin()),
+                                   std::make_move_iterator(m_unpublished.end()));
+                m_published.emplace_back(c, sv.substr(0, pos + 1));
+            }
+            m_unpublished.clear();
+            if (sv.size() > pos + 1)
+            {
+                m_unpublished.emplace_back(c, sv.substr(pos + 1));
+            }
+        }
+        else
+        {
+            m_unpublished.emplace_back(c, sv);
+        }
+    }
+
+    void BGMessageSink::print_published()
+    {
+        std::vector<std::pair<Color, std::string>> tmp;
+        for (;;)
+        {
+            {
+                std::lock_guard<std::mutex> lk(m_published_lock);
+                swap(tmp, m_published);
+            }
+
+            if (tmp.empty())
+            {
+                return;
+            }
+
+            for (auto&& msg : tmp)
+            {
+                out_sink.print(msg.first, msg.second);
+            }
+
+            tmp.clear();
+        }
+    }
+
+    void BGMessageSink::publish_directly_to_out_sink()
+    {
+        std::lock_guard<std::mutex> print_lk(m_print_directly_lock);
+        std::lock_guard<std::mutex> lk(m_published_lock);
+
+        m_print_directly_to_out_sink = true;
+        for (auto& messages : {&m_published, &m_unpublished})
+        {
+            for (auto&& msg : *messages)
+            {
+                out_sink.print(msg.first, msg.second);
+            }
+            messages->clear();
+        }
+    }
 }

src/vcpkg/base/strings.cpp

@@ -330,6 +330,12 @@ const char* Strings::find_first_of(StringView input, StringView chars)
     return std::find_first_of(input.begin(), input.end(), chars.begin(), chars.end());
 }
 
+std::string::size_type Strings::find_last(StringView searched, char c)
+{
+    auto iter = std::find(searched.rbegin(), searched.rend(), c);
+    return iter == searched.rend() ? std::string::npos : (&*iter - searched.begin());
+}
+
 std::vector<StringView> Strings::find_all_enclosed(StringView input, StringView left_delim, StringView right_delim)
 {
     auto it_left = input.begin();