Binary cache: async push_success

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

@@ -2276,6 +2276,7 @@ DECLARE_MESSAGE(SpecifyTargetArch,
                 "Specify the target architecture triplet. See 'vcpkg help triplet'.\n(default: '{env_var}')")
 DECLARE_MESSAGE(StartCodeUnitInContinue, (), "", "found start code unit in continue position")
 DECLARE_MESSAGE(StoredBinaryCache, (msg::path), "", "Stored binary cache: \"{path}\"")
+DECLARE_MESSAGE(StoredBinariesToDestinations, (msg::count), "", "Stored binaries in {count} destinations.")
 DECLARE_MESSAGE(StoreOptionMissingSha, (), "", "--store option is invalid without a sha512")
 DECLARE_MESSAGE(SuccessfulyExported, (msg::package_name, msg::path), "", "Exported {package_name} to {path}")
 DECLARE_MESSAGE(SuggestGitPull, (), "", "The result may be outdated. Run `git pull` to get the latest results.")
@@ -2527,11 +2528,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),
                 "",
@@ -2765,6 +2761,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

@@ -5,6 +5,8 @@
 #include <vcpkg/base/files.h>
 #include <vcpkg/base/messages.h>
 
+#include <mutex>
+
 namespace vcpkg
 {
 
@@ -87,4 +89,31 @@ namespace vcpkg
         CombiningSink(MessageSink& first, MessageSink& second) : m_first(first), m_second(second) { }
         void print(Color c, StringView sv) override;
     };
+
+    struct BGMessageSink : 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_lock;
+        // guarded by m_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/binarycaching.h

@@ -7,12 +7,17 @@
 #include <vcpkg/base/downloads.h>
 #include <vcpkg/base/expected.h>
 #include <vcpkg/base/files.h>
+#include <vcpkg/base/message_sinks.h>
 
 #include <vcpkg/packagespec.h>
+#include <vcpkg/sourceparagraph.h>
 
+#include <condition_variable>
 #include <iterator>
+#include <queue>
 #include <set>
 #include <string>
+#include <thread>
 #include <unordered_map>
 #include <vector>
 
@@ -42,6 +47,25 @@ namespace vcpkg
         const IBinaryProvider* m_available_provider = nullptr; // meaningful iff m_status == available
     };
 
+    struct BinaryPackageInformation
+    {
+        explicit BinaryPackageInformation(const InstallPlanAction& action, std::string&& nuspec = "");
+        std::string package_abi;
+        PackageSpec spec;
+        std::string raw_version;
+        std::string nuspec; // only filled if BinaryCache has a provider that returns true for needs_nuspec_data()
+    };
+
+    struct BinaryProviderPushRequest
+    {
+        BinaryProviderPushRequest(BinaryPackageInformation&& info, Path package_dir)
+            : info(std::move(info)), package_dir(std::move(package_dir))
+        {
+        }
+        BinaryPackageInformation info;
+        Path package_dir;
+    };
+
     struct IBinaryProvider
     {
         virtual ~IBinaryProvider() = default;
@@ -52,7 +76,8 @@ namespace vcpkg
 
         /// Called upon a successful build of `action` to store those contents in the binary cache.
         /// Prerequisite: action has a package_abi()
-        virtual void push_success(const InstallPlanAction& action) const = 0;
+        /// returns the number of successful uploads
+        virtual int push_success(const BinaryProviderPushRequest& request, MessageSink& msg_sink) = 0;
 
         /// Gives the IBinaryProvider an opportunity to batch any downloading or server communication for
         /// executing `actions`.
@@ -68,6 +93,8 @@ namespace vcpkg
         /// to the action at the same index in `actions`. The provider must mark the cache status as appropriate.
         /// Prerequisite: `actions` have package ABIs.
         virtual void precheck(View<InstallPlanAction> actions, View<CacheStatus*> cache_status) const = 0;
+
+        virtual bool needs_nuspec_data() const { return false; }
     };
 
     struct UrlTemplate
@@ -77,7 +104,7 @@ namespace vcpkg
         std::vector<std::string> headers_for_get;
 
         LocalizedString valid() const;
-        std::string instantiate_variables(const InstallPlanAction& action) const;
+        std::string instantiate_variables(const BinaryPackageInformation& info) const;
     };
 
     struct BinaryConfigParserState
