winget-cli

Unnamed repository; edit this file 'description' to name the repository.
Log | Files | Refs | README | LICENSE

ContextOrchestrator.h (10204B)

      1 // Copyright (c) Microsoft Corporation.
      2 // Licensed under the MIT License.
      3 #pragma once
      4 #include <AppInstallerLogging.h>
      5 #include <winget/RepositorySource.h>
      6 #include "ExecutionReporter.h"
      7 #include "ExecutionArgs.h"
      8 #include "ExecutionContextData.h"
      9 #include "CompletionData.h"
     10 #include "Command.h"
     11 #include "COMContext.h"
     12 #include <wil/resource.h>
     13 #include <string>
     14 #include <string_view>
     15 
     16 namespace AppInstaller::CLI::Execution
     17 {
     18     enum class OrchestratorQueueItemState
     19     {
     20         // Created but not yet queued
     21         NotQueued,
     22         // Queued and waiting to be run
     23         Queued,
     24         // Running in the thread pool
     25         Running,
     26         // Cancelled before it was run; will be deleted when we try to run it
     27         Cancelled
     28     };
     29 
     30     std::string_view ToString(OrchestratorQueueItemState state);
     31 
     32     struct OrchestratorQueueItemId
     33     {
     34         OrchestratorQueueItemId(std::wstring packageId, std::wstring sourceId) : m_packageId(std::move(packageId)), m_sourceId(std::move(sourceId)) {}
     35         std::wstring_view GetPackageId() const { return m_packageId; }
     36         std::wstring_view GetSourceId() const { return m_sourceId; }
     37 
     38         bool IsSame(const OrchestratorQueueItemId& comparisonQueueItemId) const;
     39     private:
     40         std::wstring m_packageId;
     41         std::wstring m_sourceId;
     42     };
     43 
     44     struct OrchestratorQueue;
     45 
     46     enum class PackageOperationType
     47     {
     48         Search,
     49         Install,
     50         Upgrade,
     51         Uninstall,
     52         Download,
     53         Repair,
     54     };
     55 
     56     struct ContextOrchestrator;
     57 
     58     struct OrchestratorQueueItem
     59     {
     60         OrchestratorQueueItem(OrchestratorQueueItemId id, std::unique_ptr<COMContext> context, PackageOperationType operationType) :
     61             m_id(std::move(id)), m_context(std::move(context)), m_operationType(operationType) {}
     62 
     63         OrchestratorQueueItemState GetState() const { return m_state; }
     64         void SetState(OrchestratorQueueItemState state) { m_state = state; }
     65 
     66         OrchestratorQueue* GetCurrentQueue() const { return m_currentQueue; }
     67         void SetCurrentQueue(OrchestratorQueue* currentQueue) { m_currentQueue = currentQueue; }
     68 
     69         COMContext& GetContext() const { return *m_context; }
     70         const wil::unique_event& GetCompletedEvent() const { return m_completedEvent; }
     71         const OrchestratorQueueItemId& GetId() const { return m_id; }
     72 
     73         void AddCommand(std::unique_ptr<Command> command) { m_commands.push_back(std::move(command)); }
     74         const Command& GetNextCommand() const { return *m_commands.front(); }
     75         std::unique_ptr<Command> PopNextCommand()
     76         {
     77             m_isOnFirstCommand = false;
     78             std::unique_ptr<Command> command = std::move(m_commands.front());
     79             m_commands.pop_front();
     80             return command;
     81         }
     82 
     83         bool IsOnFirstCommand() const { return m_isOnFirstCommand; }
     84         bool IsComplete() const { return m_commands.empty(); }
     85         bool IsApplicableForInstallingSource() const { return m_operationType == PackageOperationType::Install || m_operationType == PackageOperationType::Upgrade; }
     86         PackageOperationType GetPackageOperationType() const { return m_operationType; }
     87         std::string_view GetItemCommandName() const;
     88 
     89         void HandleItemCompletion(ContextOrchestrator& orchestrator) const;
     90 
     91     private:
     92         OrchestratorQueueItemState m_state = OrchestratorQueueItemState::NotQueued;
     93         std::unique_ptr<COMContext> m_context;
     94         wil::unique_event m_completedEvent{ wil::EventOptions::ManualReset };
     95         OrchestratorQueueItemId m_id;
     96         std::deque<std::unique_ptr<Command>> m_commands;
     97         bool m_isOnFirstCommand = true;
     98         OrchestratorQueue* m_currentQueue = nullptr;
     99         PackageOperationType m_operationType;
    100     };
    101 
    102     struct OrchestratorQueueItemFactory
    103     {
    104         // Create queue item for install/upgrade
    105         static std::unique_ptr<OrchestratorQueueItem> CreateItemForInstall(std::wstring packageId, std::wstring sourceId, std::unique_ptr<COMContext> context, bool isUpgrade);
    106         // Create queue item for uninstall
    107         static std::unique_ptr<OrchestratorQueueItem> CreateItemForUninstall(std::wstring packageId, std::wstring sourceId, std::unique_ptr<COMContext> context);
    108         // Create queue item for finding existing entry from the orchestrator queue
    109         static std::unique_ptr<OrchestratorQueueItem> CreateItemForSearch(std::wstring packageId, std::wstring sourceId, std::unique_ptr<COMContext> context);
    110         // Create queue item for download
    111         static std::unique_ptr<OrchestratorQueueItem> CreateItemForDownload(std::wstring packageId, std::wstring sourceId, std::unique_ptr<COMContext> context);
    112         // Create queue item for repair
    113         static std::unique_ptr<OrchestratorQueueItem> CreateItemForRepair(std::wstring packageId, std::wstring sourceId, std::unique_ptr<COMContext> context);
    114     };
    115 
    116     struct ContextOrchestrator
    117     {
    118         ContextOrchestrator();
    119         ContextOrchestrator(unsigned int hardwareConcurrency);
    120         static ContextOrchestrator& Instance();
    121 
    122         void EnqueueAndRunItem(const std::shared_ptr<OrchestratorQueueItem>& queueItem);
    123         void CancelQueueItem(const OrchestratorQueueItem& item);
    124 
    125         std::shared_ptr<OrchestratorQueueItem> GetQueueItem(const OrchestratorQueueItemId& queueItemId);
    126 
    127         void AddItemManifestToInstallingSource(const OrchestratorQueueItem& queueItem);
    128         void RemoveItemManifestFromInstallingSource(const OrchestratorQueueItem& queueItem);
    129 
    130         // Functions for ServerShutdownSynchronization::ComponentSystem registration
    131         static void RegisterForShutdownSynchronization();
    132         static void StaticDisable(CancelReason reason);
    133         static void StaticCancelQueuedItems(CancelReason reason);
    134         static void StaticWaitForRunningItems();
    135 
    136         void Disable(CancelReason reason);
    137         void CancelQueuedItems(CancelReason reason);
    138         void WaitForRunningItems();
    139 
    140         // Waits for running items to complete; waits up to full time out in *each* queue.
    141         // Returns true to indicate all queues are empty before the timeout.
    142         bool WaitForRunningItems(DWORD timeoutMilliseconds);
    143 
    144         // Gets a string that represents the current state of the orchestrator.
    145         std::string GetStatusString();
    146 
    147     private:
    148         std::mutex m_queueLock;
    149         bool m_enabled = true;
    150         CancelReason m_disabledReason = CancelReason::None;
    151         void AddCommandQueue(std::string_view commandName, UINT32 allowedThreads);
    152         void RemoveItemInState(const OrchestratorQueueItem& item, OrchestratorQueueItemState state);
    153 
    154         _Requires_lock_held_(m_queueLock)
    155         std::shared_ptr<OrchestratorQueueItem> FindById(const OrchestratorQueueItemId& queueItemId);
    156 
    157         Repository::Source m_installingWriteableSource;
    158         std::map<std::string, std::unique_ptr<OrchestratorQueue>> m_commandQueues;
    159     };
    160 
    161     // One of the queues used by the orchestrator.
    162     // All items in the queue execute the same command.
    163     // The queue allows multiple items to run at the same time, up to a limit.
    164     struct OrchestratorQueue
    165     {
    166         OrchestratorQueue(ContextOrchestrator& orchestrator, std::string_view commandName, UINT32 allowedThreads);
    167         ~OrchestratorQueue();
    168 
    169         // Name of the command this queue can execute
    170         std::string_view CommandName() const { return m_commandName; }
    171 
    172         // Enqueues an item to be run when there are threads available.
    173         void EnqueueAndRunItem(const std::shared_ptr<OrchestratorQueueItem>& item);
    174 
    175         // Removes an item by id, provided that it is in the given state.
    176         // Returns true if an item was removed.
    177         // The item can be removed globally from the orchestrator, or from just this queue.
    178         bool RemoveItemInState(const OrchestratorQueueItem& item, OrchestratorQueueItemState state, bool isGlobalRemove);
    179 
    180         // Finds an item by id, if it is in the queue.
    181         _Requires_lock_held_(m_itemLock)
    182         std::shared_ptr<OrchestratorQueueItem> FindById(const OrchestratorQueueItemId& queueItemId);
    183 
    184         // Runs a single item from the queue.
    185         void RunItem(const OrchestratorQueueItemId& itemId);
    186 
    187         // Cancels and "removes" all items in the queue.
    188         void CancelAllItems(CancelReason reason);
    189 
    190         // Waits until the empty queue event is signaled.
    191         void WaitForEmptyQueue();
    192 
    193         // Waits until the empty queue event is signaled.
    194         // Returns true to indicate the queue is empty before the timeout.
    195         bool WaitForEmptyQueue(DWORD timeoutMilliseconds);
    196 
    197         // Gets a string that represents the current state of the queue.
    198         std::string GetStatusString();
    199 
    200     private:
    201         // Enqueues an item.
    202         void EnqueueItem(const std::shared_ptr<OrchestratorQueueItem>& item);
    203 
    204         _Requires_lock_held_(m_itemLock)
    205         std::deque<std::shared_ptr<OrchestratorQueueItem>>::iterator FindIteratorById(const OrchestratorQueueItemId& comparisonQueueItemId);
    206 
    207         ContextOrchestrator& m_orchestrator;
    208         std::string_view m_commandName;
    209 
    210         // Number of threads allowed to run items in this queue.
    211         const UINT32 m_allowedThreads;
    212 
    213         // Thread pool for this queue, and associated objects.
    214         // All work items will be added to the callback environment, and the cleanup group
    215         // will manage their closing.
    216         // See https://docs.microsoft.com/windows/win32/procthread/using-the-thread-pool-functions
    217         TP_CALLBACK_ENVIRON m_threadPoolCallbackEnviron;
    218         wil::unique_any<PTP_POOL, decltype(CloseThreadpool), CloseThreadpool> m_threadPool;
    219         wil::unique_any<PTP_CLEANUP_GROUP, decltype(CloseThreadpoolCleanupGroup), CloseThreadpoolCleanupGroup> m_threadPoolCleanupGroup;
    220 
    221         std::mutex m_itemLock;
    222         std::deque<std::shared_ptr<OrchestratorQueueItem>> m_queueItems;
    223         wil::slim_event_manual_reset m_queueEmpty{ true };
    224     };
    225 }