Binary cache: async push_success

include/vcpkg/base/batch-queue.h

@@ -0,0 +1,61 @@
+#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() const { 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

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

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

@@ -2888,11 +2888,6 @@ DECLARE_MESSAGE(
     (),
     "",
     "If you are sure you want to rebuild the above packages, run this command with the --no-dry-run option.")
-DECLARE_MESSAGE(UploadedBinaries, (msg::count, msg::vendor), "", "Uploaded binaries to {count} {vendor}.")
-DECLARE_MESSAGE(UploadedPackagesToVendor,
-                (msg::count, msg::elapsed, msg::vendor),
-                "",
-                "Uploaded {count} package(s) to {vendor} in {elapsed}")
 DECLARE_MESSAGE(UploadingBinariesToVendor,
                 (msg::spec, msg::vendor, msg::path),
                 "",
@@ -3128,6 +3123,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(WarningMessage, (), "", "warning: ")
 DECLARE_MESSAGE(WarningMessageMustUsePrintWarning,
                 (msg::value),

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

@@ -159,6 +159,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

@@ -3,12 +3,15 @@
 #include <vcpkg/base/fwd/message_sinks.h>
 
 #include <vcpkg/fwd/binarycaching.h>
+#include <vcpkg/fwd/build.h>
 #include <vcpkg/fwd/dependencies.h>
 #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>
@@ -18,6 +21,7 @@
 #include <iterator>
 #include <set>
 #include <string>
+#include <thread>
 #include <unordered_map>
 #include <vector>
 
@@ -196,23 +200,41 @@ namespace vcpkg
 
     struct BinaryCache : ReadOnlyBinaryCache
     {
-        static ExpectedL<BinaryCache> make(const VcpkgCmdArguments& args, const VcpkgPaths& paths, MessageSink& sink);
+        static ExpectedL<std::unique_ptr<BinaryCache>> make(const VcpkgCmdArguments& args,
+                                                            const VcpkgPaths& paths,
+                                                            MessageSink& sink);
 
         BinaryCache(const Filesystem& fs);
         BinaryCache(const BinaryCache&) = delete;
-        BinaryCache(BinaryCache&&) = default;
+        BinaryCache(BinaryCache&&) = delete;
         ~BinaryCache();
 
         /// Called upon a successful build of `action` to store those contents in the binary cache.
         void push_success(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;
+        };
+
+        void push_thread_main();
+
+        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;
     };
 
     ExpectedL<DownloadManagerConfig> parse_download_configuration(const Optional<std::string>& arg);

locales/messages.json

@@ -1613,10 +1613,6 @@
   "UpgradeInManifest": "Upgrade upgrades a classic mode installation and thus does not support manifest mode. Consider updating your dependencies by updating your baseline to a current value with vcpkg x-update-baseline and running vcpkg install.",
   "_UpgradeInManifest.comment": "'vcpkg x-update-baseline' and 'vcpkg install' are command lines and should not be localized.",
   "UpgradeRunWithNoDryRun": "If you are sure you want to rebuild the above packages, run this command with the --no-dry-run option.",
-  "UploadedBinaries": "Uploaded binaries to {count} {vendor}.",
-  "_UploadedBinaries.comment": "An example of {count} is 42. An example of {vendor} is Azure.",
-  "UploadedPackagesToVendor": "Uploaded {count} package(s) to {vendor} in {elapsed}",
-  "_UploadedPackagesToVendor.comment": "An example of {count} is 42. An example of {elapsed} is 3.532 min. An example of {vendor} is Azure.",
   "UploadingBinariesToVendor": "Uploading binaries for '{spec}' to '{vendor}' source \"{path}\".",
   "_UploadingBinariesToVendor.comment": "An example of {spec} is zlib:x64-windows. An example of {vendor} is Azure. An example of {path} is /foo/bar.",
   "UploadingBinariesUsingVendor": "Uploading binaries for '{spec}' using '{vendor}' \"{path}\".",
@@ -1718,6 +1714,8 @@
   "VersionTableHeader": "Version",
   "VersionVerifiedOK": "OK: {version_spec} -> {commit_sha}",
   "_VersionVerifiedOK.comment": "An example of {version_spec} is zlib:[email protected]. An example of {commit_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 vcpkg::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

@@ -323,6 +323,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();