@@ -124,17 +151,21 @@ namespace vcpkg
 
     struct BinaryCache
     {
-        BinaryCache() = default;
+        BinaryCache(Filesystem& filesystem);
         explicit BinaryCache(const VcpkgCmdArguments& args, const VcpkgPaths& paths);
 
+        ~BinaryCache();
+
         void install_providers(std::vector<std::unique_ptr<IBinaryProvider>>&& providers);
         void install_providers_for(const VcpkgCmdArguments& args, const VcpkgPaths& paths);
 
         /// Attempts to restore the package referenced by `action` into the packages directory.
         RestoreResult try_restore(const InstallPlanAction& action);
 
         /// Called upon a successful build of `action` to store those contents in the binary cache.
-        void push_success(const InstallPlanAction& action);
+        void push_success(const InstallPlanAction& action, Path package_dir);
+
+        void print_push_success_messages();
 
         /// Gives the IBinaryProvider an opportunity to batch any downloading or server communication for
         /// executing `actions`.
@@ -145,9 +176,27 @@ namespace vcpkg
         /// Returns a vector where each index corresponds to the matching index in `actions`.
         std::vector<CacheAvailability> precheck(View<InstallPlanAction> actions);
 
+        void wait_for_async_complete();
+
     private:
+        struct ActionToPush
+        {
+            BinaryProviderPushRequest request;
+            bool clean_after_push = false;
+        };
+        void push_thread_main();
+
+        BGMessageSink bg_msg_sink;
         std::unordered_map<std::string, CacheStatus> m_status;
         std::vector<std::unique_ptr<IBinaryProvider>> m_providers;
+        bool needs_nuspec_data = false;
+        std::condition_variable actions_to_push_notifier;
+        std::mutex actions_to_push_mutex;
+        std::vector<ActionToPush> actions_to_push;
+        std::thread push_thread;
+        std::atomic_bool end_push_thread;
+        std::atomic_int remaining_packages_to_push = 0;
+        Filesystem& filesystem;
     };
 
     ExpectedL<DownloadManagerConfig> parse_download_configuration(const Optional<std::string>& arg);

include/vcpkg/binarycaching.private.h

@@ -5,6 +5,7 @@
 
 #include <vcpkg/base/strings.h>
 
+#include <vcpkg/binarycaching.h>
 #include <vcpkg/dependencies.h>
 
 namespace vcpkg
@@ -35,6 +36,10 @@ namespace vcpkg
     {
         return {Strings::concat(prefix, spec.dir()), format_version_for_nugetref(raw_version, abi_tag)};
     }
+    inline NugetReference make_nugetref(const BinaryPackageInformation& info, const std::string& prefix)
+    {
+        return make_nugetref(info.spec, info.raw_version, info.package_abi, prefix);
+    }
     inline NugetReference make_nugetref(const InstallPlanAction& action, const std::string& prefix)
     {
         return make_nugetref(action.spec,

include/vcpkg/build.h

@@ -200,7 +200,6 @@ namespace vcpkg
     ExtendedBuildResult build_package(const VcpkgCmdArguments& args,
                                       const VcpkgPaths& paths,
                                       const InstallPlanAction& config,
-                                      BinaryCache& binary_cache,
                                       const IBuildLogsRecorder& build_logs_recorder,
                                       const StatusParagraphs& status_db);
 

include/vcpkg/fwd/binarycaching.h

@@ -25,4 +25,5 @@ namespace vcpkg
     struct IBinaryProvider;
     struct BinaryCache;
     struct BinaryConfigParserState;
+    struct BinaryProviderPushRequest;
 }

locales/messages.json

@@ -1266,6 +1266,8 @@
   "_SpecifyTargetArch.comment": "An example of {env_var} is VCPKG_DEFAULT_TRIPLET.",
   "StartCodeUnitInContinue": "found start code unit in continue position",
   "StoreOptionMissingSha": "--store option is invalid without a sha512",
+  "StoredBinariesToDestinations": "Stored binaries in {count} destinations.",
+  "_StoredBinariesToDestinations.comment": "An example of {count} is 42.",
   "StoredBinaryCache": "Stored binary cache: \"{path}\"",
   "_StoredBinaryCache.comment": "An example of {path} is /foo/bar.",
   "SuccessfulyExported": "Exported {package_name} to {path}",
@@ -1410,10 +1412,6 @@
   "_UpdateBaselineUpdatedBaseline.comment": "example of {old_value}, {new_value} is '5507daa796359fe8d45418e694328e878ac2b82f' An example of {url} is https://github.com/microsoft/vcpkg.",
   "UpgradeInManifest": "The upgrade command does not currently support manifest mode. Instead, modify your vcpkg.json and run install.",
   "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}\".",
@@ -1512,6 +1510,8 @@
   "VersionTableHeader": "Version",
   "VersionVerifiedOK": "OK: {package_name}@{version} -> {commit_sha}",
   "_VersionVerifiedOK.comment": "An example of {package_name} is zlib. An example of {version} is 1.3.8. 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/binarycaching.cpp

@@ -24,17 +24,21 @@ struct KnowNothingBinaryProvider : IBinaryProvider
         return RestoreResult::unavailable;
     }
 
