Binary cache: async push_success

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

@@ -2768,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,
@@ -3018,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,

include/vcpkg/binarycaching.h

@@ -211,6 +211,41 @@ 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.
@@ -249,8 +284,7 @@ namespace vcpkg
 
         BGMessageSink m_bg_msg_sink;
         BackgroundWorkQueue<ActionToPush> m_actions_to_push;
-        std::atomic<int> m_total_packages_to_push = 0;
-        std::atomic<int> m_total_packages_pushed = 0;
+        BinaryCacheSynchronizer m_synchronizer;
         std::thread m_push_thread;
 
         void push_thread_main();

locales/messages.json

@@ -1456,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.",
@@ -1597,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",

src/vcpkg-test/binarycaching.cpp

@@ -469,3 +469,36 @@ Description: a spiffy compression library wrapper
 </packages>
 )");
 }
+
+TEST_CASE ("Synchronizer operations", "[BinaryCache]")
+{
+    {
+        BinaryCacheSynchronizer sync;
+        auto result = sync.fetch_add_completed();
+        REQUIRE(result.jobs_submitted == 0);
+        REQUIRE(result.jobs_completed == 1);
+        REQUIRE_FALSE(result.submission_complete);
+    }
+
+    {
+        BinaryCacheSynchronizer sync;
+        sync.add_submitted();
+        sync.add_submitted();
+        auto result = sync.fetch_add_completed();
+        REQUIRE(result.jobs_submitted == 2);
+        REQUIRE(result.jobs_completed == 1);
+        REQUIRE_FALSE(result.submission_complete);
+    }
+
+    {
+        BinaryCacheSynchronizer sync;
+        sync.add_submitted();
+        REQUIRE(sync.fetch_incomplete_mark_submission_complete() == 1);
+        sync.add_submitted();
+        REQUIRE(sync.fetch_incomplete_mark_submission_complete() == 2);
+        auto result = sync.fetch_add_completed();
+        REQUIRE(result.jobs_submitted == 2);
+        REQUIRE(result.jobs_completed == 1);
+        REQUIRE(result.submission_complete);
+    }
+}

src/vcpkg/binarycaching.cpp

@@ -2213,6 +2213,44 @@ namespace vcpkg
         });
     }
 
+    void BinaryCacheSynchronizer::add_submitted() noexcept
+    {
+        // This can set the unused bit but if that happens we are terminating anyway.
+        if ((m_state.fetch_add(1, std::memory_order_acq_rel) & SubmittedMask) == SubmittedMask)
+        {
+            Checks::unreachable(VCPKG_LINE_INFO, "Maximum job count exceeded");
+        }
+    }
+
+    BinaryCacheSyncState BinaryCacheSynchronizer::fetch_add_completed() noexcept
+    {
+        auto old = m_state.load(std::memory_order_acquire);
+        backing_uint_t local;
+        do
+        {
+            local = old;
+            if ((local & CompletedMask) == CompletedMask)
+            {
+                Checks::unreachable(VCPKG_LINE_INFO, "Maximum job count exceeded");
+            }
+
+            local += OneCompleted;
+        } while (m_state.compare_exchange_weak(old, local, std::memory_order_acq_rel));
+
+        BinaryCacheSyncState result;
+        result.jobs_submitted = local & SubmittedMask;
+        result.jobs_completed = (local & CompletedMask) >> UpperShift;
+        result.submission_complete = (local & SubmissionCompleteBit) != 0;
+        return result;
+    }
+
+    BinaryCacheSynchronizer::counter_uint_t BinaryCacheSynchronizer::
+        fetch_incomplete_mark_submission_complete() noexcept
+    {
+        auto state = m_state.fetch_or(SubmissionCompleteBit, std::memory_order_acq_rel);
+        return (state & SubmittedMask) - ((state & CompletedMask) >> UpperShift);
+    }
+
     bool BinaryCache::install_providers(const VcpkgCmdArguments& args,
                                         const VcpkgPaths& paths,
                                         MessageSink& status_sink)
@@ -2468,7 +2506,9 @@ namespace vcpkg
                         generate_nuspec(request.package_dir, action, m_config.nuget_prefix, m_config.nuget_repo);
                 }
 
-                m_total_packages_to_push++;
+                m_synchronizer.add_submitted();
+                msg::println(msg::format(
+                    msgSubmittingBinaryCacheBackground, msg::spec = action.spec, msg::count = m_config.write.size()));
                 m_actions_to_push.push(ActionToPush{std::move(request), clean_packages});
                 return;
             }
@@ -2485,10 +2525,12 @@ namespace vcpkg
     void BinaryCache::wait_for_async_complete_and_join()
     {
         m_bg_msg_sink.print_published();
-        if (m_total_packages_pushed.load() < m_total_packages_to_push.load())
+        auto incomplete_count = m_synchronizer.fetch_incomplete_mark_submission_complete();
+        if (incomplete_count != 0)
         {
-            msg::println(msgWaitUntilPackagesUploaded, msg::count = m_total_packages_to_push.load());
+            msg::println(msgWaitUntilPackagesUploaded, msg::count = incomplete_count);
         }
+
         m_bg_msg_sink.publish_directly_to_out_sink();
         m_actions_to_push.stop();
         if (m_push_thread.joinable())
@@ -2534,11 +2576,17 @@ namespace vcpkg
                     m_fs.remove_all(action_to_push.request.package_dir, VCPKG_LINE_INFO);
                 }
 
-                m_bg_msg_sink.println(
-                    msg::format(
-                        msgStoredBinariesToDestinations, msg::count = num_destinations, msg::elapsed = timer.elapsed())
-                        .append_raw(
-                            fmt::format(" ({}/{})", ++(m_total_packages_pushed), m_total_packages_to_push.load())));
+                auto sync_state = m_synchronizer.fetch_add_completed();
+                auto message = msg::format(msgSubmittingBinaryCacheComplete,
+                                           msg::spec = action_to_push.request.spec,
+                                           msg::count = num_destinations,
+                                           msg::elapsed = timer.elapsed());
+                if (sync_state.submission_complete)
+                {
+                    message.append_raw(fmt::format(" ({}/{})", sync_state.jobs_completed, sync_state.jobs_submitted));
+                }
+
+                m_bg_msg_sink.println(message);
             }
         }
     }

src/vcpkg/dependencies.cpp

@@ -661,7 +661,7 @@ namespace vcpkg
     bool RemovePlan::has_non_user_requested() const
     {
         return Util::any_of(remove,
-                             [](const RemovePlanAction& a) { return a.request_type != RequestType::USER_REQUESTED; });
+                            [](const RemovePlanAction& a) { return a.request_type != RequestType::USER_REQUESTED; });
     }
 
     RemovePlan create_remove_plan(const std::vector<PackageSpec>& specs, const StatusParagraphs& status_db)