Binary cache: async push_success

include/vcpkg/archives.h

@@ -9,6 +9,8 @@
 #include <vcpkg/fwd/tools.h>
 #include <vcpkg/fwd/vcpkgpaths.h>
 
+#include <vcpkg/base/diagnostics.h>
+#include <vcpkg/base/optional.h>
 #include <vcpkg/base/path.h>
 
 namespace vcpkg
@@ -54,22 +56,20 @@ namespace vcpkg
 
     struct ZipTool
     {
-        static ExpectedL<ZipTool> make(const ToolCache& tools, MessageSink& status_sink);
+        void setup(const ToolCache& tools, MessageSink& status_sink);
 
-    private:
-        ZipTool() = default;
-
-#if defined _WIN32
-        Path seven_zip;
-#endif
-
-    public:
         // Compress the source directory into the destination file.
-        ExpectedL<Unit> compress_directory_to_zip(const Filesystem& fs,
-                                                  const Path& source,
-                                                  const Path& destination) const;
+        bool compress_directory_to_zip(DiagnosticContext& context,
+                                       const Filesystem& fs,
+                                       const Path& source,
+                                       const Path& destination) const;
 
         Command decompress_zip_archive_cmd(const Path& dst, const Path& archive_path) const;
+
+    private:
+#if defined _WIN32
+        Optional<Path> seven_zip;
+#endif
     };
 
     std::vector<ExpectedL<Unit>> decompress_in_parallel(View<Command> jobs);

include/vcpkg/base/background-work-queue.h

@@ -0,0 +1,51 @@
+#pragma once
+
+#include <condition_variable>
+#include <mutex>
+#include <vector>
+
+template<class WorkItem>
+struct BackgroundWorkQueue
+{
+    template<class... Args>
+    void push(Args&&... args)
+    {
+        std::lock_guard<std::mutex> lock(m_mtx);
+        m_tasks.emplace_back(std::forward<Args>(args)...);
+        m_cv.notify_one();
+    }
+
+    bool get_work(std::vector<WorkItem>& out)
+    {
+        std::unique_lock<std::mutex> lock(m_mtx);
+        for (;;)
+        {
+            if (!m_tasks.empty())
+            {
+                out.clear();
+                swap(out, m_tasks);
+                return true;
+            }
+
+            if (!m_running)
+            {
+                return false;
+            }
+
+            m_cv.wait(lock);
+        }
+    }
+
+    void stop()
+    {
+        std::lock_guard<std::mutex> lock(m_mtx);
+        m_running = false;
+        m_cv.notify_all();
+    }
+
+private:
+    std::mutex m_mtx;
+    std::condition_variable m_cv;
+    std::vector<WorkItem> m_tasks;
+    bool m_running = true;
+};

include/vcpkg/base/fwd/message_sinks.h

@@ -13,4 +13,5 @@ namespace vcpkg
 
     struct FileSink;
     struct TeeSink;
+    struct BGMessageSink;
 }

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

@@ -919,7 +919,6 @@ DECLARE_MESSAGE(CommandFailed,
                 "failed with the following output:")
 DECLARE_MESSAGE(CommunityTriplets, (), "", "Community Triplets:")
 DECLARE_MESSAGE(CompilerPath, (msg::path), "", "Compiler found: {path}")
