commit 39034e1e0bff692c40e964c1f1d5b177edef693b
parent 7d178a61e732e820bb562269ce0b59c982dcf905
Author: Chacón <lechacon@users.noreply.github.com>
Date: Tue, 16 Nov 2021 18:57:58 -0800
Implement parallel downloads for COM scenarios (#1588)
Diffstat:
7 files changed, 281 insertions(+), 79 deletions(-)
diff --git a/.github/actions/spelling/allow.txt b/.github/actions/spelling/allow.txt
@@ -532,7 +532,7 @@ Testrun
testsettingname
TEXTFORMAT
TEXTINCLUDE
-there're
+Threadpool
Timeline
todo
tokenizer
diff --git a/src/AppInstallerCLICore/Commands/COMInstallCommand.h b/src/AppInstallerCLICore/Commands/COMInstallCommand.h
@@ -8,7 +8,8 @@ namespace AppInstaller::CLI
// IMPORTANT: To use this command, the caller should have already retrieved the package manifest (GetManifest()) and added it to the Context Data
struct COMDownloadCommand final : public Command
{
- COMDownloadCommand(std::string_view parent) : Command("download", parent) {}
+ constexpr static std::string_view CommandName = "download"sv;
+ COMDownloadCommand(std::string_view parent) : Command(CommandName, parent) {}
protected:
void ExecuteInternal(Execution::Context& context) const override;
@@ -17,7 +18,8 @@ namespace AppInstaller::CLI
// IMPORTANT: To use this command, the caller should have already retrieved the package manifest (GetManifest()) and added it to the Context Data
struct COMInstallCommand final : public Command
{
- COMInstallCommand(std::string_view parent) : Command("install", parent) {}
+ constexpr static std::string_view CommandName = "install"sv;
+ COMInstallCommand(std::string_view parent) : Command(CommandName, parent) {}
protected:
void ExecuteInternal(Execution::Context& context) const override;
diff --git a/src/AppInstallerCLICore/ContextOrchestrator.cpp b/src/AppInstallerCLICore/ContextOrchestrator.cpp
@@ -10,6 +10,21 @@
namespace AppInstaller::CLI::Execution
{
+ namespace
+ {
+ // Callback function used by worker threads in the queue.
+ // context must be a pointer to a queue item.
+ void CALLBACK OrchestratorQueueWorkCallback(PTP_CALLBACK_INSTANCE, PVOID context, PTP_WORK)
+ {
+ auto queueItem = reinterpret_cast<OrchestratorQueueItem*>(context);
+ auto queue = queueItem->GetCurrentQueue();
+ if (queue)
+ {
+ queue->RunItem(queueItem->GetId());
+ }
+ }
+ }
+
ContextOrchestrator& ContextOrchestrator::Instance()
{
static ContextOrchestrator s_instance;
@@ -21,90 +36,195 @@ namespace AppInstaller::CLI::Execution
ProgressCallback progress;
m_installingWriteableSource = Repository::Source(Repository::PredefinedSource::Installing);
m_installingWriteableSource.Open(progress);
+
+ // Decide how many threads to use for each command.
+ // We always allow only one install at a time.
+ // For download, if we can find the number of supported concurrent threads,
+ // use that as the maximum (up to 3); otherwise use a single thread.
+ const auto supportedConcurrentThreads = std::thread::hardware_concurrency();
+ const UINT32 maxDownloadThreads = 3;
+ const UINT32 installThreads = 1;
+ const UINT32 downloadThreads = std::min(supportedConcurrentThreads ? supportedConcurrentThreads - 1 : 1, maxDownloadThreads);
+
+ AddCommandQueue(COMDownloadCommand::CommandName, downloadThreads);
+ AddCommandQueue(COMInstallCommand::CommandName, installThreads);
}
- _Requires_lock_held_(m_queueLock)
- std::deque<std::shared_ptr<OrchestratorQueueItem>>::iterator ContextOrchestrator::FindIteratorById(const OrchestratorQueueItemId& comparisonQueueItemId)
+ void ContextOrchestrator::AddCommandQueue(std::string_view commandName, UINT32 allowedThreads)
{
- return std::find_if(m_queueItems.begin(), m_queueItems.end(), [&comparisonQueueItemId](const std::shared_ptr<OrchestratorQueueItem>& item) {return (item->GetId().IsSame(comparisonQueueItemId)); });
-
+ m_commandQueues.emplace(commandName, std::make_unique<OrchestratorQueue>(commandName, allowedThreads));
}
+
_Requires_lock_held_(m_queueLock)
std::shared_ptr<OrchestratorQueueItem> ContextOrchestrator::FindById(const OrchestratorQueueItemId& comparisonQueueItemId)
{
- auto itr = FindIteratorById(comparisonQueueItemId);
- if (itr != m_queueItems.end())
+ for (const auto& queue : m_commandQueues)
{
- return *itr;
+ auto item = queue.second->FindById(comparisonQueueItemId);
+ if (item)
+ {
+ return item;
+ }
}
+
return {};
}
-
- void ContextOrchestrator::EnqueueItem(std::shared_ptr<OrchestratorQueueItem> item)
+
+ void ContextOrchestrator::EnqueueAndRunItem(std::shared_ptr<OrchestratorQueueItem> item)
{
- {
- std::lock_guard<std::mutex> lockQueue{ m_queueLock };
+ std::lock_guard<std::mutex> lockQueue{ m_queueLock };
+ if (item->IsOnFirstCommand())
+ {
THROW_HR_IF(HRESULT_FROM_WIN32(ERROR_INSTALL_ALREADY_RUNNING), FindById(item->GetId()));
- m_queueItems.push_back(item);
}
- // Add the package to the Installing source so that it can be queried using the Source interface.
- const auto& manifest = item->GetContext().Get<Execution::Data::Manifest>();
- m_installingWriteableSource.AddPackageVersion(manifest, std::filesystem::path{ manifest.Id + '.' + manifest.Version });
+ m_commandQueues.at(std::string(item->GetNextCommand().Name()))->EnqueueAndRunItem(item);
+ }
+ void ContextOrchestrator::RemoveItemInState(const OrchestratorQueueItem& item, OrchestratorQueueItemState state)
+ {
+ std::lock_guard<std::mutex> lockQueue{ m_queueLock };
+ for (const auto& queue : m_commandQueues)
{
- std::lock_guard<std::mutex> lockQueue{ m_queueLock };
- item->SetState(OrchestratorQueueItemState::Queued);
+ if (queue.second->RemoveItemInState(item, state, true))
+ {
+ return;
+ }
}
}
- void ContextOrchestrator::RequeueItem(OrchestratorQueueItem& item)
+ void ContextOrchestrator::CancelQueueItem(const OrchestratorQueueItem& item)
{
- std::lock_guard<std::mutex> lockQueue{ m_queueLock };
+ // Always cancel the item, even if it isn't running yet, to get the terminationHR set correctly.
+ item.GetContext().Cancel(false, true);
- item.SetState(OrchestratorQueueItemState::Queued);
+ RemoveItemInState(item, OrchestratorQueueItemState::Queued);
}
- void ContextOrchestrator::EnqueueAndRunItem(std::shared_ptr<OrchestratorQueueItem> item)
+ std::shared_ptr<OrchestratorQueueItem> ContextOrchestrator::GetQueueItem(const OrchestratorQueueItemId& queueItemId)
{
- EnqueueItem(item);
+ std::lock_guard<std::mutex> lock{ m_queueLock };
- std::thread runnerThread(&ContextOrchestrator::RunItems, this);
- runnerThread.detach();
+ return FindById(queueItemId);
}
- std::shared_ptr<OrchestratorQueueItem> ContextOrchestrator::GetNextItem()
+ void ContextOrchestrator::AddItemManifestToInstallingSource(const OrchestratorQueueItem& queueItem)
{
- std::lock_guard<std::mutex> lockQueue{ m_queueLock };
+ const auto& manifest = queueItem.GetContext().Get<Execution::Data::Manifest>();
+ m_installingWriteableSource.AddPackageVersion(manifest, std::filesystem::path{ manifest.Id + '.' + manifest.Version });
+ }
+
+ void ContextOrchestrator::RemoveItemManifestFromInstallingSource(const OrchestratorQueueItem& queueItem)
+ {
+ const auto& manifest = queueItem.GetContext().Get<Execution::Data::Manifest>();
+ m_installingWriteableSource.RemovePackageVersion(manifest, std::filesystem::path{ manifest.Id + '.' + manifest.Version });
+ }
- if (m_queueItems.empty())
+ _Requires_lock_held_(m_queueLock)
+ std::deque<std::shared_ptr<OrchestratorQueueItem>>::iterator OrchestratorQueue::FindIteratorById(const OrchestratorQueueItemId& comparisonQueueItemId)
+ {
+ return std::find_if(m_queueItems.begin(), m_queueItems.end(), [&comparisonQueueItemId](const std::shared_ptr<OrchestratorQueueItem>& item) {return (item->GetId().IsSame(comparisonQueueItemId)); });
+ }
+
+ _Requires_lock_held_(m_queueLock)
+ std::shared_ptr<OrchestratorQueueItem> OrchestratorQueue::FindById(const OrchestratorQueueItemId& comparisonQueueItemId)
+ {
+ auto itr = FindIteratorById(comparisonQueueItemId);
+ if (itr != m_queueItems.end())
{
- return {};
+ return *itr;
}
- std::shared_ptr<OrchestratorQueueItem> item = m_queueItems.front();
+ return {};
+ }
+
+ void OrchestratorQueue::EnqueueItem(std::shared_ptr<OrchestratorQueueItem> item)
+ {
+ {
+ std::lock_guard<std::mutex> lockQueue{ m_queueLock };
+ m_queueItems.push_back(item);
+ }
- // Check if item can be dequeued.
- // Since only one item can be installed at a time currently the logic is very simple,
- // and can just check if the first item is ready to run. This logic will need to become
- // more complicated if multiple operation types (e.g. Download & Install) are added that can
- // run simultaneously.
- if (item->GetState() != OrchestratorQueueItemState::Queued)
+ // Add the package to the Installing source so that it can be queried using the Source interface.
+ // Only do this the first time the item is queued.
+ if (item->IsOnFirstCommand())
{
- return {};
+ ContextOrchestrator::Instance().AddItemManifestToInstallingSource(*item);
}
- // Running state must be set inside the queueLock so that multiple threads don't try to run the same item.
- item->SetState(OrchestratorQueueItemState::Running);
- return item;
+ {
+ std::lock_guard<std::mutex> lockQueue{ m_queueLock };
+ item->SetState(OrchestratorQueueItemState::Queued);
+ }
+ }
+
+ OrchestratorQueue::OrchestratorQueue(std::string_view commandName, UINT32 allowedThreads) :
+ m_commandName(commandName), m_allowedThreads(allowedThreads)
+ {
+ m_threadPool.reset(CreateThreadpool(nullptr));
+ THROW_LAST_ERROR_IF_NULL(m_threadPool);
+ m_threadPoolCleanupGroup.reset(CreateThreadpoolCleanupGroup());
+ THROW_LAST_ERROR_IF_NULL(m_threadPoolCleanupGroup);
+ InitializeThreadpoolEnvironment(&m_threadPoolCallbackEnviron);
+ SetThreadpoolCallbackPool(&m_threadPoolCallbackEnviron, m_threadPool.get());
+ SetThreadpoolCallbackCleanupGroup(&m_threadPoolCallbackEnviron, m_threadPoolCleanupGroup.get(), nullptr);
+
+ THROW_LAST_ERROR_IF(!SetThreadpoolThreadMinimum(m_threadPool.get(), 1));
+ SetThreadpoolThreadMaximum(m_threadPool.get(), m_allowedThreads);
+ }
+
+ OrchestratorQueue::~OrchestratorQueue()
+ {
+ CloseThreadpoolCleanupGroupMembers(m_threadPoolCleanupGroup.get(), false, nullptr);
+ }
+
+ void OrchestratorQueue::EnqueueAndRunItem(std::shared_ptr<OrchestratorQueueItem> item)
+ {
+ EnqueueItem(item);
+
+ item->SetCurrentQueue(this);
+ auto work = CreateThreadpoolWork(OrchestratorQueueWorkCallback, item.get(), &m_threadPoolCallbackEnviron);
+ SubmitThreadpoolWork(work);
}
- void ContextOrchestrator::RunItems()
+ void OrchestratorQueue::RunItem(const OrchestratorQueueItemId& itemId)
{
- std::shared_ptr<OrchestratorQueueItem> item = GetNextItem();
- while(item != nullptr)
+ try
{
+ std::shared_ptr<OrchestratorQueueItem> item;
+ bool isCancelled = false;
+
+ // Try to find the item in the queue.
+ {
+ std::lock_guard<std::mutex> lockQueue{ m_queueLock };
+ item = FindById(itemId);
+
+ if (!item)
+ {
+ // Item should be in the queue; this shouldn't happen.
+ return;
+ }
+
+ // Only run if the item is queued and not cancelled.
+ if (item->GetState() == OrchestratorQueueItemState::Queued)
+ {
+ // Mark it as running so that it cannot be cancelled by other threads.
+ item->SetState(OrchestratorQueueItemState::Running);
+ }
+ else if (item->GetState() == OrchestratorQueueItemState::Cancelled)
+ {
+ isCancelled = true;
+ }
+ }
+
+ if (isCancelled)
+ {
+ // Do this separate from above block as the Remove function needs to manage the lock.
+ RemoveItemInState(*item, OrchestratorQueueItemState::Cancelled, true);
+ }
+
+ // Get the item's command and execute it.
HRESULT terminationHR = S_OK;
try
{
@@ -133,21 +253,24 @@ namespace AppInstaller::CLI::Execution
if (FAILED(terminationHR) || item->IsComplete())
{
- RemoveItemInState(*item, OrchestratorQueueItemState::Running);
+ RemoveItemInState(*item, OrchestratorQueueItemState::Running, true);
}
else
{
- RequeueItem(*item);
+ // Remove item from this queue and add it to the queue for the next command.
+ RemoveItemInState(*item, OrchestratorQueueItemState::Running, false);
+ ContextOrchestrator::Instance().EnqueueAndRunItem(item);
}
-
- item = GetNextItem();
+ }
+ catch (...)
+ {
}
}
- void ContextOrchestrator::RemoveItemInState(const OrchestratorQueueItem& item, OrchestratorQueueItemState state)
+ bool OrchestratorQueue::RemoveItemInState(const OrchestratorQueueItem& item, OrchestratorQueueItemState state, bool isGlobalRemove)
{
// OrchestratorQueueItemState::Running items should only be removed by the thread that ran the item.
- // Queued items can be removed by any thread.
+ // Queued items can be removed by any thread.
// NotQueued items should not be removed since, if found in the queue, they are in the process of being queued by another thread.
bool foundItem = false;
@@ -159,32 +282,29 @@ namespace AppInstaller::CLI::Execution
if (itr != m_queueItems.end() && (*itr)->GetState() == state)
{
foundItem = true;
- m_queueItems.erase(itr);
+
+ // The item must only be removed from the queue by the thread that runs
+ // it, because the callback uses it. If any other thread tries to remove
+ // it, we simply mark it as cancelled.
+ if (state == OrchestratorQueueItemState::Running || state == OrchestratorQueueItemState::Cancelled)
+ {
+ (*itr)->SetCurrentQueue(nullptr);
+ m_queueItems.erase(itr);
+ }
+ else if (state == OrchestratorQueueItemState::Queued)
+ {
+ (*itr)->SetState(OrchestratorQueueItemState::Cancelled);
+ }
}
}
- if (foundItem)
+ if (foundItem && isGlobalRemove)
{
- const auto& manifest = item.GetContext().Get<Execution::Data::Manifest>();
- m_installingWriteableSource.RemovePackageVersion(manifest, std::filesystem::path{ manifest.Id + '.' + manifest.Version });
-
+ ContextOrchestrator::Instance().RemoveItemManifestFromInstallingSource(item);
item.GetCompletedEvent().SetEvent();
}
- }
- void ContextOrchestrator::CancelQueueItem(const OrchestratorQueueItem& item)
- {
- // Always cancel the item, even if it isn't running yet, to get the terminationHR set correctly.
- item.GetContext().Cancel(false, true);
-
- RemoveItemInState(item, OrchestratorQueueItemState::Queued);
- }
-
- std::shared_ptr<OrchestratorQueueItem> ContextOrchestrator::GetQueueItem(const OrchestratorQueueItemId& queueItemId)
- {
- std::lock_guard<std::mutex> lock{ m_queueLock };
-
- return FindById(queueItemId);
+ return foundItem;
}
bool OrchestratorQueueItemId::IsSame(const OrchestratorQueueItemId& comparedId) const
@@ -200,5 +320,4 @@ namespace AppInstaller::CLI::Execution
item->AddCommand(std::make_unique<::AppInstaller::CLI::COMInstallCommand>(RootCommand::CommandName));
return item;
}
-
}
diff --git a/src/AppInstallerCLICore/ContextOrchestrator.h b/src/AppInstallerCLICore/ContextOrchestrator.h
@@ -16,9 +16,14 @@ namespace AppInstaller::CLI::Execution
{
enum class OrchestratorQueueItemState
{
+ // Created but not yet queued
NotQueued,
+ // Queued and waiting to be run
Queued,
- Running
+ // Running in the thread pool
+ Running,
+ // Cancelled before it was run; will be deleted when we try to run it
+ Cancelled
};
struct OrchestratorQueueItemId
@@ -33,29 +38,43 @@ namespace AppInstaller::CLI::Execution
std::wstring m_sourceId;
};
+ struct OrchestratorQueue;
+
struct OrchestratorQueueItem
{
OrchestratorQueueItem(OrchestratorQueueItemId id, std::unique_ptr<COMContext> context) : m_id(std::move(id)), m_context(std::move(context)) {}
OrchestratorQueueItemState GetState() const { return m_state; }
void SetState(OrchestratorQueueItemState state) { m_state = state; }
+
+ OrchestratorQueue* GetCurrentQueue() const { return m_currentQueue; }
+ void SetCurrentQueue(OrchestratorQueue* currentQueue) { m_currentQueue = currentQueue; }
+
COMContext& GetContext() const { return *m_context; }
const wil::unique_event& GetCompletedEvent() const { return m_completedEvent; }
const OrchestratorQueueItemId& GetId() const { return m_id; }
+
void AddCommand(std::unique_ptr<Command> command) { m_commands.push_back(std::move(command)); }
+ const Command& GetNextCommand() const { return *m_commands.front(); }
std::unique_ptr<Command> PopNextCommand()
{
+ m_isOnFirstCommand = false;
std::unique_ptr<Command> command = std::move(m_commands.front());
m_commands.pop_front();
return command;
}
+
+ bool IsOnFirstCommand() const { return m_isOnFirstCommand; }
bool IsComplete() const { return m_commands.empty(); }
+
private:
OrchestratorQueueItemState m_state = OrchestratorQueueItemState::NotQueued;
std::unique_ptr<COMContext> m_context;
wil::unique_event m_completedEvent{ wil::EventOptions::ManualReset };
OrchestratorQueueItemId m_id;
std::deque<std::unique_ptr<Command>> m_commands;
+ bool m_isOnFirstCommand = true;
+ OrchestratorQueue* m_currentQueue = nullptr;
};
struct OrchestratorQueueItemFactory
@@ -73,20 +92,68 @@ namespace AppInstaller::CLI::Execution
std::shared_ptr<OrchestratorQueueItem> GetQueueItem(const OrchestratorQueueItemId& queueItemId);
+ void AddItemManifestToInstallingSource(const OrchestratorQueueItem& queueItem);
+ void RemoveItemManifestFromInstallingSource(const OrchestratorQueueItem& queueItem);
+
private:
std::mutex m_queueLock;
- void RunItems();
- std::shared_ptr<OrchestratorQueueItem> GetNextItem();
- void EnqueueItem(std::shared_ptr<OrchestratorQueueItem> item);
- void RequeueItem(OrchestratorQueueItem& item);
+ void AddCommandQueue(std::string_view commandName, UINT32 allowedThreads);
void RemoveItemInState(const OrchestratorQueueItem& item, OrchestratorQueueItemState state);
_Requires_lock_held_(m_queueLock)
- std::deque<std::shared_ptr<OrchestratorQueueItem>>::iterator FindIteratorById(const OrchestratorQueueItemId& queueItemId);
- _Requires_lock_held_(m_queueLock)
std::shared_ptr<OrchestratorQueueItem> FindById(const OrchestratorQueueItemId& queueItemId);
Repository::Source m_installingWriteableSource;
+ std::map<std::string, std::unique_ptr<OrchestratorQueue>> m_commandQueues;
+ };
+
+ // One of the queues used by the orchestrator.
+ // All items in the queue execute the same command.
+ // The queue allows multiple items to run at the same time, up to a limit.
+ struct OrchestratorQueue
+ {
+ OrchestratorQueue(std::string_view commandName, UINT32 allowedThreads);
+ ~OrchestratorQueue();
+
+ // Name of the command this queue can execute
+ std::string_view CommandName() const { return m_commandName; }
+
+ // Enqueues an item to be run when there are threads available.
+ void EnqueueAndRunItem(std::shared_ptr<OrchestratorQueueItem> item);
+
+ // Removes an item by id, provided that it is in the given state.
+ // Returns true if an item was removed.
+ // The item can be removed globally from the orchestrator, or from just this queue.
+ bool RemoveItemInState(const OrchestratorQueueItem& item, OrchestratorQueueItemState state, bool isGlobalRemove);
+
+ // Finds an item by id, if it is in the queue.
+ _Requires_lock_held_(m_queueLock)
+ std::shared_ptr<OrchestratorQueueItem> FindById(const OrchestratorQueueItemId& queueItemId);
+
+ // Runs a single item from the queue.
+ void RunItem(const OrchestratorQueueItemId& itemId);
+
+ private:
+ // Enqueues an item.
+ void EnqueueItem(std::shared_ptr<OrchestratorQueueItem> item);
+
+ _Requires_lock_held_(m_queueLock)
+ std::deque<std::shared_ptr<OrchestratorQueueItem>>::iterator FindIteratorById(const OrchestratorQueueItemId& comparisonQueueItemId);
+
+ std::string_view m_commandName;
+
+ // Number of threads allowed to run items in this queue.
+ const UINT32 m_allowedThreads;
+
+ // Thread pool for this queue, and associated objects.
+ // All work items will be added to the callback environment, and the cleanup group
+ // will manage their closing.
+ // See https://docs.microsoft.com/windows/win32/procthread/using-the-thread-pool-functions
+ TP_CALLBACK_ENVIRON m_threadPoolCallbackEnviron;
+ wil::unique_any<PTP_POOL, decltype(CloseThreadpool), CloseThreadpool> m_threadPool;
+ wil::unique_any<PTP_CLEANUP_GROUP, decltype(CloseThreadpoolCleanupGroup), CloseThreadpoolCleanupGroup> m_threadPoolCleanupGroup;
+
+ std::mutex m_queueLock;
std::deque<std::shared_ptr<OrchestratorQueueItem>> m_queueItems;
};
}
diff --git a/src/AppInstallerCLICore/Workflows/WorkflowBase.cpp b/src/AppInstallerCLICore/Workflows/WorkflowBase.cpp
@@ -1039,7 +1039,7 @@ namespace AppInstaller::CLI::Workflow
// If we cannot find a package using PackageFamilyName or ProductId, try manifest Id and Name pair
SearchRequest searchRequest;
searchRequest.Inclusions.emplace_back(PackageMatchFilter(PackageMatchField::Id, MatchType::CaseInsensitive, manifest.Id));
- // In case there're same Ids from different sources, filter the result using package name
+ // In case there are same Ids from different sources, filter the result using package name
searchRequest.Filters.emplace_back(PackageMatchFilter(PackageMatchField::Name, MatchType::CaseInsensitive, manifest.DefaultLocalization.Get<Manifest::Localization::PackageName>()));
context.Add<Execution::Data::SearchResult>(source.Search(searchRequest));
diff --git a/src/AppInstallerRepositoryCore/Microsoft/SQLiteIndex.cpp b/src/AppInstallerRepositoryCore/Microsoft/SQLiteIndex.cpp
@@ -151,6 +151,7 @@ namespace AppInstaller::Repository::Microsoft
SQLiteIndex::IdType SQLiteIndex::AddManifestInternal(const Manifest::Manifest& manifest, const std::optional<std::filesystem::path>& relativePath)
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
AICLI_LOG(Repo, Verbose, << "Adding manifest for [" << manifest.Id << ", " << manifest.Version << "] at relative path [" << relativePath.value_or("") << "]");
SQLite::Savepoint savepoint = SQLite::Savepoint::Create(m_dbconn, "sqliteindex_addmanifest");
@@ -184,6 +185,7 @@ namespace AppInstaller::Repository::Microsoft
bool SQLiteIndex::UpdateManifestInternal(const Manifest::Manifest& manifest, const std::optional<std::filesystem::path>& relativePath)
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
AICLI_LOG(Repo, Verbose, << "Updating manifest for [" << manifest.Id << ", " << manifest.Version << "] at relative path [" << relativePath.value_or("") << "]");
SQLite::Savepoint savepoint = SQLite::Savepoint::Create(m_dbconn, "sqliteindex_updatemanifest");
@@ -216,6 +218,7 @@ namespace AppInstaller::Repository::Microsoft
void SQLiteIndex::RemoveManifest(const Manifest::Manifest& manifest)
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
SQLite::Savepoint savepoint = SQLite::Savepoint::Create(m_dbconn, "sqliteindex_removemanifest");
m_interface->RemoveManifest(m_dbconn, manifest);
@@ -238,6 +241,7 @@ namespace AppInstaller::Repository::Microsoft
void SQLiteIndex::PrepareForPackaging()
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
AICLI_LOG(Repo, Info, << "Preparing index for packaging");
m_interface->PrepareForPackaging(m_dbconn);
@@ -245,6 +249,7 @@ namespace AppInstaller::Repository::Microsoft
bool SQLiteIndex::CheckConsistency(bool log) const
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
AICLI_LOG(Repo, Info, << "Checking index consistency...");
bool result = m_interface->CheckConsistency(m_dbconn, log);
@@ -256,6 +261,7 @@ namespace AppInstaller::Repository::Microsoft
Schema::ISQLiteIndex::SearchResult SQLiteIndex::Search(const SearchRequest& request) const
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
AICLI_LOG(Repo, Verbose, << "Performing search: " << request.ToString());
return m_interface->Search(m_dbconn, request);
@@ -263,16 +269,19 @@ namespace AppInstaller::Repository::Microsoft
std::optional<std::string> SQLiteIndex::GetPropertyByManifestId(IdType manifestId, PackageVersionProperty property) const
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
return m_interface->GetPropertyByManifestId(m_dbconn, manifestId, property);
}
std::vector<std::string> SQLiteIndex::GetMultiPropertyByManifestId(IdType manifestId, PackageVersionMultiProperty property) const
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
return m_interface->GetMultiPropertyByManifestId(m_dbconn, manifestId, property);
}
std::optional<SQLiteIndex::IdType> SQLiteIndex::GetManifestIdByKey(IdType id, std::string_view version, std::string_view channel) const
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
return m_interface->GetManifestIdByKey(m_dbconn, id, version, channel);
}
@@ -283,21 +292,25 @@ namespace AppInstaller::Repository::Microsoft
std::vector<Utility::VersionAndChannel> SQLiteIndex::GetVersionKeysById(IdType id) const
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
return m_interface->GetVersionKeysById(m_dbconn, id);
}
SQLiteIndex::MetadataResult SQLiteIndex::GetMetadataByManifestId(SQLite::rowid_t manifestId) const
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
return m_interface->GetMetadataByManifestId(m_dbconn, manifestId);
}
void SQLiteIndex::SetMetadataByManifestId(IdType manifestId, PackageVersionMetadata metadata, std::string_view value)
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
m_interface->SetMetadataByManifestId(m_dbconn, manifestId, metadata, value);
}
Utility::NormalizedName SQLiteIndex::NormalizeName(std::string_view name, std::string_view publisher) const
{
+ std::lock_guard<std::mutex> lockInterface{ *m_interfaceLock };
return m_interface->NormalizeName(name, publisher);
}
diff --git a/src/AppInstallerRepositoryCore/Microsoft/SQLiteIndex.h b/src/AppInstallerRepositoryCore/Microsoft/SQLiteIndex.h
@@ -163,5 +163,6 @@ namespace AppInstaller::Repository::Microsoft
SQLite::Connection m_dbconn;
Schema::Version m_version;
std::unique_ptr<Schema::ISQLiteIndex> m_interface;
+ std::unique_ptr<std::mutex> m_interfaceLock = std::make_unique<std::mutex>();
};
}