-    virtual void push_success(const InstallPlanAction& action) const override { CHECK(action.has_package_abi()); }
+    int push_success(const BinaryProviderPushRequest& request, MessageSink&) override
+    {
+        CHECK_FALSE(request.info.package_abi.empty());
+        return 0;
+    }
 
-    virtual void prefetch(View<InstallPlanAction> actions, View<CacheStatus* const> cache_status) const override
+    void prefetch(View<InstallPlanAction> actions, View<CacheStatus* const> cache_status) const override
     {
         REQUIRE(actions.size() == cache_status.size());
         for (size_t idx = 0; idx < cache_status.size(); ++idx)
         {
             CHECK(actions[idx].has_package_abi() == (cache_status[idx] != nullptr));
         }
     }
-    virtual void precheck(View<InstallPlanAction> actions, View<CacheStatus* const> cache_status) const override
+    void precheck(View<InstallPlanAction> actions, View<CacheStatus* const> cache_status) const override
     {
         REQUIRE(actions.size() == cache_status.size());
         for (const auto c : cache_status)
@@ -365,7 +369,7 @@ Features: a, b
 TEST_CASE ("Provider nullptr checks", "[BinaryCache]")
 {
     // create a binary cache to test
-    BinaryCache uut;
+    BinaryCache uut(get_real_filesystem());
     std::vector<std::unique_ptr<IBinaryProvider>> providers;
     providers.emplace_back(std::make_unique<KnowNothingBinaryProvider>());
     uut.install_providers(std::move(providers));
@@ -391,7 +395,7 @@ Version: 1.5
     InstallPlanAction& ipa_without_abi = install_plan.back();
 
     // test that the binary cache does the right thing. See also CHECKs etc. in KnowNothingBinaryProvider
-    uut.push_success(ipa_without_abi); // should have no effects
+    uut.push_success(ipa_without_abi, {}); // should have no effects
     CHECK(uut.try_restore(ipa_without_abi) == RestoreResult::unavailable);
     uut.prefetch(install_plan); // should have no effects
 }

src/vcpkg/base/message_sinks.cpp

@@ -57,4 +57,61 @@ 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;
+        }
+
+        std::string s = sv.to_string();
+        auto pos = s.find_last_of('\n');
+        if (pos != std::string::npos)
+        {
+            {
+                std::lock_guard<std::mutex> lk(m_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, s.substr(0, pos + 1));
+            }
+            m_unpublished.clear();
+            if (s.size() > pos + 1)
+            {
+                m_unpublished.emplace_back(c, s.substr(pos + 1));
+            }
+        }
+        else
+        {
+            m_unpublished.emplace_back(c, std::move(s));
+        }
+    }
+
+    void BGMessageSink::print_published()
+    {
+        std::lock_guard<std::mutex> lk(m_lock);
+        for (auto&& m : m_published)
+        {
+            out_sink.print(m.first, m.second);
+        }
+        m_published.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_lock);
+
+        m_print_directly_to_out_sink = true;
+        for (auto& messages : {&m_published, &m_unpublished})
+        {
+            for (auto&& m : *messages)
+            {
+                out_sink.print(m.first, m.second);
+            }
+            messages->clear();
+        }
+    }
 }