-DECLARE_MESSAGE(CompressFolderFailed, (msg::path), "", "Failed to compress folder \"{path}\":")
 DECLARE_MESSAGE(ComputingInstallPlan, (), "", "Computing installation plan...")
 DECLARE_MESSAGE(ConfigurationErrorRegistriesWithoutBaseline,
                 (msg::path, msg::url),
@@ -2769,11 +2768,15 @@ DECLARE_MESSAGE(SpecifyTargetArch,
                 "'vcpkg help triplet' is a command line that should not be localized",
                 "Target triplet. See 'vcpkg help triplet' (default: {env_var})")
 DECLARE_MESSAGE(StartCodeUnitInContinue, (), "", "found start code unit in continue position")
-DECLARE_MESSAGE(StoredBinariesToDestinations,
-                (msg::count, msg::elapsed),
-                "",
-                "Stored binaries in {count} destinations in {elapsed}.")
 DECLARE_MESSAGE(StoreOptionMissingSha, (), "", "--store option is invalid without a sha512")
+DECLARE_MESSAGE(SubmittingBinaryCacheBackground,
+                (msg::spec, msg::count),
+                "",
+                "Starting submission of {spec} to {count} binary cache(s) in the background")
+DECLARE_MESSAGE(SubmittingBinaryCacheComplete,
+                (msg::spec, msg::count, msg::elapsed),
+                "",
+                "Completed submission of {spec} to {count} binary cache(s) in {elapsed}")
 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.")
 DECLARE_MESSAGE(SuggestStartingBashShell,
@@ -3019,7 +3022,7 @@ DECLARE_MESSAGE(
 DECLARE_MESSAGE(UploadingBinariesToVendor,
                 (msg::spec, msg::vendor, msg::path),
                 "",
-                "Uploading binaries for '{spec}' to '{vendor}' source \"{path}\".")
+                "Uploading binaries for {spec} to {vendor} from {path}")
 DECLARE_MESSAGE(UsageTextHere, (), "", "the usage file is here")
 DECLARE_MESSAGE(UsageInstallInstructions, (), "", "you can install the usage file with the following CMake")
 DECLARE_MESSAGE(UseEnvVar,
@@ -3267,6 +3270,10 @@ 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),
+                "",
+                "Waiting for {count} remaining binary cache submissions...")
 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,7 @@
 
 #include <vcpkg/base/messages.h>
 
+#include <mutex>
 #include <string>
 #include <vector>
 
@@ -79,4 +80,26 @@ namespace vcpkg
         virtual void println(Color color, const LocalizedString& line) override;
         virtual void println(Color color, LocalizedString&& line) override;
     };
+
+    struct BGMessageSink final : MessageSink
+    {
+        BGMessageSink(MessageSink& out_sink) : out_sink(out_sink) { }
+        ~BGMessageSink() { publish_directly_to_out_sink(); }
+        // must be called from producer
+        virtual void println(const MessageLine& line) override;
+        virtual void println(MessageLine&& line) override;
+        using MessageSink::println;
+
+        // 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<MessageLine> m_published;
+        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/background-work-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>
 
@@ -186,14 +189,17 @@ namespace vcpkg
     struct ReadOnlyBinaryCache
     {
         ReadOnlyBinaryCache() = default;
-        ReadOnlyBinaryCache(BinaryProviders&& providers);
+        ReadOnlyBinaryCache(const ReadOnlyBinaryCache&) = delete;
+        ReadOnlyBinaryCache& operator=(const ReadOnlyBinaryCache&) = delete;
 
         /// Gives the IBinaryProvider an opportunity to batch any downloading or server communication for
         /// executing `actions`.
         void fetch(View<InstallPlanAction> actions);
 
         bool is_restored(const InstallPlanAction& ipa) const;
 
+        void install_read_provider(std::unique_ptr<IReadBinaryProvider>&& provider);
+
         /// Checks whether the `actions` are present in the cache, without restoring them. Used by CI to determine
         /// missing packages.
         /// Returns a vector where each index corresponds to the matching index in `actions`.
@@ -205,24 +211,83 @@ namespace vcpkg
         std::unordered_map<std::string, CacheStatus> m_status;
     };
 
+    struct BinaryCacheSyncState;
+    struct BinaryCacheSynchronizer
+    {
+        using backing_uint_t = std::conditional_t<sizeof(size_t) == 4, uint32_t, uint64_t>;
+        using counter_uint_t = std::conditional_t<sizeof(size_t) == 4, uint16_t, uint32_t>;
+        static constexpr backing_uint_t SubmissionCompleteBit = static_cast<backing_uint_t>(1)
+                                                                << (sizeof(backing_uint_t) * 8 - 1);
+        static constexpr backing_uint_t UpperShift = sizeof(counter_uint_t) * 8;
+        static constexpr backing_uint_t SubmittedMask =
+            static_cast<backing_uint_t>(static_cast<counter_uint_t>(-1) >> 1u);
+        static constexpr backing_uint_t CompletedMask = SubmittedMask << UpperShift;
+        static constexpr backing_uint_t OneCompleted = static_cast<backing_uint_t>(1) << UpperShift;
+
+        void add_submitted() noexcept;
+        BinaryCacheSyncState fetch_add_completed() noexcept;
+        counter_uint_t fetch_incomplete_mark_submission_complete() noexcept;
+
+    private:
+        // This is morally:
+        // struct State {
+        //    counter_uint_t jobs_submitted;
+        //    bool unused;
+        //    counter_uint_t_minus_one_bit jobs_completed;
+        //    bool submission_complete;
+        // };
+        std::atomic<backing_uint_t> m_state = 0;
+    };
+
+    struct BinaryCacheSyncState
+    {
+        BinaryCacheSynchronizer::counter_uint_t jobs_submitted;
+        BinaryCacheSynchronizer::counter_uint_t jobs_completed;
+        bool submission_complete;
+    };
+
+    // 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 written by the foreground thread.
     struct BinaryCache : ReadOnlyBinaryCache
     {
-        static ExpectedL<BinaryCache> make(const VcpkgCmdArguments& args, const VcpkgPaths& paths, MessageSink& sink);
+        bool install_providers(const VcpkgCmdArguments& args, const VcpkgPaths& paths, MessageSink& status_sink);
 
-        BinaryCache(const Filesystem& fs);
+        explicit BinaryCache(const Filesystem& fs);
         BinaryCache(const BinaryCache&) = delete;
-        BinaryCache(BinaryCache&&) = default;
-
+        BinaryCache& operator=(const BinaryCache&) = delete;
+        ~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_updates();
+        void wait_for_async_complete_and_join();
+
     private:
-        BinaryCache(BinaryProviders&& providers, const Filesystem& fs);
+        struct ActionToPush
+        {
+            BinaryPackageWriteInfo request;
+            CleanPackages clean_after_push;
+        };
 
-        const Filesystem& m_fs;
-        Optional<ZipTool> m_zip_tool;
+        ZipTool m_zip_tool;
         bool m_needs_nuspec_data = false;
         bool m_needs_zip_file = false;
+
+        const Filesystem& m_fs;
+
+        BGMessageSink m_bg_msg_sink;
+        BackgroundWorkQueue<ActionToPush> m_actions_to_push;
+        BinaryCacheSynchronizer m_synchronizer;
+        std::thread m_push_thread;
+
+        void push_thread_main();
     };
 
     ExpectedL<AssetCachingSettings> parse_download_configuration(const Optional<std::string>& arg);

locales/messages.json

@@ -531,8 +531,6 @@
   "CommunityTriplets": "Community Triplets:",
   "CompilerPath": "Compiler found: {path}",
   "_CompilerPath.comment": "An example of {path} is /foo/bar.",
-  "CompressFolderFailed": "Failed to compress folder \"{path}\":",
-  "_CompressFolderFailed.comment": "An example of {path} is /foo/bar.",
   "ComputingInstallPlan": "Computing installation plan...",
   "ConfigurationErrorRegistriesWithoutBaseline": "The configuration defined in {path} is invalid.\n\nUsing registries requires that a baseline is set for the default registry or that the default registry is null.\n\nSee {url} for more details.",
   "_ConfigurationErrorRegistriesWithoutBaseline.comment": "An example of {path} is /foo/bar. An example of {url} is https://github.com/microsoft/vcpkg.",
@@ -1458,8 +1456,10 @@
   "_SpecifyTargetArch.comment": "'vcpkg help triplet' is a command line that should not be localized 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 in {elapsed}.",
-  "_StoredBinariesToDestinations.comment": "An example of {count} is 42. An example of {elapsed} is 3.532 min.",
+  "SubmittingBinaryCacheBackground": "Starting submission of {spec} to {count} binary cache(s) in the background",
+  "_SubmittingBinaryCacheBackground.comment": "An example of {spec} is zlib:x64-windows. An example of {count} is 42.",
+  "SubmittingBinaryCacheComplete": "Completed submission of {spec} to {count} binary cache(s) in {elapsed}",
+  "_SubmittingBinaryCacheComplete.comment": "An example of {spec} is zlib:x64-windows. An example of {count} is 42. An example of {elapsed} is 3.532 min.",
   "SuccessfulyExported": "Exported {package_name} to {path}",
   "_SuccessfulyExported.comment": "An example of {package_name} is zlib. An example of {path} is /foo/bar.",
   "SuggestGitPull": "The result may be outdated. Run `git pull` to get the latest results.",
@@ -1599,7 +1599,7 @@
   "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.",
-  "UploadingBinariesToVendor": "Uploading binaries for '{spec}' to '{vendor}' source \"{path}\".",
+  "UploadingBinariesToVendor": "Uploading binaries for {spec} to {vendor} from {path}",
   "_UploadingBinariesToVendor.comment": "An example of {spec} is zlib:x64-windows. An example of {vendor} is Azure. An example of {path} is /foo/bar.",
   "UsageInstallInstructions": "you can install the usage file with the following CMake",
   "UsageTextHere": "the usage file is here",
@@ -1721,6 +1721,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": "Waiting for {count} remaining binary cache submissions...",
+  "_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.",